net/service now uses ctxcloser
Juan Batiz-Benet committed
Oct 25, 2014 at 04:44 UTC
f2b8803a572ea4a6d9c96355a920af63deb69298
4 files changed
+95
-83
core/core.go
+3
-13
@@ -115,19 +115,9 @@ func NewIpfsNode(cfg *config.Config, online bool) (*IpfsNode, error) {
115
116
if online {
117
118
- dhtService := netservice.NewService(nil) // nil handler for now, need to patch it
119
- exchangeService := netservice.NewService(nil) // nil handler for now, need to patch it
120
- diagService := netservice.NewService(nil)
121
-
122
- if err := dhtService.Start(ctx); err != nil {
123
- return nil, err
124
- }
125
- if err := exchangeService.Start(ctx); err != nil {
126
- return nil, err
127
- }
128
- if err := diagService.Start(ctx); err != nil {
129
- return nil, err
130
- }
118
+ dhtService := netservice.NewService(ctx, nil) // nil handler for now, need to patch it
119
+ exchangeService := netservice.NewService(ctx, nil) // nil handler for now, need to patch it
120
+ diagService := netservice.NewService(ctx, nil)
121
122
net, err = inet.NewIpfsNetwork(ctx, local, peerstore, &mux.ProtocolMap{
123
mux.ProtocolID_Routing: dhtService,
net/service/service.go
+48
-40
@@ -2,10 +2,12 @@ package service
2
3
import (
4
"errors"
5
+ "fmt"
6
"sync"
7
8
msg "github.com/jbenet/go-ipfs/net/message"
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
)
@@ -39,10 +41,7 @@ type Sender interface {
41
// incomig (SetHandler) requests.
42
type Service interface {
43
Sender
42
-
43
- // Start + Stop Service
44
- Start(ctx context.Context) error
45
- Stop()
44
+ ctxc.ContextCloser
45
46
// GetPipe
47
GetPipe() *msg.Pipe
@@ -56,45 +55,30 @@ type Service interface {
55
// messages over the same channel, and to issue + handle requests.
56
type service struct {
57
// Handler is the object registered to handle incoming requests.
59
- Handler Handler
58
+ Handler Handler
59
+ HandlerLock sync.RWMutex
60
61
// Requests are all the pending requests on this service.
62
Requests RequestMap
63
RequestsLock sync.RWMutex
64
65
- // cancel is the function to stop the Service
66
- cancel context.CancelFunc
67
-
65
// Message Pipe (connected to the outside world)
66
*msg.Pipe
67
+ ctxc.ContextCloser
68
}
69
70
// NewService creates a service object with given type ID and Handler
73
-func NewService(h Handler) Service {
74
- return &service{
75
- Handler: h,
76
- Requests: RequestMap{},
77
- Pipe: msg.NewPipe(10),
78
- }
79
-}
80
-
81
-// Start kicks off the Service goroutines.
82
-func (s *service) Start(ctx context.Context) error {
83
- if s.cancel != nil {
84
- return errors.New("Service already started.")
71
+func NewService(ctx context.Context, h Handler) Service {
72
+ s := &service{
73
+ Handler: h,
74
+ Requests: RequestMap{},
75
+ Pipe: msg.NewPipe(10),
76
+ ContextCloser: ctxc.NewContextCloser(ctx, nil),
77
}
78
87
- // make a cancellable context.
88
- ctx, s.cancel = context.WithCancel(ctx)
89
-
90
- go s.handleIncomingMessages(ctx)
91
- return nil
92
-}
93
-
94
-// Stop stops Service activity.
95
-func (s *service) Stop() {
96
- s.cancel()
97
- s.cancel = context.CancelFunc(nil)
79
+ s.Children().Add(1)
80
+ go s.handleIncomingMessages()
81
+ return s
82
}
83
84
// GetPipe implements the mux.Protocol interface
@@ -132,6 +116,15 @@ func (s *service) SendMessage(ctx context.Context, m msg.NetMessage) error {
116
// SendRequest sends a request message out and awaits a response.
117
func (s *service) SendRequest(ctx context.Context, m msg.NetMessage) (msg.NetMessage, error) {
118
119
+ // check if we should bail given our contexts
120
+ select {
121
+ default:
122
+ case <-s.Closing():
123
+ return nil, fmt.Errorf("service closed: %s", s.Context().Err())
124
+ case <-ctx.Done():
125
+ return nil, ctx.Err()
126
+ }
127
+
128
// create a request
129
r, err := NewRequest(m.Peer().ID())
130
if err != nil {
@@ -153,6 +146,8 @@ func (s *service) SendRequest(ctx context.Context, m msg.NetMessage) (msg.NetMes
146
// check if we should bail after waiting for mutex
147
select {
148
default:
149
+ case <-s.Closing():
150
+ return nil, fmt.Errorf("service closed: %s", s.Context().Err())
151
case <-ctx.Done():
152
return nil, ctx.Err()
153
}
@@ -165,6 +160,8 @@ func (s *service) SendRequest(ctx context.Context, m msg.NetMessage) (msg.NetMes
160
err = nil
161
select {
162
case m = <-r.Response:
163
+ case <-s.Closed():
164
+ err = fmt.Errorf("service closed: %s", s.Context().Err())
165
case <-ctx.Done():
166
err = ctx.Err()
167
}
@@ -178,43 +175,50 @@ func (s *service) SendRequest(ctx context.Context, m msg.NetMessage) (msg.NetMes
175
176
// handleIncoming consumes the messages on the s.Incoming channel and
177
// routes them appropriately (to requests, or handler).
181
-func (s *service) handleIncomingMessages(ctx context.Context) {
178
+func (s *service) handleIncomingMessages() {
179
+ defer s.Children().Done()
180
+
181
for {
182
select {
183
case m, more := <-s.Incoming:
184
if !more {
185
return
186
}
188
- go s.handleIncomingMessage(ctx, m)
187
+ s.Children().Add(1)
188
+ go s.handleIncomingMessage(m)
189
190
- case <-ctx.Done():
190
+ case <-s.Closing():
191
return
192
}
193
}
194
}
195
196
-func (s *service) handleIncomingMessage(ctx context.Context, m msg.NetMessage) {
196
+func (s *service) handleIncomingMessage(m msg.NetMessage) {
197
+ defer s.Children().Done()
198
199
// unwrap the incoming message
200
data, rid, err := unwrapData(m.Data())
201
if err != nil {
201
- log.Errorf("de-serializing error: %v", err)
202
+ log.Errorf("service de-serializing error: %v", err)
203
+ return
204
}
205
+
206
m2 := msg.New(m.Peer(), data)
207
208
// if it's a request (or has no RequestID), handle it
209
if rid == nil || rid.IsRequest() {
207
- if s.Handler == nil {
210
+ handler := s.GetHandler()
211
+ if handler == nil {
212
log.Errorf("service dropped msg: %v", m)
213
return // no handler, drop it.
214
}
215
216
// should this be "go HandleMessage ... ?"
213
- r1 := s.Handler.HandleMessage(ctx, m2)
217
+ r1 := handler.HandleMessage(s.Context(), m2)
218
219
// if handler gave us a response, send it back out!
220
if r1 != nil {
217
- err := s.sendMessage(ctx, r1, rid.Response())
221
+ err := s.sendMessage(s.Context(), r1, rid.Response())
222
if err != nil {
223
log.Errorf("error sending response message: %v", err)
224
}
@@ -239,16 +243,20 @@ func (s *service) handleIncomingMessage(ctx context.Context, m msg.NetMessage) {
243
244
select {
245
case r.Response <- m2:
242
- case <-ctx.Done():
246
+ case <-s.Closing():
247
}
248
}
249
250
// SetHandler assigns the request Handler for this service.
251
func (s *service) SetHandler(h Handler) {
252
+ s.HandlerLock.Lock()
253
+ defer s.HandlerLock.Unlock()
254
s.Handler = h
255
}
256
257
// GetHandler returns the request Handler for this service.
258
func (s *service) GetHandler() Handler {
259
+ s.HandlerLock.RLock()
260
+ defer s.HandlerLock.RUnlock()
261
return s.Handler
262
}
net/service/service_test.go
+43
-25
@@ -38,13 +38,9 @@ func newPeer(t *testing.T, id string) peer.Peer {
38
func TestServiceHandler(t *testing.T) {
39
ctx := context.Background()
40
h := &ReverseHandler{}
41
- s := NewService(h)
41
+ s := NewService(ctx, h)
42
peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
43
44
- if err := s.Start(ctx); err != nil {
45
- t.Error(err)
46
- }
47
-
44
d, err := wrapData([]byte("beep"), nil)
45
if err != nil {
46
t.Error(err)
@@ -70,16 +66,8 @@ func TestServiceHandler(t *testing.T) {
66
67
func TestServiceRequest(t *testing.T) {
68
ctx := context.Background()
73
- s1 := NewService(&ReverseHandler{})
74
- s2 := NewService(&ReverseHandler{})
75
-
76
- if err := s1.Start(ctx); err != nil {
77
- t.Error(err)
78
- }
79
-
80
- if err := s2.Start(ctx); err != nil {
81
- t.Error(err)
82
- }
69
+ s1 := NewService(ctx, &ReverseHandler{})
70
+ s2 := NewService(ctx, &ReverseHandler{})
71
72
peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
73
@@ -110,18 +98,10 @@ func TestServiceRequest(t *testing.T) {
98
99
func TestServiceRequestTimeout(t *testing.T) {
100
ctx, _ := context.WithTimeout(context.Background(), time.Millisecond)
113
- s1 := NewService(&ReverseHandler{})
114
- s2 := NewService(&ReverseHandler{})
101
+ s1 := NewService(ctx, &ReverseHandler{})
102
+ s2 := NewService(ctx, &ReverseHandler{})
103
peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
104
117
- if err := s1.Start(ctx); err != nil {
118
- t.Error(err)
119
- }
120
-
121
- if err := s2.Start(ctx); err != nil {
122
- t.Error(err)
123
- }
124
-
105
// patch services together
106
go func() {
107
for {
@@ -143,3 +123,41 @@ func TestServiceRequestTimeout(t *testing.T) {
123
t.Error("should've timed out")
124
}
125
}
126
+
127
+func TestServiceClose(t *testing.T) {
128
+ ctx := context.Background()
129
+ s1 := NewService(ctx, &ReverseHandler{})
130
+ s2 := NewService(ctx, &ReverseHandler{})
131
+
132
+ peer1 := newPeer(t, "11140beec7b5ea3f0fdbc95d0dd47f3c5bc275aaaaaa")
133
+
134
+ // patch services together
135
+ go func() {
136
+ for {
137
+ select {
138
+ case m := <-s1.GetPipe().Outgoing:
139
+ s2.GetPipe().Incoming <- m
140
+ case m := <-s2.GetPipe().Outgoing:
141
+ s1.GetPipe().Incoming <- m
142
+ case <-ctx.Done():
143
+ return
144
+ }
145
+ }
146
+ }()
147
+
148
+ m1 := msg.New(peer1, []byte("beep"))
149
+ m2, err := s1.SendRequest(ctx, m1)
150
+ if err != nil {
151
+ t.Error(err)
152
+ }
153
+
154
+ if !bytes.Equal(m2.Data(), []byte("peeb")) {
155
+ t.Errorf("service handler data incorrect: %v != %v", m2.Data(), "oof")
156
+ }
157
+
158
+ s1.Close()
159
+ s2.Close()
160
+
161
+ <-s1.Closed()
162
+ <-s2.Closed()
163
+}
routing/dht/dht_test.go
+1
-5
@@ -23,11 +23,7 @@ import (
23
func setupDHT(ctx context.Context, t *testing.T, p peer.Peer) *IpfsDHT {
24
peerstore := peer.NewPeerstore()
25
26
- dhts := netservice.NewService(nil) // nil handler for now, need to patch it
27
- if err := dhts.Start(ctx); err != nil {
28
- t.Fatal(err)
29
- }
30
-
26
+ dhts := netservice.NewService(ctx, nil) // nil handler for now, need to patch it
27
net, err := inet.NewIpfsNetwork(ctx, p, peerstore, &mux.ProtocolMap{
28
mux.ProtocolID_Routing: dhts,
29
})