[occ] Add basic worker task and scheduler shell

baseapp/abci.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/armon/go-metrics"
+	"github.com/cosmos/cosmos-sdk/types/occ"
 	"github.com/gogo/protobuf/proto"
 	abci "github.com/tendermint/tendermint/abci/types"
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
@@ -237,13 +238,22 @@ func (app *BaseApp) CheckTx(ctx context.Context, req *abci.RequestCheckTx) (*abc
 // DeliverTxBatch executes multiple txs
 // TODO: support occ logic with scheduling
 func (app *BaseApp) DeliverTxBatch(ctx sdk.Context, req sdk.DeliverTxBatchRequest) (res sdk.DeliverTxBatchResponse) {
-	// TODO: replace with actual scheduler logic
-	// This is stubbed so that it does something sensible
+	//TODO: inject multiversion store without import cycle (figure out right place for this)
+	// ctx = ctx.WithMultiVersionStore(multiversion.NewMultiVersionStore())
+
+	tasks := make([]*occ.Task, 0, len(req.TxEntries))
+	for idx, tx := range req.TxEntries {
+		tasks = append(tasks, occ.NewTask(tx.Request, idx))
+	}
+
+	scheduler := occ.NewScheduler(app.concurrencyWorkers, app.DeliverTx)
+	if err := scheduler.ProcessAll(ctx, tasks); err != nil {
+		//TODO: handle error
+	}
+
 	responses := make([]*sdk.DeliverTxResult, 0, len(req.TxEntries))
-	for _, tx := range req.TxEntries {
-		responses = append(responses, &sdk.DeliverTxResult{
-			Response: app.DeliverTx(ctx, tx.Request),
-		})
+	for _, tx := range tasks {
+		responses = append(responses, &sdk.DeliverTxResult{Response: *tx.Response})
 	}
 	return sdk.DeliverTxBatchResponse{Results: responses}
 }

types/occ/scheduler.go

@@ -1,12 +1,17 @@
-package scheduler
+package occ
 
-import "errors"
+import (
+	"errors"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/tendermint/tendermint/abci/types"
+	"golang.org/x/sync/errgroup"
+)
 
 var (
 	ErrReadEstimate = errors.New("multiversion store value contains estimate, cannot read, aborting")
 )
 
-// define the return struct for abort due to conflict
+// Abort contains the information for a transaction's conflict
 type Abort struct {
 	DependentTxIdx int
 	Err            error
@@ -18,3 +23,147 @@ func NewEstimateAbort(dependentTxIdx int) Abort {
 		Err:            ErrReadEstimate,
 	}
 }
+
+type TaskStatus string
+
+const (
+	TaskStatusPending   TaskStatus = "pending"
+	TaskStatusExecuted  TaskStatus = "executed"
+	TaskStatusAborted   TaskStatus = "aborted"
+	TaskStatusValidated TaskStatus = "validated"
+)
+
+type Task struct {
+	Status      TaskStatus
+	Index       int
+	Incarnation int
+	Request     types.RequestDeliverTx
+	Response    *types.ResponseDeliverTx
+}
+
+// TODO: (TBD) this might not be necessary to externalize, unless others
+func NewTask(request types.RequestDeliverTx, index int) *Task {
+	return &Task{
+		Request: request,
+		Index:   index,
+		Status:  TaskStatusPending,
+	}
+}
+
+// Scheduler processes tasks concurrently
+type Scheduler interface {
+	ProcessAll(ctx sdk.Context, tasks []*Task) error
+}
+
+type scheduler struct {
+	deliverTx func(ctx sdk.Context, req types.RequestDeliverTx) (res types.ResponseDeliverTx)
+	workers   int
+}
+
+// NewScheduler creates a new scheduler
+func NewScheduler(workers int, deliverTxFunc func(ctx sdk.Context, req types.RequestDeliverTx) (res types.ResponseDeliverTx)) Scheduler {
+	return &scheduler{
+		workers:   workers,
+		deliverTx: deliverTxFunc,
+	}
+}
+
+func (s *scheduler) ProcessAll(ctx sdk.Context, tasks []*Task) error {
+	toExecute := tasks
+	for len(toExecute) > 0 {
+
+		// execute sets statuses of tasks to either executed or aborted
+		err := s.executeAll(ctx, toExecute)
+		if err != nil {
+			return err
+		}
+
+		// validate returns any that should be re-executed
+		// note this processes ALL tasks, not just those recently executed
+		toExecute, err = s.validateAll(ctx, tasks)
+		if err != nil {
+			return err
+		}
+		for _, t := range toExecute {
+			t.Incarnation++
+			t.Status = TaskStatusPending
+			//TODO: reset anything that needs resetting
+		}
+	}
+	return nil
+}
+
+// TODO: validate each task
+// TODO: return list of tasks that are invalid
+func (s *scheduler) validateAll(ctx sdk.Context, tasks []*Task) ([]*Task, error) {
+	var res []*Task
+	for _, t := range tasks {
+		// any aborted tx is known to be suspect here
+		if t.Status == TaskStatusAborted {
+			res = append(res, t)
+		} else {
+			//TODO: validate the task and add it if invalid
+			//TODO: create and handle abort for validation
+			t.Status = TaskStatusValidated
+		}
+	}
+	return res, nil
+}
+
+// ExecuteAll executes all tasks concurrently
+// Tasks are updated with their status
+// TODO: retries on aborted tasks
+// TODO: error scenarios
+func (s *scheduler) executeAll(ctx sdk.Context, tasks []*Task) error {
+	ch := make(chan *Task, len(tasks))
+	grp, gCtx := errgroup.WithContext(ctx.Context())
+
+	// a workers value < 1 means no limit
+	workers := s.workers
+	if s.workers < 1 {
+		workers = len(tasks)
+	}
+
+	for i := 0; i < workers; i++ {
+		grp.Go(func() error {
+			for {
+				select {
+				case <-gCtx.Done():
+					return gCtx.Err()
+				case task, ok := <-ch:
+					if !ok {
+						return nil
+					}
+					//TODO: ensure version multi store is on context
+					//abortCh := make(chan Abort)
+					resp := s.deliverTx(ctx, task.Request)
+
+					//if _, ok := <-abortCh; ok {
+					//	task.Status = TaskStatusAborted
+					//	continue
+					//}
+
+					task.Status = TaskStatusExecuted
+					task.Response = &resp
+				}
+			}
+		})
+	}
+	grp.Go(func() error {
+		defer close(ch)
+		for _, task := range tasks {
+			select {
+			case <-gCtx.Done():
+				return gCtx.Err()
+			case ch <- task:
+			}
+		}
+		return nil
+	})
+
+	if err := grp.Wait(); err != nil {
+		return err
+	}
+
+	return nil
+}

types/occ/scheduler_test.go

@@ -0,0 +1,68 @@
+package occ
+
+import (
+	"context"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/tendermint/tendermint/abci/types"
+	"testing"
+)
+
+type mockDeliverTxFunc func(ctx sdk.Context, req types.RequestDeliverTx) types.ResponseDeliverTx
+
+func (f mockDeliverTxFunc) DeliverTx(ctx sdk.Context, req types.RequestDeliverTx) types.ResponseDeliverTx {
+	return f(ctx, req)
+}
+
+func createTasks(n int) []*Task {
+	tasks := make([]*Task, n)
+	for i := 0; i < n; i++ {
+		tasks[i] = NewTask(types.RequestDeliverTx{}, i)
+	}
+	return tasks
+}
+
+func TestProcessAll(t *testing.T) {
+	tests := []struct {
+		name           string
+		workers        int
+		tasks          []*Task
+		deliverTxFunc  mockDeliverTxFunc
+		expectedStatus TaskStatus
+		expectedErr    error
+		expectedInc    int
+	}{
+		{
+			name:    "All tasks processed without aborts",
+			workers: 2,
+			tasks:   createTasks(5),
+			deliverTxFunc: func(ctx sdk.Context, req types.RequestDeliverTx) types.ResponseDeliverTx {
+				return types.ResponseDeliverTx{}
+			},
+			expectedStatus: TaskStatusValidated,
+			expectedErr:    nil,
+			expectedInc:    0,
+		},
+		//TODO: Add more test cases
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := NewScheduler(tt.workers, tt.deliverTxFunc.DeliverTx)
+			ctx := sdk.Context{}.WithContext(context.Background())
+
+			err := s.ProcessAll(ctx, tt.tasks)
+			if err != tt.expectedErr {
+				t.Errorf("Expected error %v, got %v", tt.expectedErr, err)
+			}
+
+			for _, task := range tt.tasks {
+				if task.Status != tt.expectedStatus {
+					t.Errorf("Expected task status to be %s, got %s", tt.expectedStatus, task.Status)
+				}
+				if task.Incarnation != tt.expectedInc {
+					t.Errorf("Expected task incarnation to be %d, got %d", tt.expectedInc, task.Incarnation)
+				}
+			}
+		})
+	}
+}