Revert "cleanup"

This reverts commit e2a259a.

Author: stevenlanders

tasks/scheduler.go

@@ -52,7 +52,8 @@ type deliverTxTask struct {
 	ValidateCh    chan struct{}
 }
 
-func (dt *deliverTxTask) Reset() {
+func (dt *deliverTxTask) Increment() {
+	dt.Incarnation++
 	dt.Status = statusPending
 	dt.Response = nil
 	dt.Abort = nil
@@ -62,10 +63,6 @@ func (dt *deliverTxTask) Reset() {
 	dt.ValidateCh = make(chan struct{}, 1)
 }
 
-func (dt *deliverTxTask) Increment() {
-	dt.Incarnation++
-}
-
 // Scheduler processes tasks concurrently
 type Scheduler interface {
 	ProcessAll(ctx sdk.Context, reqs []*sdk.DeliverTxEntry) ([]types.ResponseDeliverTx, error)
@@ -247,8 +244,8 @@ func (s *scheduler) validateTask(ctx sdk.Context, task *deliverTxTask) bool {
 	_, span := s.traceSpan(ctx, "SchedulerValidate", task)
 	defer span.End()
 
-	if s.shouldRerun(task) {
-		task.Reset()
+	if ok := s.shouldRerun(task); ok {
+		task.Increment()
 		return false
 	}
 	return true
@@ -275,8 +272,7 @@ func (s *scheduler) validateAll(ctx sdk.Context, tasks []*deliverTxTask) ([]*del
 		wg.Add(1)
 		go func(task *deliverTxTask) {
 			defer wg.Done()
-			if !s.validateTask(ctx, task) {
-				task.Increment()
+			if ok := s.validateTask(ctx, task); !ok {
 				mx.Lock()
 				res = append(res, task)
 				mx.Unlock()
@@ -304,10 +300,9 @@ func (s *scheduler) executeAll(ctx sdk.Context, tasks []*deliverTxTask) error {
 		workers = len(tasks)
 	}
 
-	// validationWg waits for all validations to complete
-	// validations happen in separate goroutines in order to wait on previous index
-	validationWg := &sync.WaitGroup{}
-	validationWg.Add(len(tasks))
+	wg := &sync.WaitGroup{}
+
+	wg.Add(len(tasks))
 	for i := 0; i < workers; i++ {
 		grp.Go(func() error {
 			for {
@@ -318,21 +313,27 @@ func (s *scheduler) executeAll(ctx sdk.Context, tasks []*deliverTxTask) error {
 					if !ok {
 						return nil
 					}
-					s.prepareAndRunTask(validationWg, ctx, task)
+					s.prepareAndRunTask(wg, ctx, task)
 				}
 			}
 		})
 	}
-
-	for _, task := range tasks {
-		ch <- task
-	}
-	close(ch)
+	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
 	}
-	validationWg.Wait()
+	wg.Wait()
 
 	return nil
 }
@@ -346,7 +347,6 @@ func (s *scheduler) prepareAndRunTask(wg *sync.WaitGroup, ctx sdk.Context, task
 	go func() {
 		defer wg.Done()
 		defer close(task.ValidateCh)
-		// wait on previous task to finish validation
 		if task.Index > 0 {
 			<-s.allTasks[task.Index-1].ValidateCh
 		}