@cryptotaxi247 / kubo / commits / 97e0f73e7

msgio: updated to remove max size

Juan Batiz-Benet committed Dec 6, 2014 at 11:23 UTC 97e0f73e79a34030669a43b9111eb70cd7a13255
9 files changed +745 -80
Godeps/Godeps.json
+1 -1
@@ -107,7 +107,7 @@
107 },
108 {
109 "ImportPath": "github.com/jbenet/go-msgio",
110 - "Rev": "ab0e7a0e111d7c7d814ad238bcbf3934efb76ac3"
110 + "Rev": "8361f0f6c783f09419ec3d8dd75fe950744f5e91"
111 },
112 {
113 "ImportPath": "github.com/jbenet/go-multiaddr",
Godeps/_workspace/src/github.com/jbenet/go-msgio/chan.go
+33 -28
@@ -2,17 +2,19 @@ package msgio
2
3 import (
4 "io"
5 - "sync"
5 +
6 + mpool "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio/mpool"
7 )
8
9 +// Chan is a msgio duplex channel. It is used to have a channel interface
10 +// around a msgio.Reader or Writer.
11 type Chan struct {
9 - Buffers [][]byte
12 MsgChan chan []byte
13 ErrChan chan error
14 CloseChan chan bool
13 - BufPool *sync.Pool
15 }
16
17 +// NewChan constructs a Chan with a given buffer size.
18 func NewChan(chanSize int) *Chan {
19 return &Chan{
20 MsgChan: make(chan []byte, chanSize),
@@ -21,36 +23,27 @@ func NewChan(chanSize int) *Chan {
23 }
24 }
25
24 -func NewChanWithPool(chanSize int, pool *sync.Pool) *Chan {
25 - return &Chan{
26 - MsgChan: make(chan []byte, chanSize),
27 - ErrChan: make(chan error, 1),
28 - CloseChan: make(chan bool, 2),
29 - BufPool: pool,
30 - }
26 +// ReadFrom wraps the given io.Reader with a msgio.Reader, reads all
27 +// messages, ands sends them down the channel.
28 +func (s *Chan) ReadFrom(r io.Reader) {
29 + s.readFrom(NewReader(r))
30 }
31
33 -func (s *Chan) getBuffer(size int) []byte {
34 - if s.BufPool == nil {
35 - return make([]byte, size)
36 - } else {
37 - bufi := s.BufPool.Get()
38 - buf, ok := bufi.([]byte)
39 - if !ok {
40 - panic("Got invalid type from sync pool!")
41 - }
42 - return buf
43 - }
32 +// ReadFromWithPool wraps the given io.Reader with a msgio.Reader, reads all
33 +// messages, ands sends them down the channel. Uses given Pool
34 +func (s *Chan) ReadFromWithPool(r io.Reader, p *mpool.Pool) {
35 + s.readFrom(NewReaderWithPool(r, p))
36 }
37
46 -func (s *Chan) ReadFrom(r io.Reader, maxMsgLen int) {
47 - // new buffer per message
48 - // if bottleneck, cycle around a set of buffers
49 - mr := NewReader(r)
38 +// ReadFrom wraps the given io.Reader with a msgio.Reader, reads all
39 +// messages, ands sends them down the channel.
40 +func (s *Chan) readFrom(mr Reader) {
41 + // single reader, no need for Mutex
42 + mr.(*reader).lock = new(nullLocker)
43 +
44 Loop:
45 for {
52 - buf := s.getBuffer(maxMsgLen)
53 - l, err := mr.ReadMsg(buf)
46 + buf, err := mr.ReadMsg()
47 if err != nil {
48 if err == io.EOF {
49 break Loop // done
@@ -64,7 +57,7 @@ Loop:
57 select {
58 case <-s.CloseChan:
59 break Loop // told we're done
67 - case s.MsgChan <- buf[:l]:
60 + case s.MsgChan <- buf:
61 // ok seems fine. send it away
62 }
63 }
@@ -74,10 +67,15 @@ Loop:
67 s.CloseChan <- true
68 }
69
70 +// WriteTo wraps the given io.Writer with a msgio.Writer, listens on the
71 +// channel and writes all messages to the writer.
72 func (s *Chan) WriteTo(w io.Writer) {
73 // new buffer per message
74 // if bottleneck, cycle around a set of buffers
75 mw := NewWriter(w)
76 +
77 + // single writer, no need for Mutex
78 + mw.(*writer).lock = new(nullLocker)
79 Loop:
80 for {
81 select {
@@ -104,6 +102,13 @@ Loop:
102 s.CloseChan <- true
103 }
104
105 +// Close the Chan
106 func (s *Chan) Close() {
107 s.CloseChan <- true
108 }
109 +
110 +// nullLocker conforms to the sync.Locker interface but does nothing.
111 +type nullLocker struct{}
112 +
113 +func (l *nullLocker) Lock() {}
114 +func (l *nullLocker) Unlock() {}
Godeps/_workspace/src/github.com/jbenet/go-msgio/chan_test.go
+4 -8
@@ -5,7 +5,6 @@ import (
5 randbuf "github.com/jbenet/go-randbuf"
6 "io"
7 "math/rand"
8 - "sync"
8 "testing"
9 "time"
10 )
@@ -13,8 +12,7 @@ import (
12 func TestReadChan(t *testing.T) {
13 buf := bytes.NewBuffer(nil)
14 writer := NewWriter(buf)
16 - p := &sync.Pool{New: func() interface{} { return make([]byte, 1000) }}
17 - rchan := NewChan(10, p)
15 + rchan := NewChan(10)
16 msgs := [1000][]byte{}
17
18 r := rand.New(rand.NewSource(time.Now().UnixNano()))
@@ -30,7 +28,7 @@ func TestReadChan(t *testing.T) {
28 t.Fatal(err)
29 }
30
33 - go rchan.ReadFrom(buf, 1000)
31 + go rchan.ReadFrom(buf)
32 defer rchan.Close()
33
34 Loop:
@@ -60,7 +58,7 @@ Loop:
58 func TestWriteChan(t *testing.T) {
59 buf := bytes.NewBuffer(nil)
60 reader := NewReader(buf)
63 - wchan := NewChan(10, nil)
61 + wchan := NewChan(10)
62 msgs := [1000][]byte{}
63
64 go wchan.WriteTo(buf)
@@ -87,8 +85,7 @@ func TestWriteChan(t *testing.T) {
85 defer wchan.Close()
86
87 for i := 0; ; i++ {
90 - msg2 := make([]byte, 1000)
91 - n, err := reader.ReadMsg(msg2)
88 + msg2, err := reader.ReadMsg()
89 if err != nil {
90 if err == io.EOF {
91 if i < len(msg2) {
@@ -100,7 +97,6 @@ func TestWriteChan(t *testing.T) {
97 }
98
99 msg1 := msgs[i]
103 - msg2 = msg2[:n]
100 if !bytes.Equal(msg1, msg2) {
101 t.Fatal("message retrieved not equal\n", msg1, "\n\n", msg2)
102 }
Godeps/_workspace/src/github.com/jbenet/go-msgio/mpool/pool.go new
+109
@@ -0,0 +1,109 @@
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
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 + idx := smallerPowerOfTwo(length)
87 + sp := p.getPool(idx)
88 + sp.Put(val)
89 +}
90 +
91 +func largerPowerOfTwo(num uint32) uint32 {
92 + for p := uint32(0); p < 32; p++ {
93 + if (0x1 << p) >= num {
94 + return p
95 + }
96 + }
97 +
98 + panic("unreachable")
99 +}
100 +
101 +func smallerPowerOfTwo(num uint32) uint32 {
102 + for p := uint32(1); p < 32; p++ {
103 + if (0x1 << p) > num {
104 + return p - 1
105 + }
106 + }
107 +
108 + panic("unreachable")
109 +}
Godeps/_workspace/src/github.com/jbenet/go-msgio/mpool/pool_test.go new
+200
@@ -0,0 +1,200 @@
1 +// Copyright 2013 The Go Authors. All rights reserved.
2 +// Use of this source code is governed by a BSD-style
3 +// license that can be found in the LICENSE file.
4 +
5 +// Pool is no-op under race detector, so all these tests do not work.
6 +// +build !race
7 +
8 +package mpool
9 +
10 +import (
11 + "fmt"
12 + "runtime"
13 + "runtime/debug"
14 + "sync/atomic"
15 + "testing"
16 + "time"
17 +)
18 +
19 +func TestPool(t *testing.T) {
20 + // disable GC so we can control when it happens.
21 + defer debug.SetGCPercent(debug.SetGCPercent(-1))
22 + var p Pool
23 + if p.Get(10) != nil {
24 + t.Fatal("expected empty")
25 + }
26 + p.Put(16, "a")
27 + p.Put(2048, "b")
28 + if g := p.Get(16); g != "a" {
29 + t.Fatalf("got %#v; want a", g)
30 + }
31 + if g := p.Get(2048); g != "b" {
32 + t.Fatalf("got %#v; want b", g)
33 + }
34 + if g := p.Get(16); g != nil {
35 + t.Fatalf("got %#v; want nil", g)
36 + }
37 + if g := p.Get(2048); g != nil {
38 + t.Fatalf("got %#v; want nil", g)
39 + }
40 + if g := p.Get(1); g != nil {
41 + t.Fatalf("got %#v; want nil", g)
42 + }
43 + p.Put(1023, "d")
44 + if g := p.Get(1024); g != nil {
45 + t.Fatalf("got %#v; want nil", g)
46 + }
47 + if g := p.Get(512); g != "d" {
48 + t.Fatalf("got %#v; want d", g)
49 + }
50 +
51 + debug.SetGCPercent(100) // to allow following GC to actually run
52 + runtime.GC()
53 + if g := p.Get(10); g != nil {
54 + t.Fatalf("got %#v; want nil after GC", g)
55 + }
56 +}
57 +
58 +func TestPoolNew(t *testing.T) {
59 + // disable GC so we can control when it happens.
60 + defer debug.SetGCPercent(debug.SetGCPercent(-1))
61 +
62 + s := [32]int{}
63 + p := Pool{
64 + New: func(length int) interface{} {
65 + idx := largerPowerOfTwo(uint32(length))
66 + s[idx]++
67 + return s[idx]
68 + },
69 + }
70 + if v := p.Get(1 << 5); v != 1 {
71 + t.Fatalf("got %v; want 1", v)
72 + }
73 + if v := p.Get(1 << 2); v != 1 {
74 + t.Fatalf("got %v; want 1", v)
75 + }
76 + if v := p.Get(1 << 2); v != 2 {
77 + t.Fatalf("got %v; want 2", v)
78 + }
79 + if v := p.Get(1 << 5); v != 2 {
80 + t.Fatalf("got %v; want 2", v)
81 + }
82 + p.Put(1<<2, 42)
83 + p.Put(1<<5, 42)
84 + if v := p.Get(1 << 2); v != 42 {
85 + t.Fatalf("got %v; want 42", v)
86 + }
87 + if v := p.Get(1 << 2); v != 3 {
88 + t.Fatalf("got %v; want 3", v)
89 + }
90 + if v := p.Get(1 << 5); v != 42 {
91 + t.Fatalf("got %v; want 42", v)
92 + }
93 + if v := p.Get(1 << 5); v != 3 {
94 + t.Fatalf("got %v; want 3", v)
95 + }
96 +}
97 +
98 +// Test that Pool does not hold pointers to previously cached
99 +// resources
100 +func TestPoolGC(t *testing.T) {
101 + var p Pool
102 + var fin uint32
103 + const N = 100
104 + for i := 0; i < N; i++ {
105 + v := new(string)
106 + runtime.SetFinalizer(v, func(vv *string) {
107 + atomic.AddUint32(&fin, 1)
108 + })
109 + p.Put(uint32(i), v)
110 + }
111 + for i := 0; i < N; i++ {
112 + p.Get(uint32(i))
113 + }
114 + for i := 0; i < 5; i++ {
115 + runtime.GC()
116 + time.Sleep(time.Duration(i*100+10) * time.Millisecond)
117 + // 1 pointer can remain on stack or elsewhere
118 + if atomic.LoadUint32(&fin) >= N-1 {
119 + return
120 + }
121 + }
122 + t.Fatalf("only %v out of %v resources are finalized",
123 + atomic.LoadUint32(&fin), N)
124 +}
125 +
126 +func TestPoolStress(t *testing.T) {
127 + const P = 10
128 + N := int(1e6)
129 + if testing.Short() {
130 + N /= 100
131 + }
132 + var p Pool
133 + done := make(chan bool)
134 + for i := 0; i < P; i++ {
135 + go func() {
136 + var v interface{} = 0
137 + for j := 0; j < N; j++ {
138 + if v == nil {
139 + v = 0
140 + }
141 + p.Put(uint32(j), v)
142 + v = p.Get(uint32(j))
143 + if v != nil && v.(int) != 0 {
144 + t.Fatalf("expect 0, got %v", v)
145 + }
146 + }
147 + done <- true
148 + }()
149 + }
150 + for i := 0; i < P; i++ {
151 + <-done
152 + }
153 +}
154 +
155 +func BenchmarkPool(b *testing.B) {
156 + var p Pool
157 + b.RunParallel(func(pb *testing.PB) {
158 + i := 0
159 + for pb.Next() {
160 + i = i << 1
161 + p.Put(uint32(i), 1)
162 + p.Get(uint32(i))
163 + }
164 + })
165 +}
166 +
167 +func BenchmarkPoolOverlflow(b *testing.B) {
168 + var p Pool
169 + b.RunParallel(func(pb *testing.PB) {
170 + for pb.Next() {
171 + for pow := uint32(0); pow < 32; pow++ {
172 + for b := 0; b < 100; b++ {
173 + p.Put(uint32(1<<pow), 1)
174 + }
175 + }
176 + for pow := uint32(0); pow < 32; pow++ {
177 + for b := 0; b < 100; b++ {
178 + p.Get(uint32(1 << pow))
179 + }
180 + }
181 + }
182 + })
183 +}
184 +
185 +func ExamplePool() {
186 + var p Pool
187 +
188 + small := make([]byte, 1024)
189 + large := make([]byte, 4194304)
190 + p.Put(uint32(len(small)), small)
191 + p.Put(uint32(len(large)), large)
192 +
193 + small2 := p.Get(uint32(len(small))).([]byte)
194 + large2 := p.Get(uint32(len(large))).([]byte)
195 + fmt.Println("small2 len:", len(small2))
196 + fmt.Println("large2 len:", len(large2))
197 + // Output:
198 + // small2 len: 1024
199 + // large2 len: 4194304
200 +}
Godeps/_workspace/src/github.com/jbenet/go-msgio/msgio.go
+163 -19
@@ -3,48 +3,90 @@ package msgio
3 import (
4 "encoding/binary"
5 "io"
6 + "sync"
7 +
8 + mpool "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio/mpool"
9 )
10
11 +// NBO is NetworkByteOrder
12 var NBO = binary.BigEndian
13
14 +const lengthSize = 4
15 +
16 +// Writer is the msgio Writer interface. It writes len-framed messages.
17 type Writer interface {
18 +
19 + // Write writes passed in buffer as a single message.
20 + Write([]byte) error
21 +
22 + // WriteMsg writes the msg in the passed in buffer.
23 WriteMsg([]byte) error
24 }
25
26 +// WriteCloser is a Writer + Closer interface. Like in `golang/pkg/io`
27 type WriteCloser interface {
28 Writer
29 io.Closer
30 }
31
32 +// Reader is the msgio Reader interface. It reads len-framed messages.
33 type Reader interface {
20 - ReadMsg([]byte) (int, error)
34 +
35 + // Read reads the next message from the Reader.
36 + // The client must pass a buffer large enough, or io.ErrShortBuffer will be
37 + // returned.
38 + Read([]byte) (int, error)
39 +
40 + // ReadMsg reads the next message from the Reader.
41 + // Uses a mpool.Pool internally to reuse buffers. io.ErrShortBuffer will
42 + // be returned if the Pool.Get(...) returns nil.
43 + // User may call ReleaseMsg(msg) to signal a buffer can be reused.
44 + ReadMsg() ([]byte, error)
45 +
46 + // ReleaseMsg signals a buffer can be reused.
47 + ReleaseMsg([]byte)
48 }
49
50 +// ReadCloser combines a Reader and Closer.
51 type ReadCloser interface {
52 Reader
53 io.Closer
54 }
55
56 +// ReadWriter combines a Reader and Writer.
57 type ReadWriter interface {
58 Reader
59 Writer
60 }
61
62 +// ReadWriteCloser combines a Reader, a Writer, and Closer.
63 type ReadWriteCloser interface {
64 Reader
65 Writer
66 io.Closer
67 }
68
39 -type Writer_ struct {
69 +// writer is the underlying type that implements the Writer interface.
70 +type writer struct {
71 W io.Writer
72 +
73 + lock sync.Locker
74 }
75
76 +// NewWriter wraps an io.Writer with a msgio framed writer. The msgio.Writer
77 +// will write the length prefix of every message written.
78 func NewWriter(w io.Writer) WriteCloser {
44 - return &Writer_{w}
79 + return &writer{W: w, lock: new(sync.Mutex)}
80 +}
81 +
82 +func (s *writer) Write(msg []byte) (err error) {
83 + return s.WriteMsg(msg)
84 }
85
47 -func (s *Writer_) WriteMsg(msg []byte) (err error) {
86 +func (s *writer) WriteMsg(msg []byte) (err error) {
87 + s.lock.Lock()
88 + defer s.lock.Unlock()
89 +
90 length := uint32(len(msg))
91 if err := binary.Write(s.W, NBO, &length); err != nil {
92 return err
@@ -53,59 +95,161 @@ func (s *Writer_) WriteMsg(msg []byte) (err error) {
95 return err
96 }
97
56 -func (s *Writer_) Close() error {
98 +func (s *writer) Close() error {
99 + s.lock.Lock()
100 + defer s.lock.Unlock()
101 +
102 if c, ok := s.W.(io.Closer); ok {
103 return c.Close()
104 }
105 return nil
106 }
107
63 -type Reader_ struct {
64 - R io.Reader
108 +// reader is the underlying type that implements the Reader interface.
109 +type reader struct {
110 + R io.Reader
111 +
112 lbuf []byte
113 + next int
114 + pool *mpool.Pool
115 + lock sync.Locker
116 }
117
118 +// NewReader wraps an io.Reader with a msgio framed reader. The msgio.Reader
119 +// will read whole messages at a time (using the length). Assumes an equivalent
120 +// writer on the other side.
121 func NewReader(r io.Reader) ReadCloser {
69 - return &Reader_{r, make([]byte, 4)}
122 + return NewReaderWithPool(r, &mpool.ByteSlicePool)
123 +}
124 +
125 +// NewReaderWithPool wraps an io.Reader with a msgio framed reader. The msgio.Reader
126 +// will read whole messages at a time (using the length). Assumes an equivalent
127 +// writer on the other side. It uses a given mpool.Pool
128 +func NewReaderWithPool(r io.Reader, p *mpool.Pool) ReadCloser {
129 + if p == nil {
130 + panic("nil pool")
131 + }
132 + return &reader{
133 + R: r,
134 + lbuf: make([]byte, lengthSize),
135 + next: -1,
136 + pool: p,
137 + lock: new(sync.Mutex),
138 + }
139 +}
140 +
141 +// nextMsgLen reads the length of the next msg into s.lbuf, and returns it.
142 +// WARNING: like ReadMsg, nextMsgLen is destructive. It reads from the internal
143 +// reader.
144 +func (s *reader) nextMsgLen() (int, error) {
145 + if s.next == -1 {
146 + if _, err := io.ReadFull(s.R, s.lbuf); err != nil {
147 + return 0, err
148 + }
149 + s.next = int(NBO.Uint32(s.lbuf))
150 + }
151 + return s.next, nil
152 }
153
72 -func (s *Reader_) ReadMsg(msg []byte) (int, error) {
73 - if _, err := io.ReadFull(s.R, s.lbuf); err != nil {
154 +func (s *reader) Read(msg []byte) (int, error) {
155 + s.lock.Lock()
156 + defer s.lock.Unlock()
157 +
158 + length, err := s.nextMsgLen()
159 + if err != nil {
160 return 0, err
161 }
76 - length := int(NBO.Uint32(s.lbuf))
77 - if length < 0 || length > len(msg) {
162 +
163 + if length > len(msg) {
164 return 0, io.ErrShortBuffer
165 }
80 - _, err := io.ReadFull(s.R, msg[:length])
166 + _, err = io.ReadFull(s.R, msg[:length])
167 + s.next = -1 // signal we've consumed this msg
168 return length, err
169 }
170
84 -func (s *Reader_) Close() error {
171 +func (s *reader) ReadMsg() ([]byte, error) {
172 + s.lock.Lock()
173 + defer s.lock.Unlock()
174 +
175 + length, err := s.nextMsgLen()
176 + if err != nil {
177 + return nil, err
178 + }
179 +
180 + msgb := s.pool.Get(uint32(length))
181 + if msgb == nil {
182 + return nil, io.ErrShortBuffer
183 + }
184 + msg := msgb.([]byte)[:length]
185 + _, err = io.ReadFull(s.R, msg)
186 + s.next = -1 // signal we've consumed this msg
187 + return msg, err
188 +}
189 +
190 +func (s *reader) ReleaseMsg(msg []byte) {
191 + s.pool.Put(uint32(cap(msg)), msg)
192 +}
193 +
194 +func (s *reader) Close() error {
195 + s.lock.Lock()
196 + defer s.lock.Unlock()
197 +
198 if c, ok := s.R.(io.Closer); ok {
199 return c.Close()
200 }
201 return nil
202 }
203
91 -type ReadWriter_ struct {
204 +// readWriter is the underlying type that implements a ReadWriter.
205 +type readWriter struct {
206 Reader
207 Writer
208 }
209
210 +// NewReadWriter wraps an io.ReadWriter with a msgio.ReadWriter. Writing
211 +// and Reading will be appropriately framed.
212 func NewReadWriter(rw io.ReadWriter) ReadWriter {
97 - return &ReadWriter_{
213 + return &readWriter{
214 Reader: NewReader(rw),
215 Writer: NewWriter(rw),
216 }
217 }
218
103 -func (rw *ReadWriter_) Close() error {
219 +func (rw *readWriter) Close() error {
220 + var errs []error
221 +
222 if w, ok := rw.Writer.(WriteCloser); ok {
105 - return w.Close()
223 + if err := w.Close(); err != nil {
224 + errs = append(errs, err)
225 + }
226 }
227 if r, ok := rw.Reader.(ReadCloser); ok {
108 - return r.Close()
228 + if err := r.Close(); err != nil {
229 + errs = append(errs, err)
230 + }
231 + }
232 +
233 + if len(errs) > 0 {
234 + return multiErr(errs)
235 }
236 return nil
237 }
238 +
239 +// multiErr is a util to return multiple errors
240 +type multiErr []error
241 +
242 +func (m multiErr) Error() string {
243 + if len(m) == 0 {
244 + return "no errors"
245 + }
246 +
247 + s := "Multiple errors: "
248 + for i, e := range m {
249 + if i != 0 {
250 + s += ", "
251 + }
252 + s += e.Error()
253 + }
254 + return s
255 +}
Godeps/_workspace/src/github.com/jbenet/go-msgio/msgio_test.go
+116 -3
@@ -2,14 +2,16 @@ package msgio
2
3 import (
4 "bytes"
5 + "fmt"
6 randbuf "github.com/jbenet/go-randbuf"
7 "io"
8 "math/rand"
9 + "sync"
10 "testing"
11 "time"
12 )
13
12 -func TestReaderWriter(t *testing.T) {
14 +func TestReadWrite(t *testing.T) {
15 buf := bytes.NewBuffer(nil)
16 writer := NewWriter(buf)
17 reader := NewReader(buf)
@@ -18,7 +20,7 @@ func TestReaderWriter(t *testing.T) {
20 r := rand.New(rand.NewSource(time.Now().UnixNano()))
21 for i := range msgs {
22 msgs[i] = randbuf.RandBuf(r, r.Intn(1000))
21 - err := writer.WriteMsg(msgs[i])
23 + err := writer.Write(msgs[i])
24 if err != nil {
25 t.Fatal(err)
26 }
@@ -30,7 +32,7 @@ func TestReaderWriter(t *testing.T) {
32
33 for i := 0; ; i++ {
34 msg2 := make([]byte, 1000)
33 - n, err := reader.ReadMsg(msg2)
35 + n, err := reader.Read(msg2)
36 if err != nil {
37 if err == io.EOF {
38 if i < len(msg2) {
@@ -52,3 +54,114 @@ func TestReaderWriter(t *testing.T) {
54 t.Error(err)
55 }
56 }
57 +
58 +func TestReadWriteMsg(t *testing.T) {
59 + buf := bytes.NewBuffer(nil)
60 + writer := NewWriter(buf)
61 + reader := NewReader(buf)
62 + msgs := [1000][]byte{}
63 +
64 + r := rand.New(rand.NewSource(time.Now().UnixNano()))
65 + for i := range msgs {
66 + msgs[i] = randbuf.RandBuf(r, r.Intn(1000))
67 + err := writer.WriteMsg(msgs[i])
68 + if err != nil {
69 + t.Fatal(err)
70 + }
71 + }
72 +
73 + if err := writer.Close(); err != nil {
74 + t.Fatal(err)
75 + }
76 +
77 + for i := 0; ; i++ {
78 + msg2, err := reader.ReadMsg()
79 + if err != nil {
80 + if err == io.EOF {
81 + if i < len(msg2) {
82 + t.Error("failed to read all messages", len(msgs), i)
83 + }
84 + break
85 + }
86 + t.Error("unexpected error", err)
87 + }
88 +
89 + msg1 := msgs[i]
90 + if !bytes.Equal(msg1, msg2) {
91 + t.Fatal("message retrieved not equal\n", msg1, "\n\n", msg2)
92 + }
93 + }
94 +
95 + if err := reader.Close(); err != nil {
96 + t.Error(err)
97 + }
98 +}
99 +
100 +func TestReadWriteMsgSync(t *testing.T) {
101 + buf := bytes.NewBuffer(nil)
102 + writer := NewWriter(buf)
103 + reader := NewReader(buf)
104 + msgs := [1000][]byte{}
105 +
106 + r := rand.New(rand.NewSource(time.Now().UnixNano()))
107 + for i := range msgs {
108 + msgs[i] = randbuf.RandBuf(r, r.Intn(1000)+4)
109 + NBO.PutUint32(msgs[i][:4], uint32(i))
110 + }
111 +
112 + var wg1 sync.WaitGroup
113 + var wg2 sync.WaitGroup
114 +
115 + errs := make(chan error, 10000)
116 + for i := range msgs {
117 + wg1.Add(1)
118 + go func(i int) {
119 + defer wg1.Done()
120 +
121 + err := writer.WriteMsg(msgs[i])
122 + if err != nil {
123 + errs <- err
124 + }
125 + }(i)
126 + }
127 +
128 + wg1.Wait()
129 + if err := writer.Close(); err != nil {
130 + t.Fatal(err)
131 + }
132 +
133 + for i := 0; i < len(msgs)+1; i++ {
134 + wg2.Add(1)
135 + go func(i int) {
136 + defer wg2.Done()
137 +
138 + msg2, err := reader.ReadMsg()
139 + if err != nil {
140 + if err == io.EOF {
141 + if i < len(msg2) {
142 + errs <- fmt.Errorf("failed to read all messages", len(msgs), i)
143 + }
144 + return
145 + }
146 + errs <- fmt.Errorf("unexpected error", err)
147 + }
148 +
149 + mi := NBO.Uint32(msg2[:4])
150 + msg1 := msgs[mi]
151 + if !bytes.Equal(msg1, msg2) {
152 + errs <- fmt.Errorf("message retrieved not equal\n", msg1, "\n\n", msg2)
153 + }
154 + }(i)
155 + }
156 +
157 + wg2.Wait()
158 + close(errs)
159 +
160 + if err := reader.Close(); err != nil {
161 + t.Error(err)
162 + }
163 +
164 + for e := range errs {
165 + t.Error(e)
166 + }
167 +}
Godeps/_workspace/src/github.com/jbenet/go-msgio/multipool/pool.go new
+113
@@ -0,0 +1,113 @@
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 +}
net/conn/conn.go
+6 -21
@@ -2,11 +2,11 @@ package conn
2
3 import (
4 "fmt"
5 - "sync"
5 "time"
6
7 context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
8 msgio "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio"
9 + mpool "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-msgio/mpool"
10 ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
11 manet "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
12
@@ -28,26 +28,11 @@ const (
28 HandshakeTimeout = time.Second * 5
29 )
30
31 -// global static buffer pool for byte arrays of size MaxMessageSize
32 -var BufferPool *sync.Pool
33 -
34 -func init() {
35 - BufferPool = new(sync.Pool)
36 - BufferPool.New = func() interface{} {
37 - log.Warning("Pool returning new object")
38 - return make([]byte, MaxMessageSize)
39 - }
40 -}
41 -
31 // ReleaseBuffer puts the given byte array back into the buffer pool,
32 // first verifying that it is the correct size
33 func ReleaseBuffer(b []byte) {
34 log.Warningf("Releasing buffer! (cap,size = %d, %d)", cap(b), len(b))
46 - if cap(b) != MaxMessageSize {
47 - log.Warning("Release buffer failed (cap, size = %d, %d)", cap(b), len(b))
48 - return
49 - }
50 - BufferPool.Put(b[:cap(b)])
35 + mpool.ByteSlicePool.Put(uint32(cap(b)), b)
36 }
37
38 // msgioPipe is a pipe using msgio channels.
@@ -56,10 +41,10 @@ type msgioPipe struct {
41 incoming *msgio.Chan
42 }
43
59 -func newMsgioPipe(size int, pool *sync.Pool) *msgioPipe {
44 +func newMsgioPipe(size int) *msgioPipe {
45 return &msgioPipe{
46 outgoing: msgio.NewChan(size),
62 - incoming: msgio.NewChanWithPool(size, pool),
47 + incoming: msgio.NewChan(size),
48 }
49 }
50
@@ -81,7 +66,7 @@ func newSingleConn(ctx context.Context, local, remote peer.Peer,
66 local: local,
67 remote: remote,
68 maconn: maconn,
84 - msgio: newMsgioPipe(10, BufferPool),
69 + msgio: newMsgioPipe(10),
70 }
71
72 conn.ContextCloser = ctxc.NewContextCloser(ctx, conn.close)
@@ -96,7 +81,7 @@ func newSingleConn(ctx context.Context, local, remote peer.Peer,
81 }()
82 conn.Children().Add(1)
83 go func() {
99 - conn.msgio.incoming.ReadFrom(maconn, MaxMessageSize)
84 + conn.msgio.incoming.ReadFromWithPool(maconn, &mpool.ByteSlicePool)
85 conn.Children().Done()
86 }()
87