.

go/neo/client.go

@@ -113,19 +113,41 @@ func NewClient(clusterName, masterAddr string, net xnet.Networker) *Client {
 
 // Run starts client node and runs it until either ctx is canceled or master
 // commands it to shutdown. (TODO verify M->shutdown)
-func (cli *Client) Run(ctx context.Context) (err error) {
+func (c *Client) Run(ctx context.Context) (err error) {
 	defer task.Running(&ctx, "client")(&err)
 
 	// run process which performs master talk
 	ctx, cancel := context.WithCancel(ctx)
-	cli.runCancel = cancel
-//	cli.node.OnShutdown = cancel // XXX ok?
-//	return cli.talkMaster(ctx)
-	cli.runWG = xsync.NewWorkGroup(ctx) // XXX create it in NewClient ? (Close also uses it)
-	cli.runWG.Go(cli.node.talkMaster)
-	cli.runWG.Go(cli.recvMaster)
-
-	return cli.runWG.Wait()
+	c.runCancel = cancel
+//	c.node.OnShutdown = cancel // XXX ok?
+//	return c.talkMaster(ctx)
+//	c.runWG = xsync.NewWorkGroup(ctx) // XXX create it in NewClient ? (Close also uses it)
+//	c.runWG.Go(c.node.talkMaster)
+//	c.runWG.Go(c.recvMaster)
+
+	return c.node.TalkMaster(ctx, func(ctx context.Context, mlink *neonet.NodeLink) error {
+		// XXX errctx  ("on redial"? "connected"?)
+
+		c.head0 = c.head
+
+		wg := xsync.NewWorkGroup(ctx)
+
+		// launch master notifications receiver
+		wg.Go(func(ctx context.Context) error {
+			return c.recvMaster(ctx)
+		})
+
+		// init lastTid from master
+		// TODO better change protocol for master to send us head via notify
+		// channel right after identification.
+		wg.Go(func(ctx context.Context) error {
+			return c.initFromMaster(ctx, mlink)
+		})
+
+		return wg.Wait()
+	})
+
+//	return c.runWG.Wait()
 }
 
 // Close implements zodb.IStorageDriver.
@@ -368,16 +390,7 @@ func (c *Client) recvMaster(ctx context.Context) (err error) {
 	for {
 		req, err := c.node.RecvM1()
 		if err != nil {
-			switch {
-			// reconnected to master
-			case errors.Is(err, eventMasterReconnect):
-				c.head0 = c.head
-				// XXX -> initFromMaster
-
-			default:
-				return err
-			}
-			continue
+			return err
 		}
 
 		err = c.recvMaster1(req.Msg)

go/neo/mastered.go

@@ -29,6 +29,7 @@ import (
 
 	"lab.nexedi.com/kirr/go123/xerr"
 	"lab.nexedi.com/kirr/go123/xnet"
+	"lab.nexedi.com/kirr/go123/xsync"
 
 	"lab.nexedi.com/kirr/neo/go/internal/log"
 	"lab.nexedi.com/kirr/neo/go/internal/task"
@@ -132,17 +133,19 @@ func newMasteredNode(typ proto.NodeType, clusterName string, net xnet.Networker,
 
 
 
-// talkMaster dials master, identifies to it, and receives master notifications and requests.
+// TalkMaster dials master, identifies to it, and receives master notifications and requests.
 //
 // Notifications to node/partition tables and cluster state are automatically
 // handled, while other notifications and requests are passed through to RecvM1.
 //
 // The connection to master is persisted by redial as needed.
-func (node *_MasteredNode) talkMaster(ctx context.Context) (err error) {
+//
+// f is called on every reconnection after identification and protocol prologue. XXX
+func (node *_MasteredNode) TalkMaster(ctx context.Context, f func(context.Context, *neonet.NodeLink) error) (err error) {
 	defer task.Runningf(&ctx, "talk master(%s)", node.MasterAddr)(&err)
 
 	for {
-		err := node.talkMaster1(ctx)
+		err := node.talkMaster1(ctx, f)
 		log.Warning(ctx, err)	// XXX Warning ok? -> Error?
 		// TODO if err == "reject identification / protocol error" -> shutdown client
 		// TODO if err == shutdown -> return
@@ -160,7 +163,7 @@ func (node *_MasteredNode) talkMaster(ctx context.Context) (err error) {
 	}
 }
 
-func (node *_MasteredNode) talkMaster1(ctx context.Context) error {
+func (node *_MasteredNode) talkMaster1(ctx context.Context, f func(context.Context, *neonet.NodeLink) error) error {
 	reqID := &proto.RequestIdentification{
 		NodeType:	node.myInfo.Type,
 		NID:		node.myInfo.NID,
@@ -220,28 +223,27 @@ func (node *_MasteredNode) talkMaster1(ctx context.Context) error {
 
 		// XXX update .masterLink + notify waiters
 
-		// RecvM1 <- eventReconnect
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-
-		case node.rxm <- _RxM{Err: eventMasterReconnect}:
-			// ok
-		}
-
-
+		wg := xsync.NewWorkGroup(ctx)
 		// receive and handle notifications from master
-		defer task.Running(&ctx, "rx")(&err)
-		for {
-			req, err := mlink.Recv1()
-			if err != nil {
-				return err
+		wg.Go(func(ctx context.Context) error {
+			defer task.Running(&ctx, "rx")(&err)
+			for {
+				req, err := mlink.Recv1()
+				if err != nil {
+					return err
+				}
+				err = node.recvMaster1(ctx, req) // req ownership is passed in
+				if err != nil {
+					return err
+				}
 			}
-			err = node.recvMaster1(ctx, req) // req ownership is passed in
-			if err != nil {
-				return err
-			}
-		}
+		})
+		// run user code
+		wg.Go(func(ctx context.Context) error {
+			return f(ctx, mlink)
+		})
+
+		return wg.Wait()
 	})
 
 }