net: threadsafe mux handler add
Juan Batiz-Benet committed
Dec 16, 2014 at 06:20 UTC
c150668a1c7c9f2c210aa68b8b993ca88ac5cebf
3 files changed
+62
-37
net/interface.go
+10
-26
@@ -11,10 +11,12 @@ import (
11
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
12
)
13
14
+type ProtocolID string
15
+
16
const (
15
- ProtocolBitswap = "/ipfs/bitswap"
16
- ProtocolDHT = "/ipfs/dht"
17
- ProtocolDiag = "/ipfs/diagnostics"
17
+ ProtocolBitswap ProtocolID = "/ipfs/bitswap"
18
+ ProtocolDHT ProtocolID = "/ipfs/dht"
19
+ ProtocolDiag ProtocolID = "/ipfs/diagnostics"
20
)
21
22
// Stream represents a bidirectional channel between two agents in
@@ -34,7 +36,7 @@ type Stream interface {
36
// incoming streams must implement.
37
type StreamHandler func(Stream)
38
37
-type StreamHandlerMap map[string]StreamHandler
39
+type StreamHandlerMap map[ProtocolID]StreamHandler
40
41
// Conn is a connection to a remote peer. It multiplexes streams.
42
// Usually there is no need to use a Conn directly, but it may
@@ -47,28 +49,6 @@ type Conn interface {
49
NewStream(p peer.Peer) (Stream, error)
50
}
51
50
-// Mux provides simple stream multixplexing.
51
-// It helps you precisely when:
52
-// * You have many streams
53
-// * You have function handlers
54
-//
55
-// It contains the handlers for each protocol accepted.
56
-// It dispatches handlers for streams opened by remote peers.
57
-//
58
-// We use a totally ad-hoc encoding:
59
-// <1 byte length in bytes><string name>
60
-// So "bitswap" is 0x0762697473776170
61
-//
62
-// NOTE: only the dialer specifies this muxing line.
63
-// This is because we're using Streams :)
64
-//
65
-// WARNING: this datastructure IS NOT threadsafe.
66
-// do not modify it once the network is using it.
67
-type Mux struct {
68
- Default StreamHandler // handles unknown protocols.
69
- Handlers StreamHandlerMap
70
-}
71
-
52
// Network is the interface IPFS uses for connecting to the world.
53
// It dials and listens for connections. it uses a Swarm to pool
54
// connnections (see swarm pkg, and peerstream.Swarm). Connections
@@ -77,6 +57,10 @@ type Network interface {
57
Dialer
58
io.Closer
59
60
+ // SetHandler sets the protocol handler on the Network's Muxer.
61
+ // This operation is threadsafe.
62
+ SetHandler(ProtocolID, StreamHandler)
63
+
64
// NewStream returns a new stream to given peer p.
65
// If there is no connection to p, attempts to create one.
66
NewStream(p peer.Peer) (Stream, error)
net/mux.go
+41
-6
@@ -4,6 +4,7 @@ import (
4
"errors"
5
"fmt"
6
"io"
7
+ "sync"
8
9
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
10
eventlog "github.com/jbenet/go-ipfs/util/eventlog"
@@ -12,6 +13,30 @@ import (
13
14
var log = eventlog.Logger("mux2")
15
16
+// Mux provides simple stream multixplexing.
17
+// It helps you precisely when:
18
+// * You have many streams
19
+// * You have function handlers
20
+//
21
+// It contains the handlers for each protocol accepted.
22
+// It dispatches handlers for streams opened by remote peers.
23
+//
24
+// We use a totally ad-hoc encoding:
25
+// <1 byte length in bytes><string name>
26
+// So "bitswap" is 0x0762697473776170
27
+//
28
+// NOTE: only the dialer specifies this muxing line.
29
+// This is because we're using Streams :)
30
+//
31
+// WARNING: this datastructure IS NOT threadsafe.
32
+// do not modify it once the network is using it.
33
+type Mux struct {
34
+ Default StreamHandler // handles unknown protocols.
35
+ Handlers StreamHandlerMap
36
+
37
+ sync.RWMutex
38
+}
39
+
40
// NextName reads the stream and returns the next protocol name
41
// according to the muxer encoding.
42
func (m *Mux) NextName(s io.Reader) (string, error) {
@@ -41,16 +66,26 @@ func (m *Mux) NextHandler(s io.Reader) (string, StreamHandler, error) {
66
return "", nil, err
67
}
68
44
- h, found := m.Handlers[name]
45
- if !found {
46
- if m.Default == nil {
47
- return name, nil, errors.New("no handler with name: " + name)
48
- }
69
+ m.RLock()
70
+ h, found := m.Handlers[ProtocolID(name)]
71
+ m.RUnlock()
72
73
+ switch {
74
+ case !found && m.Default != nil:
75
return name, m.Default, nil
76
+ case !found && m.Default == nil:
77
+ return name, nil, errors.New("no handler with name: " + name)
78
+ default:
79
+ return name, h, nil
80
}
81
+}
82
53
- return name, h, nil
83
+// SetHandler sets the protocol handler on the Network's Muxer.
84
+// This operation is threadsafe.
85
+func (m *Mux) SetHandler(p ProtocolID, h StreamHandler) {
86
+ m.Lock()
87
+ m.Handlers[p] = h
88
+ m.Unlock()
89
}
90
91
// Handle reads the next name off the Stream, and calls a function
net/net.go
+11
-5
@@ -81,9 +81,9 @@ type network struct {
81
cg ctxgroup.ContextGroup // for Context closing
82
}
83
84
-// NewConn is the structure that implements the network interface
85
-func NewConn(ctx context.Context, listen []ma.Multiaddr, local peer.Peer,
86
- peers peer.Peerstore, m Mux) (*network, error) {
84
+// NewNetwork constructs a new network and starts listening on given addresses.
85
+func NewNetwork(ctx context.Context, listen []ma.Multiaddr, local peer.Peer,
86
+ peers peer.Peerstore) (*network, error) {
87
88
s, err := swarm.NewSwarm(ctx, listen, local, peers)
89
if err != nil {
@@ -93,12 +93,12 @@ func NewConn(ctx context.Context, listen []ma.Multiaddr, local peer.Peer,
93
n := &network{
94
local: local,
95
swarm: s,
96
- mux: m,
96
+ mux: Mux{},
97
cg: ctxgroup.WithContext(ctx),
98
}
99
100
s.SetStreamHandler(func(s *swarm.Stream) {
101
- m.Handle((*stream)(s))
101
+ n.mux.Handle((*stream)(s))
102
})
103
104
n.cg.AddChildGroup(s.CtxGroup())
@@ -153,3 +153,9 @@ func (n *network) Connectedness(p peer.Peer) Connectedness {
153
}
154
return NotConnected
155
}
156
+
157
+// SetHandler sets the protocol handler on the Network's Muxer.
158
+// This operation is threadsafe.
159
+func (n *network) SetHandler(p ProtocolID, h StreamHandler) {
160
+ n.mux.SetHandler(p, h)
161
+}