Merge pull request #132 from cloudstruct/feat/chainsync-friendly-interface

agaffney · web-flow · commit 2fa8b17c3525 · 2022-11-25T10:21:59.000-06:00
feat: make the chainsync interface friendlier
diff --git a/cmd/go-ouroboros-network/chainsync.go b/cmd/go-ouroboros-network/chainsync.go
@@ -13,11 +13,10 @@ import (
 )
 
 type chainSyncState struct {
-	oConn                 *ouroboros.Ouroboros
-	nodeToNode            bool
-	readyForNextBlockChan chan bool
-	byronEpochBaseSlot    uint64
-	byronEpochSlot        uint64
+	oConn              *ouroboros.Ouroboros
+	nodeToNode         bool
+	byronEpochBaseSlot uint64
+	byronEpochSlot     uint64
 }
 
 var syncState chainSyncState
@@ -78,12 +77,8 @@ var eraIntersect = map[int]map[string][]interface{}{
 
 func buildChainSyncConfig() chainsync.Config {
 	return chainsync.Config{
-		AwaitReplyFunc:        chainSyncAwaitReplyHandler,
-		RollBackwardFunc:      chainSyncRollBackwardHandler,
-		RollForwardFunc:       chainSyncRollForwardHandler,
-		IntersectFoundFunc:    chainSyncIntersectFoundHandler,
-		IntersectNotFoundFunc: chainSyncIntersectNotFoundHandler,
-		DoneFunc:              chainSyncDoneHandler,
+		RollBackwardFunc: chainSyncRollBackwardHandler,
+		RollForwardFunc:  chainSyncRollForwardHandler,
 	}
 }
 
@@ -132,56 +127,36 @@ func testChainSync(f *globalFlags) {
 		os.Exit(1)
 	}
 	o.ChainSync.Client.Start()
-	o.BlockFetch.Client.Start()
+	if f.ntnProto {
+		o.BlockFetch.Client.Start()
+	}
 
 	syncState.oConn = o
-	syncState.readyForNextBlockChan = make(chan bool)
 	syncState.nodeToNode = f.ntnProto
 	var point chainsync.Point
 	if len(eraIntersect[f.networkMagic][chainSyncFlags.startEra]) > 0 {
 		// Slot
-		point.Slot = uint64(eraIntersect[f.networkMagic][chainSyncFlags.startEra][0].(int))
+		slot := uint64(eraIntersect[f.networkMagic][chainSyncFlags.startEra][0].(int))
 		// Block hash
 		hash, _ := hex.DecodeString(eraIntersect[f.networkMagic][chainSyncFlags.startEra][1].(string))
-		point.Hash = hash
+		point = chainsync.NewPoint(slot, hash)
+	} else {
+		point = chainsync.NewPointOrigin()
 	}
-	if err := o.ChainSync.Client.FindIntersect([]chainsync.Point{point}); err != nil {
-		fmt.Printf("ERROR: FindIntersect: %s\n", err)
+	if err := o.ChainSync.Client.Sync([]chainsync.Point{point}); err != nil {
+		fmt.Printf("ERROR: failed to start chain-sync: %s\n", err)
 		os.Exit(1)
 	}
-	// Wait until ready for next block
-	<-syncState.readyForNextBlockChan
-	// Pipeline the initial block requests to speed things up a bit
-	// Using a value higher than 10 seems to cause problems with NtN
-	for i := 0; i < 10; i++ {
-		err := o.ChainSync.Client.RequestNext()
-		if err != nil {
-			fmt.Printf("ERROR: RequestNext: %s\n", err)
-			os.Exit(1)
-		}
-	}
-	for {
-		err := o.ChainSync.Client.RequestNext()
-		if err != nil {
-			fmt.Printf("ERROR: RequestNext: %s\n", err)
-			os.Exit(1)
-		}
-		// Wait until ready for next block
-		<-syncState.readyForNextBlockChan
-	}
-}
-
-func chainSyncAwaitReplyHandler() error {
-	return nil
+	// Wait forever...the rest of the sync operations are async
+	select {}
 }
 
-func chainSyncRollBackwardHandler(point interface{}, tip interface{}) error {
+func chainSyncRollBackwardHandler(point chainsync.Point, tip chainsync.Tip) error {
 	fmt.Printf("roll backward: point = %#v, tip = %#v\n", point, tip)
-	syncState.readyForNextBlockChan <- true
 	return nil
 }
 
-func chainSyncRollForwardHandler(blockType uint, blockData interface{}) error {
+func chainSyncRollForwardHandler(blockType uint, blockData interface{}, tip chainsync.Tip) error {
 	if syncState.nodeToNode {
 		var blockSlot uint64
 		var blockHash []byte
@@ -241,23 +216,6 @@ func chainSyncRollForwardHandler(blockType uint, blockData interface{}) error {
 			fmt.Printf("%s\n", utils.DumpCborStructure(blockData, ""))
 		}
 	}
-	syncState.readyForNextBlockChan <- true
-	return nil
-}
-
-func chainSyncIntersectFoundHandler(point interface{}, tip interface{}) error {
-	fmt.Printf("found intersect: point = %#v, tip = %#v\n", point, tip)
-	syncState.readyForNextBlockChan <- true
-	return nil
-}
-
-func chainSyncIntersectNotFoundHandler(tip interface{}) error {
-	fmt.Printf("ERROR: failed to find intersection\n")
-	os.Exit(1)
-	return nil
-}
-
-func chainSyncDoneHandler() error {
 	return nil
 }
 
diff --git a/protocol/chainsync/chainsync.go b/protocol/chainsync/chainsync.go
@@ -90,21 +90,13 @@ type ChainSync struct {
 }
 
 type Config struct {
-	AwaitReplyFunc        AwaitReplyFunc
-	RollBackwardFunc      RollBackwardFunc
-	RollForwardFunc       RollForwardFunc
-	IntersectFoundFunc    IntersectFoundFunc
-	IntersectNotFoundFunc IntersectNotFoundFunc
-	DoneFunc              DoneFunc
+	RollBackwardFunc RollBackwardFunc
+	RollForwardFunc  RollForwardFunc
 }
 
 // Callback function types
-type AwaitReplyFunc func() error
-type RollBackwardFunc func(interface{}, interface{}) error
-type RollForwardFunc func(uint, interface{}) error
-type IntersectFoundFunc func(interface{}, interface{}) error
-type IntersectNotFoundFunc func(interface{}) error
-type DoneFunc func() error
+type RollBackwardFunc func(Point, Tip) error
+type RollForwardFunc func(uint, interface{}, Tip) error
 
 func New(protoOptions protocol.ProtocolOptions, cfg *Config) *ChainSync {
 	c := &ChainSync{
diff --git a/protocol/chainsync/client.go b/protocol/chainsync/client.go
@@ -4,11 +4,15 @@ import (
 	"fmt"
 	"github.com/cloudstruct/go-cardano-ledger"
 	"github.com/cloudstruct/go-ouroboros-network/protocol"
+	"sync"
 )
 
 type Client struct {
 	*protocol.Protocol
-	config *Config
+	config                *Config
+	busyMutex             sync.Mutex
+	intersectResultChan   chan error
+	readyForNextBlockChan chan bool
 }
 
 func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
@@ -21,7 +25,9 @@ func NewClient(protoOptions protocol.ProtocolOptions, cfg *Config) *Client {
 		msgFromCborFunc = NewMsgFromCborNtN
 	}
 	c := &Client{
-		config: cfg,
+		config:                cfg,
+		intersectResultChan:   make(chan error),
+		readyForNextBlockChan: make(chan bool),
 	}
 	protoConfig := protocol.ProtocolConfig{
 		Name:                PROTOCOL_NAME,
@@ -52,36 +58,73 @@ func (c *Client) messageHandler(msg protocol.Message, isResponse bool) error {
 		err = c.handleIntersectFound(msg)
 	case MESSAGE_TYPE_INTERSECT_NOT_FOUND:
 		err = c.handleIntersectNotFound(msg)
-	case MESSAGE_TYPE_DONE:
-		err = c.handleDone()
 	default:
 		err = fmt.Errorf("%s: received unexpected message type %d", PROTOCOL_NAME, msg.Type())
 	}
 	return err
 }
 
-func (c *Client) RequestNext() error {
-	msg := NewMsgRequestNext()
-	return c.SendMessage(msg)
+func (c *Client) Stop() error {
+	c.busyMutex.Lock()
+	defer c.busyMutex.Unlock()
+	msg := NewMsgDone()
+	if err := c.SendMessage(msg); err != nil {
+		return err
+	}
+	return nil
 }
 
-func (c *Client) FindIntersect(points []Point) error {
-	msg := NewMsgFindIntersect(points)
-	return c.SendMessage(msg)
+func (c *Client) Sync(intersectPoints []Point) error {
+	c.busyMutex.Lock()
+	defer c.busyMutex.Unlock()
+	msg := NewMsgFindIntersect(intersectPoints)
+	if err := c.SendMessage(msg); err != nil {
+		return err
+	}
+	if err := <-c.intersectResultChan; err != nil {
+		return err
+	}
+	// Pipeline the initial block requests to speed things up a bit
+	// Using a value higher than 10 seems to cause problems with NtN
+	for i := 0; i < 10; i++ {
+		msg := NewMsgRequestNext()
+		if err := c.SendMessage(msg); err != nil {
+			return err
+		}
+	}
+	go c.syncLoop()
+	return nil
 }
 
-func (c *Client) handleAwaitReply() error {
-	if c.config.AwaitReplyFunc == nil {
-		return fmt.Errorf("received chain-sync AwaitReply message but no callback function is defined")
+func (c *Client) syncLoop() {
+	for {
+		// Wait for a block to be received
+		<-c.readyForNextBlockChan
+		c.busyMutex.Lock()
+		// Request the next block
+		// In practice we already have multiple block requests pipelined
+		// and this just adds another one to the pile
+		msg := NewMsgRequestNext()
+		if err := c.SendMessage(msg); err != nil {
+			c.SendError(err)
+			return
+		}
+		c.busyMutex.Unlock()
 	}
-	// Call the user callback function
-	return c.config.AwaitReplyFunc()
+}
+
+func (c *Client) handleAwaitReply() error {
+	return nil
 }
 
 func (c *Client) handleRollForward(msgGeneric protocol.Message) error {
 	if c.config.RollForwardFunc == nil {
 		return fmt.Errorf("received chain-sync RollForward message but no callback function is defined")
 	}
+	// Signal that we're ready for the next block after we finish handling this one
+	defer func() {
+		c.readyForNextBlockChan <- true
+	}()
 	if c.Mode() == protocol.ProtocolModeNodeToNode {
 		msg := msgGeneric.(*MsgRollForwardNtN)
 		var blockHeader interface{}
@@ -112,15 +155,15 @@ func (c *Client) handleRollForward(msgGeneric protocol.Message) error {
 			}
 		}
 		// Call the user callback function
-		return c.config.RollForwardFunc(blockType, blockHeader)
+		return c.config.RollForwardFunc(blockType, blockHeader, msg.Tip)
 	} else {
 		msg := msgGeneric.(*MsgRollForwardNtC)
 		blk, err := ledger.NewBlockFromCbor(msg.BlockType(), msg.BlockCbor())
 		if err != nil {
 			return err
 		}
 		// Call the user callback function
-		return c.config.RollForwardFunc(msg.BlockType(), blk)
+		return c.config.RollForwardFunc(msg.BlockType(), blk, msg.Tip)
 	}
 }
 
@@ -134,27 +177,11 @@ func (c *Client) handleRollBackward(msgGeneric protocol.Message) error {
 }
 
 func (c *Client) handleIntersectFound(msgGeneric protocol.Message) error {
-	if c.config.IntersectFoundFunc == nil {
-		return fmt.Errorf("received chain-sync IntersectFound message but no callback function is defined")
-	}
-	msg := msgGeneric.(*MsgIntersectFound)
-	// Call the user callback function
-	return c.config.IntersectFoundFunc(msg.Point, msg.Tip)
+	c.intersectResultChan <- nil
+	return nil
 }
 
 func (c *Client) handleIntersectNotFound(msgGeneric protocol.Message) error {
-	if c.config.IntersectNotFoundFunc == nil {
-		return fmt.Errorf("received chain-sync IntersectNotFound message but no callback function is defined")
-	}
-	msg := msgGeneric.(*MsgIntersectNotFound)
-	// Call the user callback function
-	return c.config.IntersectNotFoundFunc(msg.Tip)
-}
-
-func (c *Client) handleDone() error {
-	if c.config.DoneFunc == nil {
-		return fmt.Errorf("received chain-sync Done message but no callback function is defined")
-	}
-	// Call the user callback function
-	return c.config.DoneFunc()
+	c.intersectResultChan <- IntersectNotFoundError{}
+	return nil
 }
diff --git a/protocol/chainsync/error.go b/protocol/chainsync/error.go
@@ -0,0 +1,8 @@
+package chainsync
+
+type IntersectNotFoundError struct {
+}
+
+func (e IntersectNotFoundError) Error() string {
+	return "chain intersection not found"
+}
diff --git a/protocol/chainsync/messages.go b/protocol/chainsync/messages.go
@@ -228,6 +228,17 @@ type Point struct {
 	Hash []byte
 }
 
+func NewPoint(slot uint64, blockHash []byte) Point {
+	return Point{
+		Slot: slot,
+		Hash: blockHash,
+	}
+}
+
+func NewPointOrigin() Point {
+	return Point{}
+}
+
 // A "point" can sometimes be empty, but the CBOR library gets grumpy about this
 // when doing automatic decoding from an array, so we have to handle this case specially
 func (p *Point) UnmarshalCBOR(data []byte) error {
diff --git a/protocol/chainsync/server.go b/protocol/chainsync/server.go
@@ -40,10 +40,29 @@ func NewServer(protoOptions protocol.ProtocolOptions, cfg *Config) *Server {
 
 func (s *Server) messageHandler(msg protocol.Message, isResponse bool) error {
 	var err error
-	// TODO: add cases for messages from client
 	switch msg.Type() {
+	case MESSAGE_TYPE_REQUEST_NEXT:
+		err = s.handleRequestNext(msg)
+	case MESSAGE_TYPE_FIND_INTERSECT:
+		err = s.handleFindIntersect(msg)
+	case MESSAGE_TYPE_DONE:
+		err = s.handleDone()
 	default:
 		err = fmt.Errorf("%s: received unexpected message type %d", PROTOCOL_NAME, msg.Type())
 	}
 	return err
 }
+
+func (s *Server) handleRequestNext(msg protocol.Message) error {
+	// TODO
+	return nil
+}
+
+func (s *Server) handleFindIntersect(msg protocol.Message) error {
+	// TODO
+	return nil
+}
+
+func (s *Server) handleDone() error {
+	return nil
+}
