msgio pooling first hack
Jeromy committed
Oct 27, 2014 at 20:23 UTC
07733b17b3fa5be48a89d497ca7649fef7ef3d73
8 files changed
+97
-26
Godeps/_workspace/src/github.com/jbenet/go-msgio/chan.go
+9
-2
@@ -2,6 +2,7 @@ package msgio
2
3
import (
4
"io"
5
+ "sync"
6
)
7
8
type Chan struct {
@@ -9,13 +10,15 @@ type Chan struct {
10
MsgChan chan []byte
11
ErrChan chan error
12
CloseChan chan bool
13
+ BufPool *sync.Pool
14
}
15
14
-func NewChan(chanSize int) *Chan {
16
+func NewChan(chanSize int, pool *sync.Pool) *Chan {
17
return &Chan{
18
MsgChan: make(chan []byte, chanSize),
19
ErrChan: make(chan error, 1),
20
CloseChan: make(chan bool, 2),
21
+ BufPool: pool,
22
}
23
}
24
@@ -25,7 +28,11 @@ func (s *Chan) ReadFrom(r io.Reader, maxMsgLen int) {
28
mr := NewReader(r)
29
Loop:
30
for {
28
- buf := make([]byte, maxMsgLen)
31
+ bufi := s.BufPool.Get()
32
+ buf, ok := bufi.([]byte)
33
+ if !ok {
34
+ panic("Got invalid type from sync pool!")
35
+ }
36
l, err := mr.ReadMsg(buf)
37
if err != nil {
38
if err == io.EOF {
Godeps/_workspace/src/github.com/jbenet/go-msgio/chan_test.go
+4
-2
@@ -5,6 +5,7 @@ import (
5
randbuf "github.com/jbenet/go-randbuf"
6
"io"
7
"math/rand"
8
+ "sync"
9
"testing"
10
"time"
11
)
@@ -12,7 +13,8 @@ import (
13
func TestReadChan(t *testing.T) {
14
buf := bytes.NewBuffer(nil)
15
writer := NewWriter(buf)
15
- rchan := NewChan(10)
16
+ p := &sync.Pool{New: func() interface{} { return make([]byte, 1000) }}
17
+ rchan := NewChan(10, p)
18
msgs := [1000][]byte{}
19
20
r := rand.New(rand.NewSource(time.Now().UnixNano()))
@@ -58,7 +60,7 @@ Loop:
60
func TestWriteChan(t *testing.T) {
61
buf := bytes.NewBuffer(nil)
62
reader := NewReader(buf)
61
- wchan := NewChan(10)
63
+ wchan := NewChan(10, nil)
64
msgs := [1000][]byte{}
65
66
go wchan.WriteTo(buf)
crypto/spipe/handshake.go
+35
-4
@@ -184,10 +184,41 @@ func (s *SecurePipe) handshake() error {
184
}
185
186
cmp := bytes.Compare(myPubKey, proposeResp.GetPubkey())
187
- mIV, tIV, mCKey, tCKey, mMKey, tMKey := ci.KeyStretcher(cmp, cipherType, hashType, secret)
188
-
189
- go s.handleSecureIn(hashType, cipherType, tIV, tCKey, tMKey)
190
- go s.handleSecureOut(hashType, cipherType, mIV, mCKey, mMKey)
187
+ //mIV, tIV, mCKey, tCKey, mMKey, tMKey := ci.KeyStretcher(cmp, cipherType, hashType, secret)
188
+ ci.KeyStretcher(cmp, cipherType, hashType, secret)
189
+
190
+ //go s.handleSecureIn(hashType, cipherType, tIV, tCKey, tMKey)
191
+ //go s.handleSecureOut(hashType, cipherType, mIV, mCKey, mMKey)
192
+
193
+ // Disable Secure Channel
194
+ go func(sp *SecurePipe) {
195
+ for {
196
+ select {
197
+ case <-sp.ctx.Done():
198
+ return
199
+ case m, ok := <-sp.insecure.In:
200
+ if !ok {
201
+ sp.cancel()
202
+ return
203
+ }
204
+ sp.In <- m
205
+ }
206
+ }
207
+ }(s)
208
+ go func(sp *SecurePipe) {
209
+ for {
210
+ select {
211
+ case <-sp.ctx.Done():
212
+ return
213
+ case m, ok := <-sp.Out:
214
+ if !ok {
215
+ sp.cancel()
216
+ return
217
+ }
218
+ sp.insecure.Out <- m
219
+ }
220
+ }
221
+ }(s)
222
223
finished := []byte("Finished")
224
merkledag/merkledag.go
+18
@@ -241,3 +241,21 @@ func (n *dagService) Remove(nd *Node) error {
241
}
242
return n.Blocks.DeleteBlock(k)
243
}
244
+
245
+func FetchGraph(ctx context.Context, root *Node, serv *DAGService) {
246
+ for _, l := range root.Links {
247
+ go func(lnk *Link) {
248
+ select {
249
+ case <-ctx.Done():
250
+ return
251
+ }
252
+
253
+ nd, err := lnk.GetNode(serv)
254
+ if err != nil {
255
+ log.Error(err)
256
+ return
257
+ }
258
+ FetchGraph(ctx, nd, serv)
259
+ }(l)
260
+ }
261
+}
net/conn/conn.go
+20
-4
@@ -2,6 +2,7 @@ package conn
2
3
import (
4
"fmt"
5
+ "sync"
6
"time"
7
8
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
@@ -27,16 +28,31 @@ const (
28
HandshakeTimeout = time.Second * 5
29
)
30
31
+var BufferPool *sync.Pool
32
+
33
+func init() {
34
+ BufferPool = new(sync.Pool)
35
+ BufferPool.New = func() interface{} {
36
+ log.Warning("Pool returning new object")
37
+ return make([]byte, MaxMessageSize)
38
+ }
39
+}
40
+
41
+func ReleaseBuffer(b []byte) {
42
+ log.Warningf("Releasing buffer! (size = %d)", cap(b))
43
+ BufferPool.Put(b[:cap(b)])
44
+}
45
+
46
// msgioPipe is a pipe using msgio channels.
47
type msgioPipe struct {
48
outgoing *msgio.Chan
49
incoming *msgio.Chan
50
}
51
36
-func newMsgioPipe(size int) *msgioPipe {
52
+func newMsgioPipe(size int, pool *sync.Pool) *msgioPipe {
53
return &msgioPipe{
38
- outgoing: msgio.NewChan(10),
39
- incoming: msgio.NewChan(10),
54
+ outgoing: msgio.NewChan(size, nil),
55
+ incoming: msgio.NewChan(size, pool),
56
}
57
}
58
@@ -58,7 +74,7 @@ func newSingleConn(ctx context.Context, local, remote peer.Peer,
74
local: local,
75
remote: remote,
76
maconn: maconn,
61
- msgio: newMsgioPipe(10),
77
+ msgio: newMsgioPipe(10, BufferPool),
78
}
79
80
conn.ContextCloser = ctxc.NewContextCloser(ctx, conn.close)
net/mux/mux.go
+3
-1
@@ -1,9 +1,10 @@
1
-package mux
1
+tpackage mux
2
3
import (
4
"errors"
5
"sync"
6
7
+ conn "github.com/jbenet/go-ipfs/net/conn"
8
msg "github.com/jbenet/go-ipfs/net/message"
9
pb "github.com/jbenet/go-ipfs/net/mux/internal/pb"
10
u "github.com/jbenet/go-ipfs/util"
@@ -130,6 +131,7 @@ func (m *Muxer) handleIncomingMessage(m1 msg.NetMessage) {
131
log.Errorf("muxer de-serializing error: %v", err)
132
return
133
}
134
+ conn.ReleaseBuffer(m1.Data())
135
136
m2 := msg.New(m1.Peer(), data)
137
proto, found := m.Protocols[pid]
routing/dht/pb/dht.pb.go
+1
-1
@@ -13,7 +13,7 @@ It has these top-level messages:
13
*/
14
package dht_pb
15
16
-import proto "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/gogoprotobuf/proto"
16
+import proto "code.google.com/p/gogoprotobuf/proto"
17
import math "math"
18
19
// Reference imports to suppress errors if they are not otherwise used.
unixfs/io/dagreader.go
+7
-12
@@ -9,7 +9,6 @@ import (
9
mdag "github.com/jbenet/go-ipfs/merkledag"
10
ft "github.com/jbenet/go-ipfs/unixfs"
11
ftpb "github.com/jbenet/go-ipfs/unixfs/pb"
12
- u "github.com/jbenet/go-ipfs/util"
12
)
13
14
var ErrIsDir = errors.New("this dag node is a directory")
@@ -36,11 +35,12 @@ func NewDagReader(n *mdag.Node, serv mdag.DAGService) (io.Reader, error) {
35
// Dont allow reading directories
36
return nil, ErrIsDir
37
case ftpb.Data_File:
39
- return &DagReader{
38
+ dr := &DagReader{
39
node: n,
40
serv: serv,
41
buf: bytes.NewBuffer(pb.GetData()),
43
- }, nil
42
+ }
43
+ return dr, nil
44
case ftpb.Data_Raw:
45
// Raw block will just be a single level, return a byte buffer
46
return bytes.NewBuffer(pb.GetData()), nil
@@ -55,17 +55,12 @@ func (dr *DagReader) precalcNextBuf() error {
55
if dr.position >= len(dr.node.Links) {
56
return io.EOF
57
}
58
- nxtLink := dr.node.Links[dr.position]
59
- nxt := nxtLink.Node
60
- if nxt == nil {
61
- nxtNode, err := dr.serv.Get(u.Key(nxtLink.Hash))
62
- if err != nil {
63
- return err
64
- }
65
- nxt = nxtNode
58
+ nxt, err := dr.node.Links[dr.position].GetNode(dr.serv)
59
+ if err != nil {
60
+ return err
61
}
62
pb := new(ftpb.Data)
68
- err := proto.Unmarshal(nxt.Data, pb)
63
+ err = proto.Unmarshal(nxt.Data, pb)
64
if err != nil {
65
return err
66
}