Merge pull request ethereum#43 from rus-alex/fix/api-graceful-stop

Fix of RPC graceful stop

rpc/handler.go

@@ -228,9 +228,12 @@ func (h *handler) cancelServerSubscriptions(err error) {
 // startCallProc runs fn in a new goroutine and starts tracking it in the h.calls wait group.
 func (h *handler) startCallProc(fn func(*callProc)) {
 	h.callWG.Add(1)
+	h.reg.callWG.Add(1)
 	go func() {
-		ctx, cancel := context.WithCancel(h.rootCtx)
 		defer h.callWG.Done()
+		defer h.reg.callWG.Done()
+
+		ctx, cancel := context.WithCancel(h.rootCtx)
 		defer cancel()
 		fn(&callProc{ctx: ctx})
 	}()

rpc/server.go

@@ -119,13 +119,15 @@ func (s *Server) serveSingleRequest(ctx context.Context, codec ServerCodec) {
 // requests to finish, then closes all codecs which will cancel pending requests and
 // subscriptions.
 func (s *Server) Stop() {
-	if atomic.CompareAndSwapInt32(&s.run, 1, 0) {
-		log.Debug("RPC server shutting down")
-		s.codecs.Each(func(c interface{}) bool {
-			c.(ServerCodec).close()
-			return true
-		})
+	if !atomic.CompareAndSwapInt32(&s.run, 1, 0) {
+		return
 	}
+	log.Debug("RPC server shutting down")
+	s.codecs.Each(func(c interface{}) bool {
+		c.(ServerCodec).close()
+		return true
+	})
+	s.services.callWG.Wait()
 }
 
 // RPCService gives meta information about the server.

rpc/service.go

@@ -38,6 +38,7 @@ var (
 
 type serviceRegistry struct {
 	mu       sync.Mutex
+	callWG   sync.WaitGroup
 	services map[string]service
 }