.

go/neo/master.go

@@ -120,8 +120,10 @@ type _MasteredPeer struct {
 	accept *proto.AcceptIdentification // identify decided to accept this peer with .accept
 
 	// all tasks are spawned under wg. If any task fails - whole wg is canceled.
-	wg *xsync.WorkGroup
-	// XXX +cancel
+	// cancel signals all tasks under wg to stop.
+	wg     *xsync.WorkGroup
+	cancel func()
+
 	// snapshot of nodeTab/partTab/stateCode when peer was accepted by main.
 	state0 *xneo.ClusterStateSnapshot
 	// main -> peerWG.notify  δnodeTab/δpartTab/δstateCode.
@@ -1049,19 +1051,18 @@ func (m *Master) updateNodeTab(ctx context.Context, nodeInfo proto.NodeInfo) *xn
 
 	event := &_ΔNodeTab{nodeInfo}
 	// XXX locking
-//	for nid, ch := range m.notifyTab {
-	for nid, w := range m.peerWorkTab {
+	for nid, p := range m.peerTab {
 		// TODO change limiting by buffer size to limiting by time -
 		// - i.e. detach peer if event queue grows more than 30s of time.
 		select {
-		case w.notifyq <- event:
+		case p.notifyq <- event:
 			continue // ok
 		default:
 		}
 
 		log.Warningf(ctx, "peer %s is slow -> detaching it", nid)
-		close(w.notifyqOverflow)
-		// TODO delete(m.peerWorkTab, nid)
+		close(p.notifyqOverflow)
+		// TODO delete(m.peerTab, nid) -> p.cancel()
 		// XXX what else?
 		panic("TODO")
 	}
@@ -1261,26 +1262,31 @@ func (m *Master) identify(ctx context.Context, n nodeCome) (peer *_MasteredNode,
 	node = m.updateNodeTab(ctx, nodeInfo)
 	node.SetLink(n.req.Link())
 
-	// make nodeTab/partTab snapshot to push to accepted node and subscribe it for updates
-	m.peerWorkTab[node.NID] = &_MasteredPeer{
-		peer:   node,
+
+	// create peer with nodeTab/partTab snapshot to push to accepted node
+	// and subscribe it for updates.
+	peerCtx, peerCancel := context.WithCancel(m.runCtx)
+	peer = &_MasteredPeer{
+		node:   node,
 		accept: accept,
-		wg:     xsync.NewWorkGroup(m.runCtx), // XXX wrong -> per peer ctx (derived from runCtx)
+		wg:     xsync.NewWorkGroup(peerCtx),
+		cancel: peerCancel,
 		state0: m.node.State.Snapshot(),
 		// TODO change limiting by buffer size -> to limiting by time
 		//      (see updateNodeTab for details)
 		notifyq:         make(chan _ΔClusterState, 1024),
 		notifyqOverflow: make(chan struct{}),
 	}
+	m.peerTab[node.NID] = peer
 
 	return peer, true
 }
 
 // accept sends acceptance to just identified peer, sends nodeTab and partTab
 // and spawns task to proxy their updates to the peer. XXX
-func (m *Master) accept(p *_MasteredPeer, idReq *neonet.Request, idResp proto.Msg) error {
+func (m *Master) accept(p *_MasteredPeer, idReq *neonet.Request) error {
 	// XXX errctx?
-	err := idReq.Reply(idResp)
+	err := idReq.Reply(p.accept)
 	if err != nil {
 		return fmt.Errorf("send accept: %w", err)
 	}
@@ -1305,8 +1311,8 @@ func (m *Master) accept(p *_MasteredPeer, idReq *neonet.Request, idResp proto.Ms
 
 	// XXX send clusterState too? (NEO/py does not send it)
 
-	// spawn p.notify to proxy δnodeTab/δpartTab/δcluterState to peer
-	p.wg.Go(p.notify)
+	p.wg.Go(p.notify)      // main -> peer δnodeTab/δpartTab/δcluterState to proxy to peer link
+	m.mainWG.Go(p.waitAll) // main <- peer "peer (should be) disconnected"
 	return nil
 }