implemented, needs to be thread though and exposed

cmix/interface.go

@@ -191,9 +191,13 @@ type Client interface {
 	AddService(clientID *id.ID, newService message.Service,
 		response message.Processor)
 
-	// IncreaseParallelNodeRegistration increases the number of parallel node
-	// registrations by num
-	IncreaseParallelNodeRegistration(num int) func() (stoppable.Stoppable, error)
+	//PauseNodeRegistrations stops all node registrations
+	//and returns a function to resume them
+	PauseNodeRegistrations(timeout time.Duration) error
+
+	// ChangeNumberOfNodeRegistrations changes the number of parallel node
+	// registrations up to the initialized maximum
+	ChangeNumberOfNodeRegistrations(toRun int, timeout time.Duration) error
 
 	// DeleteService deletes a message service. If only a single response is
 	// associated with the preimage, the entire preimage is removed. If there is

cmix/nodes/interfaces.go

@@ -27,9 +27,13 @@ type Registrar interface {
 	// to register with nodes.
 	StartProcesses(numParallel uint) stoppable.Stoppable
 
-	// IncreaseParallelNodeRegistration increases the number of parallel node
-	// registrations by num
-	IncreaseParallelNodeRegistration(num int) func() (stoppable.Stoppable, error)
+	//PauseNodeRegistrations stops all node registrations
+	//and returns a function to resume them
+	PauseNodeRegistrations(timeout time.Duration) error
+
+	// ChangeNumberOfNodeRegistrations changes the number of parallel node
+	// registrations up to the initialized maximum
+	ChangeNumberOfNodeRegistrations(toRun int, timeout time.Duration) error
 
 	// GetNodeKeys returns a MixCypher for the topology and a list of nodes it did
 	// not have a key for. If there are missing keys, then returns nil.

cmix/nodes/register.go

@@ -13,6 +13,7 @@ import (
 	"gitlab.com/xx_network/crypto/csprng"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/pkg/errors"
@@ -33,14 +34,28 @@ import (
 func registerNodes(r *registrar, s session, stop *stoppable.Single,
 	inProgress, attempts *sync.Map) {
 
+	atomic.AddInt64(r.numberRunning, 1)
+
 	interval := time.Duration(500) * time.Millisecond
 	t := time.NewTicker(interval)
 	for {
 		select {
+		case <-r.pauser:
+			atomic.AddInt64(r.numberRunning, -1)
+			select {
+			case <-stop.Quit():
+				// On a stop signal, close the thread
+				t.Stop()
+				stop.ToStopped()
+				return
+			case <-r.resumer:
+				atomic.AddInt64(r.numberRunning, 1)
+			}
 		case <-stop.Quit():
 			// On a stop signal, close the thread
 			t.Stop()
 			stop.ToStopped()
+			atomic.AddInt64(r.numberRunning, -1)
 			return
 
 		case gw := <-r.c:

cmix/nodes/registrar.go

@@ -20,6 +20,7 @@ import (
 	"gitlab.com/xx_network/primitives/ndf"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"time"
 )
 
@@ -51,6 +52,11 @@ type registrar struct {
 	// operator at a time, as a result this is a map of ID -> int
 	attempts sync.Map
 
+	pauser        chan interface{}
+	resumer       chan interface{}
+	numberRunning *int64
+	maxRunning    int
+
 	c chan network.NodeGateway
 }
 
@@ -60,10 +66,15 @@ func LoadRegistrar(session session, sender gateway.Sender,
 	comms RegisterNodeCommsInterface, rngGen *fastRNG.StreamGenerator,
 	c chan network.NodeGateway) (Registrar, error) {
 
+	running := int64(0)
+
 	kv := session.GetKV().Prefix(prefix)
 	r := &registrar{
 		nodes:         make(map[id.ID]*key),
 		kv:            kv,
+		pauser:        make(chan interface{}),
+		resumer:       make(chan interface{}),
+		numberRunning: &running,
 	}
 
 	obj, err := kv.Get(storeKey, currentKeyVersion)
@@ -95,6 +106,7 @@ func LoadRegistrar(session session, sender gateway.Sender,
 // to register with nodes.
 func (r *registrar) StartProcesses(numParallel uint) stoppable.Stoppable {
 	multi := stoppable.NewMulti("NodeRegistrations")
+	r.maxRunning = int(numParallel)
 
 	for i := uint(0); i < numParallel; i++ {
 		stop := stoppable.NewSingle("NodeRegistration " + strconv.Itoa(int(i)))
@@ -106,21 +118,56 @@ func (r *registrar) StartProcesses(numParallel uint) stoppable.Stoppable {
 	return multi
 }
 
-// IncreaseParallelNodeRegistration increases the number of parallel node
-// registrations by num
-func (r *registrar) IncreaseParallelNodeRegistration(numParallel int) func() (stoppable.Stoppable, error) {
-	return func() (stoppable.Stoppable, error) {
-		multi := stoppable.NewMulti("NodeRegistrationsIncrease")
+//PauseNodeRegistrations stops all node registrations
+//and returns a function to resume them
+func (r *registrar) PauseNodeRegistrations(timeout time.Duration) error {
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	numRegistrations := atomic.LoadInt64(r.numberRunning)
+	for i := int64(0); i < numRegistrations; i++ {
+		select {
+		case r.pauser <- struct{}{}:
+		case <-timer.C:
+			return errors.New("Timed out on pausing node registration")
+		}
+	}
+	return nil
+}
 
-		for i := uint(0); i < uint(numParallel); i++ {
-			stop := stoppable.NewSingle("NodeRegistration Increase" + strconv.Itoa(int(i)))
+// ChangeNumberOfNodeRegistrations changes the number of parallel node
+// registrations up to the initialized maximum
+func (r *registrar) ChangeNumberOfNodeRegistrations(toRun int,
+	timeout time.Duration) error {
+	numRunning := int(atomic.LoadInt64(r.numberRunning))
 
-			go registerNodes(r, r.session, stop, &r.inProgress, &r.attempts)
-			multi.Add(stop)
+	if toRun+numRunning > r.maxRunning {
+		return errors.Errorf("Cannot change number of " +
+			"running node registration to number greater than the max")
+	}
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	if numRunning < toRun {
+		jww.INFO.Printf("ChangeNumberOfNodeRegistrations(%d) Reducing number "+
+			"of node registrations from %d to %d", numRunning, toRun)
+		for i := 0; i < toRun-numRunning; i++ {
+			select {
+			case r.pauser <- struct{}{}:
+			case <-timer.C:
+				return errors.New("Timed out on reducing node registration")
+			}
+		}
+	} else if numRunning > toRun {
+		jww.INFO.Printf("ChangeNumberOfNodeRegistrations(%d) Increasing number "+
+			"of node registrations from %d to %d", numRunning, toRun)
+		for i := 0; i < toRun-numRunning; i++ {
+			select {
+			case r.resumer <- struct{}{}:
+			case <-timer.C:
+				return errors.New("Timed out on increasing node registration")
+			}
 		}
-
-		return multi, nil
 	}
+	return nil
 }
 
 // GetNodeKeys returns a MixCypher for the topology and a list of nodes it did