go/worker/executor: update scheduling parameters on runtime updates

.changelog/3203.bugfix.md

@@ -0,0 +1,4 @@
+Executor scheduling doesn't refresh runtime parameters
+
+Fixes the executor node to watch for runtime scheduling parameter changes and
+if needed update its scheduling configuration.

go/runtime/scheduling/api/api.go

@@ -3,11 +3,15 @@ package api
 
 import (
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
+	registry "github.com/oasisprotocol/oasis-core/go/registry/api"
 	"github.com/oasisprotocol/oasis-core/go/runtime/transaction"
 )
 
 // Scheduler defines an algorithm for scheduling incoming transactions.
 type Scheduler interface {
+	// Name is the scheduler algorithm name.
+	Name() string
+
 	// Initialize initializes the internal scheduler state.
 	// Scheduler should use the provided transaction dispatcher to dispatch
 	// transactions.
@@ -41,6 +45,9 @@ type Scheduler interface {
 	// IsQueued returns if a transaction is queued.
 	IsQueued(hash.Hash) bool
 
+	// UpdateParameters updates the scheduling parameters.
+	UpdateParameters(registry.TxnSchedulerParameters) error
+
 	// Clear clears the transaction queue.
 	Clear()
 }

go/runtime/scheduling/init.go

@@ -3,16 +3,17 @@ package scheduling
 import (
 	"fmt"
 
+	registry "github.com/oasisprotocol/oasis-core/go/registry/api"
 	"github.com/oasisprotocol/oasis-core/go/runtime/scheduling/api"
 	"github.com/oasisprotocol/oasis-core/go/runtime/scheduling/simple"
 )
 
 // New creates a new scheduler.
-func New(name string, maxQueueSize, maxBatchSize, maxBatchSizeBytes uint64) (api.Scheduler, error) {
-	switch name {
+func New(maxQueueSize uint64, params registry.TxnSchedulerParameters) (api.Scheduler, error) {
+	switch params.Algorithm {
 	case simple.Name:
-		return simple.New(maxQueueSize, maxBatchSize, maxBatchSizeBytes)
+		return simple.New(maxQueueSize, params)
 	default:
-		return nil, fmt.Errorf("invalid transaction scheduler algorithm: %s", name)
+		return nil, fmt.Errorf("invalid transaction scheduler algorithm: %s", params.Algorithm)
 	}
 }

go/runtime/scheduling/simple/incoming_queue.go

@@ -226,6 +226,13 @@ func (q *incomingQueue) Take(force bool) (transaction.RawBatch, error) {
 	return batch, nil
 }
 
+func (q *incomingQueue) updateConfig(maxBatchSize, maxBatchSizeBytes uint64) {
+	q.Lock()
+	defer q.Unlock()
+	q.maxBatchSize = maxBatchSize
+	q.maxBatchSizeBytes = maxBatchSizeBytes
+}
+
 func newIncomingQueue(maxQueueSize, maxBatchSize, maxBatchSizeBytes uint64) *incomingQueue {
 	return &incomingQueue{
 		callHashes:        make(map[hash.Hash]bool),

go/runtime/scheduling/simple/simple.go

@@ -2,6 +2,8 @@
 package simple
 
 import (
+	"fmt"
+
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
 	"github.com/oasisprotocol/oasis-core/go/common/logging"
 	registry "github.com/oasisprotocol/oasis-core/go/registry/api"
@@ -14,7 +16,6 @@ const (
 )
 
 type scheduler struct {
-	cfg           config
 	incomingQueue *incomingQueue
 
 	dispatcher api.TransactionDispatcher
@@ -132,16 +133,25 @@ func (s *scheduler) IsInitialized() bool {
 	return s.dispatcher != nil
 }
 
+func (s *scheduler) UpdateParameters(params registry.TxnSchedulerParameters) error {
+	if params.Algorithm != Name {
+		return fmt.Errorf("unexpected transaction scheduling algorithm: %s", params.Algorithm)
+	}
+	s.incomingQueue.updateConfig(params.MaxBatchSize, params.MaxBatchSizeBytes)
+	return nil
+}
+
+func (s *scheduler) Name() string {
+	return Name
+}
+
 // New creates a new simple scheduler.
-func New(maxQueueSize, maxBatchSize, maxBatchSizeBytes uint64) (api.Scheduler, error) {
-	cfg := config{
-		maxQueueSize:      maxQueueSize,
-		maxBatchSize:      maxBatchSize,
-		maxBatchSizeBytes: maxBatchSizeBytes,
+func New(maxQueueSize uint64, params registry.TxnSchedulerParameters) (api.Scheduler, error) {
+	if params.Algorithm != Name {
+		return nil, fmt.Errorf("unexpected transaction scheduling algorithm: %s", params.Algorithm)
 	}
 	scheduler := &scheduler{
-		cfg:           cfg,
-		incomingQueue: newIncomingQueue(cfg.maxQueueSize, cfg.maxBatchSize, cfg.maxBatchSizeBytes),
+		incomingQueue: newIncomingQueue(maxQueueSize, params.MaxBatchSize, params.MaxBatchSizeBytes),
 		logger:        logging.GetLogger("runtime/scheduling").With("scheduler", "simple"),
 	}
 

go/runtime/scheduling/simple/simple_test.go

@@ -5,11 +5,17 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	registry "github.com/oasisprotocol/oasis-core/go/registry/api"
 	"github.com/oasisprotocol/oasis-core/go/runtime/scheduling/tests"
 )
 
 func TestSimpleScheduler(t *testing.T) {
-	algo, err := New(100, 10, 16*1024*1024)
+	params := registry.TxnSchedulerParameters{
+		Algorithm:         Name,
+		MaxBatchSize:      10,
+		MaxBatchSizeBytes: 16 * 1024 * 1024,
+	}
+	algo, err := New(100, params)
 	require.NoError(t, err, "New()")
 
 	tests.SchedulerImplementationTests(t, algo)

go/runtime/scheduling/tests/tester.go

@@ -9,6 +9,7 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/oasisprotocol/oasis-core/go/common/crypto/hash"
+	registry "github.com/oasisprotocol/oasis-core/go/registry/api"
 	"github.com/oasisprotocol/oasis-core/go/runtime/scheduling/api"
 	"github.com/oasisprotocol/oasis-core/go/runtime/transaction"
 )
@@ -91,6 +92,7 @@ func testScheduleTransactions(t *testing.T, td *testDispatcher, scheduler api.Sc
 	require.False(t, scheduler.IsQueued(tx2Bytes), "IsQueued(tx)")
 	require.Equal(t, 1, len(td.DispatchedBatches), "one batch should be dispatched")
 	require.EqualValues(t, transaction.RawBatch{testTx2}, td.DispatchedBatches[0], "transaction should be dispatched")
+	td.Clear()
 
 	// Test schedule batch.
 	testBatch := [][]byte{
@@ -104,6 +106,39 @@ func testScheduleTransactions(t *testing.T, td *testDispatcher, scheduler api.Sc
 	for _, tx := range testBatch {
 		require.True(t, scheduler.IsQueued(hash.NewFromBytes(tx)), fmt.Sprintf("IsQueued(%s)", tx))
 	}
+	// Clear the queue.
+	err = scheduler.Flush(true)
+	require.NoError(t, err, "Flush(force=true)")
+	require.Equal(t, 0, scheduler.UnscheduledSize(), "no transactions after flushing")
+
+	// Test Update configuration.
+	// First insert a transaction.
+	err = scheduler.ScheduleTx(testTx)
+	require.NoError(t, err, "ScheduleTx(testTx)")
+	// Make sure transaction doesn't get scheduled.
+	err = scheduler.Flush(false)
+	require.NoError(t, err, "Flush(force=false)")
+	require.Equal(t, 1, scheduler.UnscheduledSize(), "transaction should remain scheduled after a non-forced flush")
+	// Update configuration to BatchSize=1.
+	err = scheduler.UpdateParameters(
+		registry.TxnSchedulerParameters{
+			Algorithm:         scheduler.Name(),
+			MaxBatchSize:      1,
+			MaxBatchSizeBytes: 10000,
+		},
+	)
+	require.NoError(t, err, "UpdateParameters")
+	// Make sure transaction gets scheduled now.
+	err = scheduler.Flush(false)
+	require.NoError(t, err, "Flush(force=false)")
+	require.Equal(t, 0, scheduler.UnscheduledSize(), "transaction should get scheduled after a non-forced flush")
+
+	// Test invalid udpate.
+	err = scheduler.UpdateParameters(
+		registry.TxnSchedulerParameters{
+			Algorithm: "invalid",
+		},
+	)
+	require.Error(t, err, "UpdateParameters invalid udpate")
 
-	td.Clear()
 }

go/worker/compute/executor/committee/node.go

@@ -1258,8 +1258,6 @@ func (n *Node) worker() {
 	defer hrtNotifier.Stop()
 
 	// Initialize transaction scheduling algorithm.
-	// TODO: watch for updates.
-	// https://github.com/oasisprotocol/oasis-core/issues/3203
 	runtime, err := n.commonNode.Runtime.RegistryDescriptor(n.ctx)
 	if err != nil {
 		n.logger.Error("failed to fetch runtime registry descriptor",
@@ -1268,10 +1266,8 @@ func (n *Node) worker() {
 		return
 	}
 	scheduler, err := scheduling.New(
-		runtime.TxnScheduler.Algorithm,
 		n.scheduleMaxQueueSize,
-		runtime.TxnScheduler.MaxBatchSize,
-		runtime.TxnScheduler.MaxBatchSizeBytes,
+		runtime.TxnScheduler,
 	)
 	if err != nil {
 		n.logger.Error("failed to create new transaction scheduler algorithm",
@@ -1293,6 +1289,16 @@ func (n *Node) worker() {
 	txnScheduleTicker := time.NewTicker(runtime.TxnScheduler.BatchFlushTimeout)
 	defer txnScheduleTicker.Stop()
 
+	// Watch runtime descriptor updates.
+	rtCh, rtSub, err := n.commonNode.Runtime.WatchRegistryDescriptor()
+	if err != nil {
+		n.logger.Error("failed to watch runtimes",
+			"err", err,
+		)
+		return
+	}
+	defer rtSub.Close()
+
 	// We are initialized.
 	close(n.initCh)
 
@@ -1373,6 +1379,19 @@ func (n *Node) worker() {
 				}
 				n.proposeBatchLocked(batch)
 			}()
+		case runtime := <-rtCh:
+			// XXX: Once there is more than one scheduling algorithm available,
+			// this might need to recreate the scheduler and reinsert
+			// the transactions.
+			// At that point also udpate the schedulerMutex usage across the
+			// code, as it will be no longer be true that the scheduler
+			// variable never gets updated.
+			if err = n.scheduler.UpdateParameters(runtime.TxnScheduler); err != nil {
+				n.logger.Error("error updating scheduler parameters",
+					"err", err,
+				)
+				return
+			}
 		case <-txnScheduleTicker.C:
 			// Flush a batch from algorithm.
 			n.scheduler.Flush(true)