Reception stubbed out to receive messages thread. Needs additional definition before it's complete.

context/params/network.go

+package params
+
+import (
+	"time"
+)
+
+type Network struct {
+	TrackNetworkPeriod  time.Duration
+	NumWorkers          int
+	MaxHistoricalRounds int
+}
+
+func GetDefaultNetwork() Network {
+	return Network{
+		TrackNetworkPeriod:  100 * time.Millisecond,
+		NumWorkers:          4,
+		MaxHistoricalRounds: 100,
+	}
+}

context/stoppable/single.go

@@ -32,7 +32,7 @@ func (s *Single) IsRunning() bool {
 }
 
 // Quit returns the read only channel it will send the stop signal on.
-func (s *Single) Quit() chan<- struct{} {
+func (s *Single) Quit() <-chan struct{} {
 	return s.quit
 }
 

go.mod

@@ -21,7 +21,7 @@ require (
 	gitlab.com/elixxir/primitives v0.0.0-20200915190719-f4586ec93f50
 	gitlab.com/xx_network/comms v0.0.0-20200915154643-d533291041b7
 	gitlab.com/xx_network/crypto v0.0.0-20200812183430-c77a5281c686
-	gitlab.com/xx_network/primitives v0.0.0-20200812183720-516a65a4a9b2
+	gitlab.com/xx_network/primitives v0.0.0-20200915204206-eb0287ed0031
 	golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a
 	golang.org/x/sys v0.0.0-20200828194041-157a740278f4 // indirect
 	google.golang.org/protobuf v1.25.0

go.sum

@@ -264,6 +264,8 @@ gitlab.com/xx_network/primitives v0.0.0-20200804183002-f99f7a7284da h1:CCVslUwNC
 gitlab.com/xx_network/primitives v0.0.0-20200804183002-f99f7a7284da/go.mod h1:OK9xevzWCaPO7b1wiluVJGk7R5ZsuC7pHY5hteZFQug=
 gitlab.com/xx_network/primitives v0.0.0-20200812183720-516a65a4a9b2 h1:Jvv2fLk+2ULDCXiVTPI8Jlg2fBrmq2NSA+dlsN5t2Pk=
 gitlab.com/xx_network/primitives v0.0.0-20200812183720-516a65a4a9b2/go.mod h1:OK9xevzWCaPO7b1wiluVJGk7R5ZsuC7pHY5hteZFQug=
+gitlab.com/xx_network/primitives v0.0.0-20200915204206-eb0287ed0031 h1:QU+UGZPdQXUsKtOfR7hLxjg3/+MI1d+NqXgREQxzOe4=
+gitlab.com/xx_network/primitives v0.0.0-20200915204206-eb0287ed0031/go.mod h1:wtdCMr7DPePz9qwctNoAUzZtbOSHSedcK++3Df3psjA=
 gitlab.com/xx_network/ring v0.0.2 h1:TlPjlbFdhtJrwvRgIg4ScdngMTaynx/ByHBRZiXCoL0=
 gitlab.com/xx_network/ring v0.0.2/go.mod h1:aLzpP2TiZTQut/PVHR40EJAomzugDdHXetbieRClXIM=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=

network/manager.go

@@ -12,6 +12,7 @@ package network
 import (
 	"github.com/pkg/errors"
 	"gitlab.com/elixxir/client/context"
+	"gitlab.com/elixxir/client/context/params"
 	"gitlab.com/elixxir/client/context/stoppable"
 	"gitlab.com/elixxir/client/network/health"
 	"gitlab.com/elixxir/comms/client"
@@ -19,6 +20,7 @@ import (
 	"gitlab.com/xx_network/crypto/signature/rsa"
 	"gitlab.com/xx_network/primitives/id"
 	//	"gitlab.com/xx_network/primitives/ndf"
+	pb "gitlab.com/elixxir/comms/mixmessages"
 	"time"
 )
 
@@ -43,6 +45,11 @@ type Manager struct {
 
 	//channels
 	nodeRegistration chan network.NodeGateway
+	roundUpdate      chan *pb.RoundInfo
+	historicalLookup chan id.Round
+
+	// Processing rounds
+	Processing *ProcessingRounds
 
 	//local pointer to user ID because it is used often
 	uid *id.ID
@@ -74,6 +81,8 @@ func NewManager(ctx *context.Context) (*Manager, error) {
 			" client network manager")
 	}
 
+	opts := params.GetDefaultNetwork()
+
 	cm := &Manager{
 		Comms:            comms,
 		Context:          ctx,
@@ -81,6 +90,11 @@ func NewManager(ctx *context.Context) (*Manager, error) {
 		health:           health.Init(ctx, 5*time.Second),
 		instance:         instance,
 		uid:              cryptoUser.GetUserID(),
+		Processing:       NewProcessingRounds(),
+		roundUpdate:      make(chan *pb.RoundInfo, opts.NumWorkers),
+		historicalLookup: make(chan id.Round, opts.NumWorkers),
+		nodeRegistration: make(chan network.NodeGateway,
+			opts.NumWorkers),
 	}
 
 	return cm, nil
@@ -111,12 +125,12 @@ func (m *Manager) StartRunners() error {
 	// Start the Network Tracker
 	m.runners.Add(StartTrackNetwork(m.Context, m))
 	// Message reception
-	m.runners.Add(StartMessageReceivers(m.Context))
+	m.runners.Add(StartMessageReceivers(m.Context, m))
 	// Node Updates
-	m.runners.Add(StartNodeKeyExchange(m.Context)) // Adding/Keys
+	m.runners.Add(StartNodeKeyExchange(m.Context, m)) // Adding/Keys
 	m.runners.Add(StartNodeRemover(m.Context))        // Removing
 	// Round history processing
-	m.runners.Add(StartProcessHistoricalRounds(m.Context))
+	m.runners.Add(StartProcessHistoricalRounds(m.Context, m))
 	// health tracker
 	m.health.Start()
 	m.runners.Add(m.health)
@@ -146,3 +160,19 @@ func (m *Manager) GetHealthTracker() context.HealthTracker {
 func (m *Manager) GetInstance() *network.Instance {
 	return m.instance
 }
+
+// GetNodeRegistrationCh returns node registration channel for node
+// events.
+func (m *Manager) GetNodeRegistrationCh() chan network.NodeGateway {
+	return m.nodeRegistration
+}
+
+// GetRoundUpdateCh returns the network managers round update channel
+func (m *Manager) GetRoundUpdateCh() chan *pb.RoundInfo {
+	return m.roundUpdate
+}
+
+// GetHistoricalLookupCh returns the historical round lookup channel
+func (m *Manager) GetHistoricalLookupCh() chan id.Round {
+	return m.historicalLookup
+}

network/nodes.go

@@ -20,16 +20,20 @@ import (
 	"gitlab.com/xx_network/primitives/id"
 	//	"sync"
 	//	"time"
+	"fmt"
+	jww "github.com/spf13/jwalterweatherman"
 )
 
 // StartNodeKeyExchange kicks off a worker pool of node key exchange routines
-func StartNodeKeyExchange(ctx *context.Context) stoppable.Stoppable {
+func StartNodeKeyExchange(ctx *context.Context,
+	network *Manager) stoppable.Stoppable {
 	stoppers := stoppable.NewMulti("NodeKeyExchangers")
-	numWorkers := params.GetDefaultNodeKeys().WorkerPoolSize
-	keyCh := make(chan network.NodeGateway, numWorkers)
+	numWorkers := params.GetDefaultNetwork().NumWorkers
+	keyCh := network.GetNodeRegistrationCh()
 	ctx.Manager.GetInstance().SetAddNodeChan(keyCh)
-	for i := 0; i < ctx.GetNumNodeKeyExchangers(); i++ {
-		stopper := stoppable.NewSingle("NodeKeyExchange" + i)
+	for i := 0; i < numWorkers; i++ {
+		stopper := stoppable.NewSingle(
+			fmt.Sprintf("NodeKeyExchange%d", i))
 		go ExchangeNodeKeys(ctx, keyCh, stopper.Quit())
 		stoppers.Add(stopper)
 	}
@@ -60,9 +64,10 @@ func StartNodeRemover(ctx *context.Context) stoppable.Stoppable {
 	numWorkers := params.GetDefaultNodeKeys().WorkerPoolSize
 	remCh := make(chan *id.ID, numWorkers)
 	ctx.Manager.GetInstance().SetRemoveNodeChan(remCh)
-	for i := 0; i < ctx.GetNumNodeRemovers(); i++ {
-		stopper := stoppable.NewSingle("RemoveNode" + i)
-		go RemoveNode(ctx, remCh, quitCh)
+	for i := uint(0); i < numWorkers; i++ {
+		stopper := stoppable.NewSingle(
+			fmt.Sprintf("RemoveNode%d", i))
+		go RemoveNode(ctx, remCh, stopper.Quit())
 		stoppers.Add(stopper)
 	}
 	return stoppers

network/processingrounds.go

@@ -14,6 +14,13 @@ type ProcessingRounds struct {
 	sync.RWMutex
 }
 
+// NewProcessingRounds returns a processing rounds object
+func NewProcessingRounds() *ProcessingRounds {
+	return &ProcessingRounds{
+		rounds: make(map[id.Round]struct{}),
+	}
+}
+
 // Add a round to the list of processing rounds
 func (pr *ProcessingRounds) Add(id id.Round) {
 	pr.Lock()

network/receive.go

@@ -7,56 +7,61 @@
 package network
 
 import (
+	"fmt"
 	"gitlab.com/elixxir/client/context"
+	"gitlab.com/elixxir/client/context/params"
 	"gitlab.com/elixxir/client/context/stoppable"
+	pb "gitlab.com/elixxir/comms/mixmessages"
 	"gitlab.com/elixxir/primitives/format"
+	//jww "github.com/spf13/jwalterweatherman"
 )
 
-// ReceiveMessage is called by a MessageReceiver routine whenever a new CMIX
-// message is available.
-func ReceiveMessage(ctx *context.Context, m *format.Message) {
-	// decrypted, err := decrypt(ctx, m) // Returns MessagePart
-	// if err != nil {
-	// 	// Add to error/garbled messages list
-	// 	jww.WARN.Errorf("Could not decode message: %+v", err)
-	// 	ctx.GetGarbledMesssages().Add(m)
-	// }
-
-	// // Reconstruct the partitioned message
-	// completeMsg := constructMessageFromPartition(ctx, decrypted) // Returns ClientMessage
-	// if completeMsg != nil {
-	// 	ctx.GetSwitchBoard().Say(completeMsg)
-	// }
+// ReceiveMessages is called by a MessageReceiver routine whenever new CMIX
+// messages are available at a gateway.
+func ReceiveMessages(ctx *context.Context, roundInfo *pb.RoundInfo) {
+	msgs := getMessagesFromGateway(ctx, roundInfo)
+	for _, m := range msgs {
+		receiveMessage(ctx, m)
+	}
 }
 
 // StartMessageReceivers starts a worker pool of message receivers, which listen
-// on a channel for messages and run them through processing.
-func StartMessageReceivers(ctx *context.Context) stoppable.Stoppable {
-	// We assume receivers channel is set up elsewhere, but note that this
-	// would also be a reasonable place under assumption of 1 call to
-	// message receivers (would also make sense to .Close it instead of
-	// using quit channel, which somewhat simplifies for loop later.
+// on a channel for rounds in which to check for messages and run them through
+// processing.
+func StartMessageReceivers(ctx *context.Context,
+	network *Manager) stoppable.Stoppable {
 	stoppers := stoppable.NewMulti("MessageReceivers")
-	// receiverCh := ctx.GetNetwork().GetMessageReceiverCh()
-	// for i := 0; i < ctx.GetNumReceivers(); i++ {
-	// 	stopper := stoppable.NewSingle("MessageReceiver" + i)
-	// 	go MessageReceiver(ctx, messagesCh, stopper.Quit())
-	// 	stoppers.Add(stopper)
-	// }
+	opts := params.GetDefaultNetwork()
+	receiverCh := network.GetRoundUpdateCh()
+	for i := 0; i < opts.NumWorkers; i++ {
+		stopper := stoppable.NewSingle(
+			fmt.Sprintf("MessageReceiver%d", i))
+		go MessageReceiver(ctx, receiverCh, stopper.Quit())
+		stoppers.Add(stopper)
+	}
 	return stoppers
 }
 
-// MessageReceiver waits until quit signal or there is a message
-// available on the messages channel.
-func MessageReceiver(ctx *context.Context, messagesCh chan format.Message,
+// MessageReceiver waits until quit signal or there is a round available
+// for which to check for messages available on the round updates channel.
+func MessageReceiver(ctx *context.Context, updatesCh chan *pb.RoundInfo,
 	quitCh <-chan struct{}) {
 	done := false
 	for !done {
 		select {
 		case <-quitCh:
 			done = true
-			// case m := <-messagesCh:
-			// 	ReceiveMessage(ctx, m)
+		case round := <-updatesCh:
+			ReceiveMessages(ctx, round)
+		}
 	}
 }
+
+func getMessagesFromGateway(ctx *context.Context,
+	roundInfo *pb.RoundInfo) []format.Message {
+	return nil
+}
+
+func receiveMessage(ctx *context.Context, msg format.Message) {
+	// do stuff
 }

network/rounds.go

@@ -8,57 +8,59 @@ package network
 
 import (
 	"gitlab.com/elixxir/client/context"
+	"gitlab.com/elixxir/client/context/params"
 	"gitlab.com/elixxir/client/context/stoppable"
 	pb "gitlab.com/elixxir/comms/mixmessages"
-	//	"time"
+	"gitlab.com/xx_network/primitives/id"
+	"time"
 )
 
 // StartProcessHistoricalRounds starts a worker for processing round
 // history.
-func StartProcessHistoricalRounds(ctx *context.Context) stoppable.Stoppable {
+func StartProcessHistoricalRounds(ctx *context.Context,
+	network *Manager) stoppable.Stoppable {
 	stopper := stoppable.NewSingle("ProcessHistoricalRounds")
-	go ProcessHistoricalRounds(ctx, stopper.Quit())
+	go ProcessHistoricalRounds(ctx, network, stopper.Quit())
 	return stopper
 }
 
 // ProcessHistoricalRounds analyzes round history to see if this Client
 // needs to check for messages at any of the gateways which completed
 // those rounds.
-func ProcessHistoricalRounds(ctx *context.Context, quitCh <-chan struct{}) {
-	// ticker := time.NewTicker(ctx.GetTrackNetworkPeriod())
-	// var rounds []RoundID
+func ProcessHistoricalRounds(ctx *context.Context, network *Manager,
+	quitCh <-chan struct{}) {
+	opts := params.GetDefaultNetwork()
+	ticker := time.NewTicker(opts.TrackNetworkPeriod)
+	var rounds []id.Round
 	done := false
 	for !done {
-		//shouldProcess := false
+		shouldProcess := false
 		select {
 		case <-quitCh:
 			done = true
-			// case <-ticker:
-			// 	if len(rounds) > 0 {
-			// 		shouldProcess = true
-			// 	}
-			// case rid := <-ctx.GetHistoricalRoundsCh():
-			// 	rounds = append(rounds, rid)
-			// 	if len(rounds) > ctx.GetSendSize() {
-			// 		shouldProcess = true
-			// 	}
-			// }
-			// if !shouldProcess {
-			// 	continue
-			// }
+		case <-ticker.C:
+			if len(rounds) > 0 {
+				shouldProcess = true
+			}
+		case rid := <-network.GetHistoricalLookupCh():
+			rounds = append(rounds, rid)
+			if len(rounds) > opts.MaxHistoricalRounds {
+				shouldProcess = true
+			}
+		}
+		if !shouldProcess {
+			continue
+		}
 
-			// var roundInfos []*RoundInfo
-			// roundInfos = processHistoricalRounds(ctx, rounds)
-			// rounds := make([]RoundID)
-			// for _, ri := range roundInfos {
-			// 	ctx.GetMessagesCh() <- ri
-			// }
+		roundInfos := processHistoricalRounds(ctx, rounds)
+		for _, ri := range roundInfos {
+			network.GetRoundUpdateCh() <- ri
 		}
 	}
 }
 
 func processHistoricalRounds(ctx *context.Context,
-	rids []uint64) []*pb.RoundInfo {
+	rids []id.Round) []*pb.RoundInfo {
 	// for loop over rids?
 	// network := ctx.GetNetwork()
 	// gw := network.GetGateway()

network/updates.go

@@ -20,9 +20,14 @@ package network
 import (
 	"encoding/binary"
 	"gitlab.com/elixxir/client/context"
+	"gitlab.com/elixxir/client/context/params"
 	"gitlab.com/elixxir/client/context/stoppable"
-	"gitlab.com/elixxir/comms/network"
-	"gitlab.com/xx_network/primitives/ndf"
+	//"gitlab.com/elixxir/comms/network"
+	//"gitlab.com/xx_network/primitives/ndf"
+	"fmt"
+	jww "github.com/spf13/jwalterweatherman"
+	pb "gitlab.com/elixxir/comms/mixmessages"
+	"gitlab.com/xx_network/primitives/id"
 	"io"
 	"math"
 	"time"
@@ -67,13 +72,14 @@ func StartTrackNetwork(ctx *context.Context, net *Manager) stoppable.Stoppable {
 // round status, and informs the client when messages can be retrieved.
 func TrackNetwork(ctx *context.Context, network *Manager,
 	quitCh <-chan struct{}) {
-	ticker := time.NewTicker(ctx.GetTrackNetworkPeriod())
+	opts := params.GetDefaultNetwork()
+	ticker := time.NewTicker(opts.TrackNetworkPeriod)
 	done := false
 	for !done {
 		select {
 		case <-quitCh:
 			done = true
-		case <-ticker:
+		case <-ticker.C:
 			trackNetwork(ctx, network)
 		}
 	}
@@ -83,12 +89,19 @@ func trackNetwork(ctx *context.Context, network *Manager) {
 	instance := ctx.Manager.GetInstance()
 	comms := network.Comms
 	ndf := instance.GetPartialNdf().Get()
-	rng := ctx.Rng
+	rng := ctx.Rng.GetStream()
+	defer rng.Close()
+	sess := ctx.Session
 
 	// Get a random gateway
 	gateways := ndf.Gateways
-	gwID := gateways[ReadRangeUint32(0, len(gateways), rng)].GetGatewayId()
-	gwHost, ok := comms.GetHost(gwHost)
+	gwIdx := ReadRangeUint32(0, uint32(len(gateways)), rng)
+	gwID, err := gateways[gwIdx].GetGatewayId()
+	if err != nil {
+		jww.ERROR.Printf(err.Error())
+		return
+	}
+	gwHost, ok := comms.GetHost(gwID)
 	if !ok {
 		jww.ERROR.Printf("could not get host for gateway %s", gwID)
 		return
@@ -96,18 +109,22 @@ func trackNetwork(ctx *context.Context, network *Manager) {
 
 	// Poll for the new NDF
 	pollReq := pb.GatewayPoll{
-		NDFHash:       instance.GetPartialNdf().GetHash(),
-		LastRound:     instance.GetLastRoundID(),
-		LastMessageID: nil,
+		Partial: &pb.NDFHash{
+			Hash: instance.GetPartialNdf().GetHash(),
+		},
+		LastUpdate:    uint64(instance.GetLastRoundID()),
+		LastMessageID: "",
 	}
-	pollResp, err := comms.SendPoll(gwHost)
+	pollResp, err := comms.SendPoll(gwHost, &pollReq)
 	if err != nil {
-		jww.ERROR.Printf(err)
+		jww.ERROR.Printf(err.Error())
+		return
 	}
-	newNDF := pollResp.NDF
-	lastRoundInfo := pollResp.RoundInfo
+	newNDF := pollResp.PartialNDF
+	lastRoundInfo := pollResp.LastRound
 	roundUpdates := pollResp.Updates
-	newMessageIDs := pollRespon.NewMessageIDs
+	// This is likely unused in favor of new API
+	//newMessageIDs := pollResp.NewMessageIDs
 
 	// ---- NODE EVENTS ----
 	// NOTE: this updates the structure AND sends events over the node
@@ -115,15 +132,64 @@ func trackNetwork(ctx *context.Context, network *Manager) {
 	instance.UpdatePartialNdf(newNDF)
 
 	// ---- Round Processing -----
+	checkedRounds := sess.GetCheckedRounds()
+	roundChecker := getRoundChecker(ctx, network, roundUpdates)
+	checkedRounds.RangeUnchecked(id.Round(lastRoundInfo.ID), roundChecker)
 
-	// rounds, err = network.UpdateRounds(ctx, ndf)
-	// if err != nil {
-	// 	// ...
-	// }
+	// FIXME: Seems odd/like a race condition to do this here, but this is
+	// spec. Fix this to either eliminate race condition or not make it
+	// weird. This is tied to if a round is processing. It appears that
+	// if it is processing OR already checked is the state we care about,
+	// because we really want to know if we should look it up and process,
+	// and that could be done via storage inside range Unchecked?
+	for _, ri := range roundUpdates {
+		checkedRounds.Check(id.Round(ri.ID))
+	}
+}
 
-	// err = rounds.GetKnownRound().MaskedRange(gateway,
-	// 	network.CheckRoundsFunction)
-	// if err != nil {
-	// 	// ...
-	// }
+// getRoundChecker passes a context and the round infos received by the
+// gateway to the funky round checker api to update round state.
+// The returned function passes round event objects over the context
+// to the rest of the message handlers for getting messages.
+func getRoundChecker(ctx *context.Context, network *Manager,
+	roundInfos []*pb.RoundInfo) func(roundID id.Round) bool {
+	return func(roundID id.Round) bool {
+		//sess := ctx.Session
+		processing := network.Processing
+
+		// Set round to processing, if we can
+		// FIXME: this appears to be a race condition -- either fix
+		// or make it not look like one.
+		if processing.IsProcessing(roundID) {
+			return false
+		}
+		processing.Add(roundID)
+		// FIXME: Spec has us SETTING processing, but not REMOVING it
+		// until the get messages thread completes the lookup, this
+		// is smell that needs refining. It seems as if there should be
+		// a state that lives with the round info as soon as we know
+		// about it that gets updated at different parts...not clear
+		// needs to be thought through.
+		//defer processing.Remove(roundID)
+
+		// TODO: Bloom filter lookup -- return true when we don't have
+		// Go get the round from the round infos, if it exists
+
+		// For now, if we have the round in th round updates,
+		// process it, otherwise call historical rounds code
+		// to go find it if possible.
+		for _, ri := range roundInfos {
+			rID := id.Round(ri.ID)
+			if rID == roundID {
+				// Send to get message processor
+				network.GetRoundUpdateCh() <- ri
+				return false
+			}
+		}
+
+		// If we didn't find it, send to historical rounds processor
+		network.GetHistoricalLookupCh() <- roundID
+
+		return false
+	}
 }