updated msgio + secio
Juan Batiz-Benet committed
Dec 14, 2014 at 10:39 UTC
3b2965302270aa35c89a3d989bcd4a4e68078f98
7 files changed
+48
-123
Godeps/Godeps.json
+1
-1
@@ -110,7 +110,7 @@
110
},
111
{
112
"ImportPath": "github.com/jbenet/go-msgio",
113
- "Rev": "753e598a1d24b311ee05c4ce001cff74e2a8e745"
113
+ "Rev": "281b085dc602c4f0377438e20331f45a91bcdf9c"
114
},
115
{
116
"ImportPath": "github.com/jbenet/go-multiaddr",
Godeps/_workspace/src/github.com/jbenet/go-msgio/msgio.go
+12
-2
@@ -45,6 +45,10 @@ type Reader interface {
45
46
// ReleaseMsg signals a buffer can be reused.
47
ReleaseMsg([]byte)
48
+
49
+ // NextMsgLen returns the length of the next (peeked) message. Does
50
+ // not destroy the message or have other adverse effects
51
+ NextMsgLen() (int, error)
52
}
53
54
// ReadCloser combines a Reader and Closer.
@@ -142,9 +146,15 @@ func NewReaderWithPool(r io.Reader, p *mpool.Pool) ReadCloser {
146
}
147
}
148
145
-// nextMsgLen reads the length of the next msg into s.lbuf, and returns it.
146
-// WARNING: like ReadMsg, nextMsgLen is destructive. It reads from the internal
149
+// NextMsgLen reads the length of the next msg into s.lbuf, and returns it.
150
+// WARNING: like Read, NextMsgLen is destructive. It reads from the internal
151
// reader.
152
+func (s *reader) NextMsgLen() (int, error) {
153
+ s.lock.Lock()
154
+ defer s.lock.Unlock()
155
+ return s.nextMsgLen()
156
+}
157
+
158
func (s *reader) nextMsgLen() (int, error) {
159
if s.next == -1 {
160
if _, err := io.ReadFull(s.R, s.lbuf); err != nil {
Godeps/_workspace/src/github.com/jbenet/go-msgio/multipool/pool.go
deleted
-113
@@ -1,113 +0,0 @@
1
-// Package mpool provides a sync.Pool equivalent that buckets incoming
2
-// requests to one of 32 sub-pools, one for each power of 2, 0-32.
3
-//
4
-// import "github.com/jbenet/go-msgio/mpool"
5
-// var p mpool.Pool
6
-//
7
-// small := make([]byte, 1024)
8
-// large := make([]byte, 4194304)
9
-// p.Put(1024, small)
10
-// p.Put(4194304, large)
11
-//
12
-// small2 := p.Get(1024).([]byte)
13
-// large2 := p.Get(4194304).([]byte)
14
-// fmt.Println("small2 len:", len(small2))
15
-// fmt.Println("large2 len:", len(large2))
16
-//
17
-// // Output:
18
-// // small2 len: 1024
19
-// // large2 len: 4194304
20
-//
21
-package mpool
22
-
23
-import (
24
- "fmt"
25
- "sync"
26
-)
27
-
28
-// ByteSlicePool is a static Pool for reusing byteslices of various sizes.
29
-var ByteSlicePool Pool
30
-
31
-func init() {
32
- ByteSlicePool.New = func(length int) interface{} {
33
- return make([]byte, length)
34
- }
35
-}
36
-
37
-// MaxLength is the maximum length of an element that can be added to the Pool.
38
-const MaxLength = (1 << 32) - 1
39
-
40
-// Pool is a pool to handle cases of reusing elements of varying sizes.
41
-// It maintains up to 32 internal pools, for each power of 2 in 0-32.
42
-type Pool struct {
43
- small int // the size of the first pool
44
- pools [32]*sync.Pool // a list of singlePools
45
- sync.Mutex // protecting list
46
-
47
- // New is a function that constructs a new element in the pool, with given len
48
- New func(len int) interface{}
49
-}
50
-
51
-func (p *Pool) getPool(idx uint32) *sync.Pool {
52
- if idx > uint32(len(p.pools)) {
53
- panic(fmt.Errorf("index too large: %d", idx))
54
- }
55
-
56
- p.Lock()
57
- defer p.Unlock()
58
-
59
- sp := p.pools[idx]
60
- if sp == nil {
61
- sp = new(sync.Pool)
62
- p.pools[idx] = sp
63
- }
64
- return sp
65
-}
66
-
67
-// Get selects an arbitrary item from the Pool, removes it from the Pool,
68
-// and returns it to the caller. Get may choose to ignore the pool and
69
-// treat it as empty. Callers should not assume any relation between values
70
-// passed to Put and the values returned by Get.
71
-//
72
-// If Get would otherwise return nil and p.New is non-nil, Get returns the
73
-// result of calling p.New.
74
-func (p *Pool) Get(length uint32) interface{} {
75
- idx := largerPowerOfTwo(length)
76
- sp := p.getPool(idx)
77
- val := sp.Get()
78
- if val == nil && p.New != nil {
79
- val = p.New(0x1 << idx)
80
- }
81
- return val
82
-}
83
-
84
-// Put adds x to the pool.
85
-func (p *Pool) Put(length uint32, val interface{}) {
86
- if length > MaxLength {
87
- length = MaxLength
88
- }
89
-
90
- idx := smallerPowerOfTwo(length)
91
- sp := p.getPool(idx)
92
- sp.Put(val)
93
-}
94
-
95
-func largerPowerOfTwo(num uint32) uint32 {
96
- for p := uint32(0); p < 32; p++ {
97
- if (0x1 << p) >= num {
98
- return p
99
- }
100
- }
101
-
102
- panic("unreachable")
103
-}
104
-
105
-func smallerPowerOfTwo(num uint32) uint32 {
106
- for p := uint32(1); p < 32; p++ {
107
- if (0x1 << p) > num {
108
- return p - 1
109
- }
110
- }
111
-
112
- panic("unreachable")
113
-}
crypto/secio/rw.go
+19
-7
@@ -87,22 +87,34 @@ func NewETMReader(r io.Reader, s cipher.Stream, mac HMAC) msgio.ReadCloser {
87
return &etmReader{msg: msgio.NewReader(r), str: s, mac: mac}
88
}
89
90
+func (r *etmReader) NextMsgLen() (int, error) {
91
+ return r.msg.NextMsgLen()
92
+}
93
+
94
func (r *etmReader) Read(buf []byte) (int, error) {
95
+ // first, check the buffer has enough space.
96
+ fullLen, err := r.msg.NextMsgLen()
97
+ if err != nil {
98
+ return 0, err
99
+ }
100
+
101
+ dataLen := fullLen - r.mac.size
102
+ if cap(buf) < dataLen {
103
+ return 0, io.ErrShortBuffer
104
+ }
105
+
106
buf2 := buf
107
changed := false
93
- if cap(buf2) < (len(buf) + r.mac.size) {
94
- buf2 = make([]byte, len(buf)+r.mac.size)
108
+ if cap(buf) < fullLen {
109
+ buf2 = make([]byte, fullLen)
110
changed = true
111
}
112
+ buf2 = buf2[:fullLen]
113
98
- // WARNING: assumes msg.Read will only read _one_ message. this is what
99
- // msgio is supposed to do. but msgio may change in the future. may this
100
- // comment be your guiding light.
101
- n, err := r.msg.Read(buf2)
114
+ n, err := io.ReadFull(r.msg, buf2)
115
if err != nil {
116
return n, err
117
}
105
- buf2 = buf2[:n]
118
119
m, err := r.macCheckThenDecrypt(buf2)
120
if err != nil {
net/conn/conn.go
+4
@@ -115,6 +115,10 @@ func (c *singleConn) Write(buf []byte) (int, error) {
115
return c.msgrw.Write(buf)
116
}
117
118
+func (c *singleConn) NextMsgLen() (int, error) {
119
+ return c.msgrw.NextMsgLen()
120
+}
121
+
122
// ReadMsg reads data, net.Conn style
123
func (c *singleConn) ReadMsg() ([]byte, error) {
124
return c.msgrw.ReadMsg()
net/conn/multiconn.go
+8
@@ -266,6 +266,14 @@ func (c *MultiConn) Write(buf []byte) (int, error) {
266
return bc.Write(buf)
267
}
268
269
+func (c *MultiConn) NextMsgLen() (int, error) {
270
+ bc := c.BestConn()
271
+ if bc == nil {
272
+ return 0, errors.New("no best connection")
273
+ }
274
+ return bc.NextMsgLen()
275
+}
276
+
277
// ReadMsg reads data, net.Conn style
278
func (c *MultiConn) ReadMsg() ([]byte, error) {
279
next, ok := <-c.fanIn
net/conn/secure_conn.go
+4
@@ -93,6 +93,10 @@ func (c *secureConn) Write(buf []byte) (int, error) {
93
return c.secure.Write(buf)
94
}
95
96
+func (c *secureConn) NextMsgLen() (int, error) {
97
+ return c.secure.NextMsgLen()
98
+}
99
+
100
// ReadMsg reads data, net.Conn style
101
func (c *secureConn) ReadMsg() ([]byte, error) {
102
return c.secure.ReadMsg()