muxer now uses ctxCloser
Juan Batiz-Benet committed
Oct 25, 2014 at 00:48 UTC
93497c2d00fe26e11986a45936a54630104d04e1
4 files changed
+48
-85
net/mux/mux.go
+32
-58
@@ -7,6 +7,7 @@ import (
7
msg "github.com/jbenet/go-ipfs/net/message"
8
pb "github.com/jbenet/go-ipfs/net/mux/internal/pb"
9
u "github.com/jbenet/go-ipfs/util"
10
+ ctxc "github.com/jbenet/go-ipfs/util/ctxcloser"
11
12
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,6 +15,8 @@ import (
15
16
var log = u.Logger("muxer")
17
18
+// ProtocolIDs used to identify each protocol.
19
+// These should probably be defined elsewhere.
20
var (
21
ProtocolID_Routing = pb.ProtocolID_Routing
22
ProtocolID_Exchange = pb.ProtocolID_Exchange
@@ -38,11 +41,6 @@ type Muxer struct {
41
// Protocols are the multiplexed services.
42
Protocols ProtocolMap
43
41
- // cancel is the function to stop the Muxer
42
- cancel context.CancelFunc
43
- ctx context.Context
44
- wg sync.WaitGroup
45
-
44
bwiLock sync.Mutex
45
bwIn uint64
46
@@ -50,45 +48,33 @@ type Muxer struct {
48
bwOut uint64
49
50
*msg.Pipe
51
+ ctxc.ContextCloser
52
}
53
54
// NewMuxer constructs a muxer given a protocol map.
56
-func NewMuxer(mp ProtocolMap) *Muxer {
57
- return &Muxer{
58
- Protocols: mp,
59
- Pipe: msg.NewPipe(10),
55
+func NewMuxer(ctx context.Context, mp ProtocolMap) *Muxer {
56
+ m := &Muxer{
57
+ Protocols: mp,
58
+ Pipe: msg.NewPipe(10),
59
+ ContextCloser: ctxc.NewContextCloser(ctx, nil),
60
}
61
-}
62
-
63
-// GetPipe implements the Protocol interface
64
-func (m *Muxer) GetPipe() *msg.Pipe {
65
- return m.Pipe
66
-}
61
68
-// Start kicks off the Muxer goroutines.
69
-func (m *Muxer) Start(ctx context.Context) error {
70
- if m == nil {
71
- panic("nix muxer")
72
- }
73
-
74
- if m.cancel != nil {
75
- return errors.New("Muxer already started.")
76
- }
77
-
78
- // make a cancellable context.
79
- m.ctx, m.cancel = context.WithCancel(ctx)
80
- m.wg = sync.WaitGroup{}
81
-
82
- m.wg.Add(1)
62
+ m.Children().Add(1)
63
go m.handleIncomingMessages()
64
for pid, proto := range m.Protocols {
85
- m.wg.Add(1)
65
+ m.Children().Add(1)
66
go m.handleOutgoingMessages(pid, proto)
67
}
68
89
- return nil
69
+ return m
70
}
71
72
+// GetPipe implements the Protocol interface
73
+func (m *Muxer) GetPipe() *msg.Pipe {
74
+ return m.Pipe
75
+}
76
+
77
+// GetBandwidthTotals return the in/out bandwidth measured over this muxer.
78
func (m *Muxer) GetBandwidthTotals() (in uint64, out uint64) {
79
m.bwiLock.Lock()
80
in = m.bwIn
@@ -100,19 +86,6 @@ func (m *Muxer) GetBandwidthTotals() (in uint64, out uint64) {
86
return
87
}
88
103
-// Stop stops muxer activity.
104
-func (m *Muxer) Stop() {
105
- if m.cancel == nil {
106
- panic("muxer stopped twice.")
107
- }
108
- // issue cancel, and wipe func.
109
- m.cancel()
110
- m.cancel = context.CancelFunc(nil)
111
-
112
- // wait for everything to wind down.
113
- m.wg.Wait()
114
-}
115
-
89
// AddProtocol adds a Protocol with given ProtocolID to the Muxer.
90
func (m *Muxer) AddProtocol(p Protocol, pid pb.ProtocolID) error {
91
if _, found := m.Protocols[pid]; found {
@@ -126,28 +99,26 @@ func (m *Muxer) AddProtocol(p Protocol, pid pb.ProtocolID) error {
99
// handleIncoming consumes the messages on the m.Incoming channel and
100
// routes them appropriately (to the protocols).
101
func (m *Muxer) handleIncomingMessages() {
129
- defer m.wg.Done()
102
+ defer m.Children().Done()
103
104
for {
132
- if m == nil {
133
- panic("nil muxer")
134
- }
135
-
105
select {
106
+ case <-m.Closing():
107
+ return
108
+
109
case msg, more := <-m.Incoming:
110
if !more {
111
return
112
}
113
+ m.Children().Add(1)
114
go m.handleIncomingMessage(msg)
142
-
143
- case <-m.ctx.Done():
144
- return
115
}
116
}
117
}
118
119
// handleIncomingMessage routes message to the appropriate protocol.
120
func (m *Muxer) handleIncomingMessage(m1 msg.NetMessage) {
121
+ defer m.Children().Done()
122
123
m.bwiLock.Lock()
124
// TODO: compensate for overhead
@@ -169,8 +140,7 @@ func (m *Muxer) handleIncomingMessage(m1 msg.NetMessage) {
140
141
select {
142
case proto.GetPipe().Incoming <- m2:
172
- case <-m.ctx.Done():
173
- log.Error(m.ctx.Err())
143
+ case <-m.Closing():
144
return
145
}
146
}
@@ -178,7 +148,7 @@ func (m *Muxer) handleIncomingMessage(m1 msg.NetMessage) {
148
// handleOutgoingMessages consumes the messages on the proto.Outgoing channel,
149
// wraps them and sends them out.
150
func (m *Muxer) handleOutgoingMessages(pid pb.ProtocolID, proto Protocol) {
181
- defer m.wg.Done()
151
+ defer m.Children().Done()
152
153
for {
154
select {
@@ -186,9 +156,10 @@ func (m *Muxer) handleOutgoingMessages(pid pb.ProtocolID, proto Protocol) {
156
if !more {
157
return
158
}
159
+ m.Children().Add(1)
160
go m.handleOutgoingMessage(pid, msg)
161
191
- case <-m.ctx.Done():
162
+ case <-m.Closing():
163
return
164
}
165
}
@@ -196,6 +167,8 @@ func (m *Muxer) handleOutgoingMessages(pid pb.ProtocolID, proto Protocol) {
167
168
// handleOutgoingMessage wraps out a message and sends it out the
169
func (m *Muxer) handleOutgoingMessage(pid pb.ProtocolID, m1 msg.NetMessage) {
170
+ defer m.Children().Done()
171
+
172
data, err := wrapData(m1.Data(), pid)
173
if err != nil {
174
log.Errorf("muxer serializing error: %v", err)
@@ -204,13 +177,14 @@ func (m *Muxer) handleOutgoingMessage(pid pb.ProtocolID, m1 msg.NetMessage) {
177
178
m.bwoLock.Lock()
179
// TODO: compensate for overhead
180
+ // TODO(jbenet): switch this to a goroutine to prevent sync waiting.
181
m.bwOut += uint64(len(data))
182
m.bwoLock.Unlock()
183
184
m2 := msg.New(m1.Peer(), data)
185
select {
186
case m.GetPipe().Outgoing <- m2:
213
- case <-m.ctx.Done():
187
+ case <-m.Closing():
188
return
189
}
190
}
net/mux/mux_test.go
+8
-19
@@ -54,23 +54,20 @@ func testWrappedMsg(t *testing.T, m msg.NetMessage, pid pb.ProtocolID, data []by
54
}
55
56
func TestSimpleMuxer(t *testing.T) {
57
+ ctx := context.Background()
58
59
// setup
60
p1 := &TestProtocol{Pipe: msg.NewPipe(10)}
61
p2 := &TestProtocol{Pipe: msg.NewPipe(10)}
62
pid1 := pb.ProtocolID_Test
63
pid2 := pb.ProtocolID_Routing
63
- mux1 := NewMuxer(ProtocolMap{
64
+ mux1 := NewMuxer(ctx, ProtocolMap{
65
pid1: p1,
66
pid2: p2,
67
})
68
peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
69
// peer2 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275bbbbbb")
70
70
- // run muxer
71
- ctx := context.Background()
72
- mux1.Start(ctx)
73
-
71
// test outgoing p1
72
for _, s := range []string{"foo", "bar", "baz"} {
73
p1.Outgoing <- msg.New(peer1, []byte(s))
@@ -105,23 +102,21 @@ func TestSimpleMuxer(t *testing.T) {
102
}
103
104
func TestSimultMuxer(t *testing.T) {
105
+ // run muxer
106
+ ctx, cancel := context.WithCancel(context.Background())
107
108
// setup
109
p1 := &TestProtocol{Pipe: msg.NewPipe(10)}
110
p2 := &TestProtocol{Pipe: msg.NewPipe(10)}
111
pid1 := pb.ProtocolID_Test
112
pid2 := pb.ProtocolID_Identify
114
- mux1 := NewMuxer(ProtocolMap{
113
+ mux1 := NewMuxer(ctx, ProtocolMap{
114
pid1: p1,
115
pid2: p2,
116
})
117
peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
118
// peer2 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275bbbbbb")
119
121
- // run muxer
122
- ctx, cancel := context.WithCancel(context.Background())
123
- mux1.Start(ctx)
124
-
120
// counts
121
total := 10000
122
speed := time.Microsecond * 1
@@ -214,22 +209,20 @@ func TestSimultMuxer(t *testing.T) {
209
}
210
211
func TestStopping(t *testing.T) {
212
+ ctx := context.Background()
213
214
// setup
215
p1 := &TestProtocol{Pipe: msg.NewPipe(10)}
216
p2 := &TestProtocol{Pipe: msg.NewPipe(10)}
217
pid1 := pb.ProtocolID_Test
218
pid2 := pb.ProtocolID_Identify
223
- mux1 := NewMuxer(ProtocolMap{
219
+ mux1 := NewMuxer(ctx, ProtocolMap{
220
pid1: p1,
221
pid2: p2,
222
})
223
peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
224
// peer2 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275bbbbbb")
225
230
- // run muxer
231
- mux1.Start(context.Background())
232
-
226
// test outgoing p1
227
for _, s := range []string{"foo1", "bar1", "baz1"} {
228
p1.Outgoing <- msg.New(peer1, []byte(s))
@@ -246,10 +239,7 @@ func TestStopping(t *testing.T) {
239
testMsg(t, <-p1.Incoming, []byte(s))
240
}
241
249
- mux1.Stop()
250
- if mux1.cancel != nil {
251
- t.Error("mux.cancel should be nil")
252
- }
242
+ mux1.Close() // waits
243
244
// test outgoing p1
245
for _, s := range []string{"foo3", "bar3", "baz3"} {
@@ -274,5 +264,4 @@ func TestStopping(t *testing.T) {
264
case <-time.After(time.Millisecond):
265
}
266
}
277
-
267
}
net/net.go
+3
-8
@@ -36,17 +36,12 @@ func NewIpfsNetwork(ctx context.Context, local peer.Peer,
36
37
in := &IpfsNetwork{
38
local: local,
39
- muxer: mux.NewMuxer(*pmap),
39
+ muxer: mux.NewMuxer(ctx, *pmap),
40
ctx: ctx,
41
cancel: cancel,
42
}
43
44
- err := in.muxer.Start(ctx)
45
- if err != nil {
46
- cancel()
47
- return nil, err
48
- }
49
-
44
+ var err error
45
in.swarm, err = swarm.NewSwarm(ctx, local, peers)
46
if err != nil {
47
cancel()
@@ -101,7 +96,7 @@ func (n *IpfsNetwork) Close() error {
96
}
97
98
n.swarm.Close()
104
- n.muxer.Stop()
99
+ n.muxer.Close()
100
101
n.cancel()
102
n.cancel = nil
util/ctxcloser/closer.go
+5
@@ -9,6 +9,8 @@ import (
9
// CloseFunc is a function used to close a ContextCloser
10
type CloseFunc func() error
11
12
+var nilCloseFunc = func() error { return nil }
13
+
14
// ContextCloser is an interface for services able to be opened and closed.
15
// It has a parent Context, and Children. But ContextCloser is not a proper
16
// "tree" like the Context tree. It is more like a Context-WaitGroup hybrid.
@@ -92,6 +94,9 @@ type contextCloser struct {
94
// NewContextCloser constructs and returns a ContextCloser. It will call
95
// cf CloseFunc before its Done() Wait signals fire.
96
func NewContextCloser(ctx context.Context, cf CloseFunc) ContextCloser {
97
+ if cf == nil {
98
+ cf = nilCloseFunc
99
+ }
100
ctx, cancel := context.WithCancel(ctx)
101
c := &contextCloser{
102
ctx: ctx,