muxer io
Juan Batiz-Benet committed
Dec 13, 2014 at 09:03 UTC
bd636e1e95cf9947d95efbf64d2fe9e924c177b3
2 files changed
+137
-297
net/mux/mux.go
+73
-87
@@ -3,16 +3,15 @@ package mux
3
4
import (
5
"errors"
6
+ "fmt"
7
"sync"
8
8
- conn "github.com/jbenet/go-ipfs/net/conn"
9
msg "github.com/jbenet/go-ipfs/net/message"
10
pb "github.com/jbenet/go-ipfs/net/mux/internal/pb"
11
u "github.com/jbenet/go-ipfs/util"
12
- ctxc "github.com/jbenet/go-ipfs/util/ctxcloser"
12
14
- context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
13
proto "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/goprotobuf/proto"
14
+ router "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-router"
15
)
16
17
var log = u.Logger("muxer")
@@ -30,7 +29,10 @@ var (
29
// encapsulates and decapsulates when interfacing with its Protocols. The
30
// Protocols do not encounter their ProtocolID.
31
type Protocol interface {
33
- GetPipe() *msg.Pipe
32
+ ProtocolID() pb.ProtocolID
33
+
34
+ // Node is a router.Node, for message connectivity.
35
+ router.Node
36
}
37
38
// ProtocolMap maps ProtocolIDs to Protocols.
@@ -39,9 +41,15 @@ type ProtocolMap map[pb.ProtocolID]Protocol
41
// Muxer is a simple multiplexor that reads + writes to Incoming and Outgoing
42
// channels. It multiplexes various protocols, wrapping and unwrapping data
43
// with a ProtocolID.
44
+//
45
+// implements router.Node and router.Route
46
type Muxer struct {
47
+ local router.Address
48
+ uplink router.Node
49
+
50
// Protocols are the multiplexed services.
51
Protocols ProtocolMap
52
+ mapLock sync.Mutex
53
54
bwiLock sync.Mutex
55
bwIn uint64
@@ -50,32 +58,16 @@ type Muxer struct {
58
bwoLock sync.Mutex
59
bwOut uint64
60
msgOut uint64
53
-
54
- *msg.Pipe
55
- ctxc.ContextCloser
61
}
62
63
// NewMuxer constructs a muxer given a protocol map.
59
-func NewMuxer(ctx context.Context, mp ProtocolMap) *Muxer {
60
- m := &Muxer{
61
- Protocols: mp,
62
- Pipe: msg.NewPipe(10),
63
- ContextCloser: ctxc.NewContextCloser(ctx, nil),
64
+// uplink is a Node to send all outgoing traffic to.
65
+func NewMuxer(local router.Address, uplink router.Node) *Muxer {
66
+ return &Muxer{
67
+ local: local,
68
+ uplink: uplink,
69
+ Protocols: ProtocolMap{},
70
}
65
-
66
- m.Children().Add(1)
67
- go m.handleIncomingMessages()
68
- for pid, proto := range m.Protocols {
69
- m.Children().Add(1)
70
- go m.handleOutgoingMessages(pid, proto)
71
- }
72
-
73
- return m
74
-}
75
-
76
-// GetPipe implements the Protocol interface
77
-func (m *Muxer) GetPipe() *msg.Pipe {
78
- return m.Pipe
71
}
72
73
// GetMessageCounts return the in/out message count measured over this muxer.
@@ -104,6 +96,9 @@ func (m *Muxer) GetBandwidthTotals() (in uint64, out uint64) {
96
97
// AddProtocol adds a Protocol with given ProtocolID to the Muxer.
98
func (m *Muxer) AddProtocol(p Protocol, pid pb.ProtocolID) error {
99
+ m.mapLock.Lock()
100
+ defer m.mapLock.Unlock()
101
+
102
if _, found := m.Protocols[pid]; found {
103
return errors.New("Another protocol already using this ProtocolID")
104
}
@@ -112,98 +107,89 @@ func (m *Muxer) AddProtocol(p Protocol, pid pb.ProtocolID) error {
107
return nil
108
}
109
115
-// handleIncoming consumes the messages on the m.Incoming channel and
116
-// routes them appropriately (to the protocols).
117
-func (m *Muxer) handleIncomingMessages() {
118
- defer m.Children().Done()
119
-
120
- for {
121
- select {
122
- case <-m.Closing():
123
- return
124
-
125
- case msg, more := <-m.Incoming:
126
- if !more {
127
- return
128
- }
129
- m.Children().Add(1)
130
- go m.handleIncomingMessage(msg)
131
- }
110
+func (m *Muxer) Address() router.Address {
111
+ return m.local
112
+}
113
+
114
+func (m *Muxer) HandlePacket(p router.Packet, from router.Node) error {
115
+ pkt, ok := p.(*msg.Packet)
116
+ if !ok {
117
+ return msg.ErrInvalidPayload
118
+ }
119
+
120
+ if from == m.uplink {
121
+ return m.handleIncomingPacket(pkt, from)
122
+ } else {
123
+ return m.handleOutgoingPacket(pkt, from)
124
}
125
}
126
135
-// handleIncomingMessage routes message to the appropriate protocol.
136
-func (m *Muxer) handleIncomingMessage(m1 msg.NetMessage) {
137
- defer m.Children().Done()
127
+// handleIncomingPacket routes message to the appropriate protocol.
128
+func (m *Muxer) handleIncomingPacket(p *msg.Packet, _ router.Node) error {
129
130
m.bwiLock.Lock()
131
// TODO: compensate for overhead
141
- m.bwIn += uint64(len(m1.Data()))
132
+ m.bwIn += uint64(len(p.Data))
133
m.msgIn++
134
m.bwiLock.Unlock()
135
145
- data, pid, err := unwrapData(m1.Data())
136
+ data, pid, err := unwrapData(p.Data)
137
if err != nil {
147
- log.Errorf("muxer de-serializing error: %v", err)
148
- return
138
+ return fmt.Errorf("muxer de-serializing error: %v", err)
139
}
150
- conn.ReleaseBuffer(m1.Data())
140
152
- m2 := msg.New(m1.Peer(), data)
141
+ // TODO: fix this when mpool is fixed.
142
+ // conn.ReleaseBuffer(m1.Data())
143
+
144
+ p.Data = data
145
+
146
+ m.mapLock.Lock()
147
proto, found := m.Protocols[pid]
148
+ m.mapLock.Unlock()
149
+
150
if !found {
155
- log.Errorf("muxer unknown protocol %v", pid)
156
- return
151
+ return fmt.Errorf("muxer: unknown protocol %v", pid)
152
}
153
159
- select {
160
- case proto.GetPipe().Incoming <- m2:
161
- case <-m.Closing():
162
- return
163
- }
154
+ log.Debugf("muxer: outgoing packet %d -> %s", proto.ProtocolID(), m.uplink.Address())
155
+ return proto.HandlePacket(p, m)
156
}
157
166
-// handleOutgoingMessages consumes the messages on the proto.Outgoing channel,
167
-// wraps them and sends them out.
168
-func (m *Muxer) handleOutgoingMessages(pid pb.ProtocolID, proto Protocol) {
169
- defer m.Children().Done()
170
-
171
- for {
172
- select {
173
- case msg, more := <-proto.GetPipe().Outgoing:
174
- if !more {
175
- return
176
- }
177
- m.handleOutgoingMessage(pid, msg)
178
-
179
- case <-m.Closing():
180
- return
158
+// handleOutgoingMessages sends out messages to the outside world
159
+func (m *Muxer) handleOutgoingPacket(p *msg.Packet, from router.Node) error {
160
+
161
+ var pid pb.ProtocolID
162
+ var proto Protocol
163
+ m.mapLock.Lock()
164
+ for pid2, proto2 := range m.Protocols {
165
+ if proto2 == from {
166
+ pid = pid2
167
+ proto = proto2
168
+ break
169
}
170
}
183
-}
171
+ m.mapLock.Unlock()
172
185
-// handleOutgoingMessage wraps out a message and sends it out the
186
-func (m *Muxer) handleOutgoingMessage(pid pb.ProtocolID, m1 msg.NetMessage) {
173
+ if proto == nil {
174
+ return errors.New("muxer: packet sent from unknown protocol")
175
+ }
176
188
- data, err := wrapData(m1.Data(), pid)
177
+ var err error
178
+ p.Data, err = wrapData(p.Data, pid)
179
if err != nil {
190
- log.Errorf("muxer serializing error: %v", err)
191
- return
180
+ return fmt.Errorf("muxer serializing error: %v", err)
181
}
182
183
m.bwoLock.Lock()
184
// TODO: compensate for overhead
185
// TODO(jbenet): switch this to a goroutine to prevent sync waiting.
197
- m.bwOut += uint64(len(data))
186
+ m.bwOut += uint64(len(p.Data))
187
m.msgOut++
188
m.bwoLock.Unlock()
189
201
- m2 := msg.New(m1.Peer(), data)
202
- select {
203
- case m.GetPipe().Outgoing <- m2:
204
- case <-m.Closing():
205
- return
206
- }
190
+ // TODO: add multiple uplinks
191
+ log.Debugf("muxer: incoming packet %s -> %d", m.uplink.Address(), proto.ProtocolID())
192
+ return m.uplink.HandlePacket(p, m)
193
}
194
195
func wrapData(data []byte, pid pb.ProtocolID) ([]byte, error) {
net/mux/mux_test.go
+64
-210
@@ -2,10 +2,7 @@ package mux
2
3
import (
4
"bytes"
5
- "fmt"
6
- "sync"
5
"testing"
8
- "time"
6
7
mh "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multihash"
8
msg "github.com/jbenet/go-ipfs/net/message"
@@ -13,15 +10,35 @@ import (
10
peer "github.com/jbenet/go-ipfs/peer"
11
testutil "github.com/jbenet/go-ipfs/util/testutil"
12
16
- context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
13
+ router "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-router"
14
)
15
16
type TestProtocol struct {
20
- *msg.Pipe
17
+ mux *Muxer
18
+ pid pb.ProtocolID
19
+ msg []*msg.Packet
20
}
21
23
-func (t *TestProtocol) GetPipe() *msg.Pipe {
24
- return t.Pipe
22
+func (t *TestProtocol) ProtocolID() pb.ProtocolID {
23
+ return t.pid
24
+}
25
+
26
+func (t *TestProtocol) Address() router.Address {
27
+ return t.pid
28
+}
29
+
30
+func (t *TestProtocol) HandlePacket(p router.Packet, from router.Node) error {
31
+ pkt, ok := p.(*msg.Packet)
32
+ if !ok {
33
+ return msg.ErrInvalidPayload
34
+ }
35
+
36
+ log.Debugf("TestProtocol %d got: %v", t, p)
37
+ if from == t.mux {
38
+ t.msg = append(t.msg, pkt)
39
+ return nil
40
+ }
41
+ return t.mux.HandlePacket(p, t)
42
}
43
44
func newPeer(t *testing.T, id string) peer.Peer {
@@ -34,14 +51,14 @@ func newPeer(t *testing.T, id string) peer.Peer {
51
return testutil.NewPeerWithID(peer.ID(mh))
52
}
53
37
-func testMsg(t *testing.T, m msg.NetMessage, data []byte) {
38
- if !bytes.Equal(data, m.Data()) {
39
- t.Errorf("Data does not match: %v != %v", data, m.Data())
54
+func testMsg(t *testing.T, m *msg.Packet, data []byte) {
55
+ if !bytes.Equal(data, m.Data) {
56
+ t.Errorf("Data does not match: %v != %v", data, m.Data)
57
}
58
}
59
43
-func testWrappedMsg(t *testing.T, m msg.NetMessage, pid pb.ProtocolID, data []byte) {
44
- data2, pid2, err := unwrapData(m.Data())
60
+func testWrappedMsg(t *testing.T, m *msg.Packet, pid pb.ProtocolID, data []byte) {
61
+ data2, pid2, err := unwrapData(m.Data)
62
if err != nil {
63
t.Error(err)
64
}
@@ -56,228 +73,65 @@ func testWrappedMsg(t *testing.T, m msg.NetMessage, pid pb.ProtocolID, data []by
73
}
74
75
func TestSimpleMuxer(t *testing.T) {
59
- ctx := context.Background()
60
-
76
// setup
62
- p1 := &TestProtocol{Pipe: msg.NewPipe(10)}
63
- p2 := &TestProtocol{Pipe: msg.NewPipe(10)}
77
+ peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
78
+ peer2 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275bbbbbb")
79
+
80
+ uplink := router.NewQueueNode("queue", make(chan router.Packet, 10))
81
+ mux1 := NewMuxer(string(peer1.ID()), uplink)
82
+
83
pid1 := pb.ProtocolID_Test
84
pid2 := pb.ProtocolID_Routing
66
- mux1 := NewMuxer(ctx, ProtocolMap{
67
- pid1: p1,
68
- pid2: p2,
69
- })
70
- peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
71
- // peer2 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275bbbbbb")
85
+ p1 := &TestProtocol{mux1, pid1, nil}
86
+ p2 := &TestProtocol{mux1, pid2, nil}
87
+ mux1.AddProtocol(p1, pid1)
88
+ mux1.AddProtocol(p2, pid2)
89
90
// test outgoing p1
91
for _, s := range []string{"foo", "bar", "baz"} {
75
- p1.Outgoing <- msg.New(peer1, []byte(s))
76
- testWrappedMsg(t, <-mux1.Outgoing, pid1, []byte(s))
77
- }
92
79
- // test incoming p1
80
- for _, s := range []string{"foo", "bar", "baz"} {
81
- d, err := wrapData([]byte(s), pid1)
82
- if err != nil {
83
- t.Error(err)
93
+ pkt := msg.Packet{Src: peer1, Dst: peer2, Data: []byte(s)}
94
+ if err := p1.HandlePacket(&pkt, nil); err != nil {
95
+ t.Fatal(err)
96
}
85
- mux1.Incoming <- msg.New(peer1, d)
86
- testMsg(t, <-p1.Incoming, []byte(s))
97
+ testWrappedMsg(t, (<-uplink.Queue()).(*msg.Packet), pid1, []byte(s))
98
}
99
89
- // test outgoing p2
90
- for _, s := range []string{"foo", "bar", "baz"} {
91
- p2.Outgoing <- msg.New(peer1, []byte(s))
92
- testWrappedMsg(t, <-mux1.Outgoing, pid2, []byte(s))
93
- }
94
-
95
- // test incoming p2
96
- for _, s := range []string{"foo", "bar", "baz"} {
97
- d, err := wrapData([]byte(s), pid2)
100
+ // test incoming p1
101
+ for i, s := range []string{"foo", "bar", "baz"} {
102
+ d, err := wrapData([]byte(s), pid1)
103
if err != nil {
104
t.Error(err)
105
}
101
- mux1.Incoming <- msg.New(peer1, d)
102
- testMsg(t, <-p2.Incoming, []byte(s))
103
- }
104
-}
105
-
106
-func TestSimultMuxer(t *testing.T) {
107
- if testing.Short() {
108
- t.SkipNow()
109
- }
110
- // run muxer
111
- ctx, cancel := context.WithCancel(context.Background())
112
-
113
- // setup
114
- p1 := &TestProtocol{Pipe: msg.NewPipe(10)}
115
- p2 := &TestProtocol{Pipe: msg.NewPipe(10)}
116
- pid1 := pb.ProtocolID_Test
117
- pid2 := pb.ProtocolID_Identify
118
- mux1 := NewMuxer(ctx, ProtocolMap{
119
- pid1: p1,
120
- pid2: p2,
121
- })
122
- peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
123
- // peer2 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275bbbbbb")
124
-
125
- // counts
126
- total := 10000
127
- speed := time.Microsecond * 1
128
- counts := [2][2][2]int{}
129
- var countsLock sync.Mutex
130
-
131
- // run producers at every end sending incrementing messages
132
- produceOut := func(pid pb.ProtocolID, size int) {
133
- limiter := time.Tick(speed)
134
- for i := 0; i < size; i++ {
135
- <-limiter
136
- s := fmt.Sprintf("proto %v out %v", pid, i)
137
- m := msg.New(peer1, []byte(s))
138
- mux1.Protocols[pid].GetPipe().Outgoing <- m
139
- countsLock.Lock()
140
- counts[pid][0][0]++
141
- countsLock.Unlock()
142
- // log.Debug("sent %v", s)
143
- }
144
- }
145
-
146
- produceIn := func(pid pb.ProtocolID, size int) {
147
- limiter := time.Tick(speed)
148
- for i := 0; i < size; i++ {
149
- <-limiter
150
- s := fmt.Sprintf("proto %v in %v", pid, i)
151
- d, err := wrapData([]byte(s), pid)
152
- if err != nil {
153
- t.Error(err)
154
- }
155
-
156
- m := msg.New(peer1, d)
157
- mux1.Incoming <- m
158
- countsLock.Lock()
159
- counts[pid][1][0]++
160
- countsLock.Unlock()
161
- // log.Debug("sent %v", s)
162
- }
163
- }
164
-
165
- consumeOut := func() {
166
- for {
167
- select {
168
- case m := <-mux1.Outgoing:
169
- data, pid, err := unwrapData(m.Data())
170
- if err != nil {
171
- t.Error(err)
172
- }
173
-
174
- // log.Debug("got %v", string(data))
175
- _ = data
176
- countsLock.Lock()
177
- counts[pid][1][1]++
178
- countsLock.Unlock()
179
-
180
- case <-ctx.Done():
181
- return
182
- }
183
- }
184
- }
106
186
- consumeIn := func(pid pb.ProtocolID) {
187
- for {
188
- select {
189
- case m := <-mux1.Protocols[pid].GetPipe().Incoming:
190
- countsLock.Lock()
191
- counts[pid][0][1]++
192
- countsLock.Unlock()
193
- // log.Debug("got %v", string(m.Data()))
194
- _ = m
195
- case <-ctx.Done():
196
- return
197
- }
107
+ pkt := msg.Packet{Src: peer1, Dst: peer2, Data: d}
108
+ if err := mux1.HandlePacket(&pkt, uplink); err != nil {
109
+ t.Fatal(err)
110
}
111
+ testMsg(t, p1.msg[i], []byte(s))
112
}
113
201
- go produceOut(pid1, total)
202
- go produceOut(pid2, total)
203
- go produceIn(pid1, total)
204
- go produceIn(pid2, total)
205
- go consumeOut()
206
- go consumeIn(pid1)
207
- go consumeIn(pid2)
208
-
209
- limiter := time.Tick(speed)
210
- for {
211
- <-limiter
212
- countsLock.Lock()
213
- got := counts[0][0][0] + counts[0][0][1] +
214
- counts[0][1][0] + counts[0][1][1] +
215
- counts[1][0][0] + counts[1][0][1] +
216
- counts[1][1][0] + counts[1][1][1]
217
- countsLock.Unlock()
114
+ // test outgoing p2
115
+ for _, s := range []string{"foo", "bar", "baz"} {
116
219
- if got == total*8 {
220
- cancel()
221
- return
117
+ pkt := msg.Packet{Src: peer1, Dst: peer2, Data: []byte(s)}
118
+ if err := p2.HandlePacket(&pkt, nil); err != nil {
119
+ t.Fatal(err)
120
}
121
+ testWrappedMsg(t, (<-uplink.Queue()).(*msg.Packet), pid2, []byte(s))
122
}
123
225
-}
226
-
227
-func TestStopping(t *testing.T) {
228
- ctx := context.Background()
229
-
230
- // setup
231
- p1 := &TestProtocol{Pipe: msg.NewPipe(10)}
232
- p2 := &TestProtocol{Pipe: msg.NewPipe(10)}
233
- pid1 := pb.ProtocolID_Test
234
- pid2 := pb.ProtocolID_Identify
235
- mux1 := NewMuxer(ctx, ProtocolMap{
236
- pid1: p1,
237
- pid2: p2,
238
- })
239
- peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
240
- // peer2 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275bbbbbb")
241
-
242
- // test outgoing p1
243
- for _, s := range []string{"foo1", "bar1", "baz1"} {
244
- p1.Outgoing <- msg.New(peer1, []byte(s))
245
- testWrappedMsg(t, <-mux1.Outgoing, pid1, []byte(s))
246
- }
247
-
248
- // test incoming p1
249
- for _, s := range []string{"foo2", "bar2", "baz2"} {
250
- d, err := wrapData([]byte(s), pid1)
124
+ // test incoming p2
125
+ for i, s := range []string{"foo", "bar", "baz"} {
126
+ d, err := wrapData([]byte(s), pid2)
127
if err != nil {
252
- t.Error(err)
128
+ t.Fatal(err)
129
}
254
- mux1.Incoming <- msg.New(peer1, d)
255
- testMsg(t, <-p1.Incoming, []byte(s))
256
- }
257
-
258
- mux1.Close() // waits
130
260
- // test outgoing p1
261
- for _, s := range []string{"foo3", "bar3", "baz3"} {
262
- p1.Outgoing <- msg.New(peer1, []byte(s))
263
- select {
264
- case m := <-mux1.Outgoing:
265
- t.Errorf("should not have received anything. Got: %v", string(m.Data()))
266
- case <-time.After(time.Millisecond):
267
- }
268
- }
269
-
270
- // test incoming p1
271
- for _, s := range []string{"foo4", "bar4", "baz4"} {
272
- d, err := wrapData([]byte(s), pid1)
273
- if err != nil {
274
- t.Error(err)
275
- }
276
- mux1.Incoming <- msg.New(peer1, d)
277
- select {
278
- case <-p1.Incoming:
279
- t.Error("should not have received anything.")
280
- case <-time.After(time.Millisecond):
131
+ pkt := msg.Packet{Src: peer1, Dst: peer2, Data: d}
132
+ if err := mux1.HandlePacket(&pkt, uplink); err != nil {
133
+ t.Fatal(err)
134
}
135
+ testMsg(t, p2.msg[i], []byte(s))
136
}
137
}