52
taskQueueFactory SequentialTaskQueueFactory[T],
53
logger log.Logger,
55
>
return &SequentialScheduler[T]{
56
>
status: common.DaemonStatusInitialized,
57
>
shutdownChan: make(chan struct{}),
58
>
options: options,
59
>
60
>
logger: logger,
61
>
62
>
queueFactory: taskQueueFactory,
63
>
queueChan: make(chan SequentialTaskQueue[T], options.QueueSize),
64
>
queues: collection.NewShardedConcurrentTxMap(1024, taskQueueHashFn),
65
>
}
66
>
}
67
69
>
if !atomic.CompareAndSwapInt32(
70
>
&s.status,
71
>
common.DaemonStatusInitialized,
72
>
common.DaemonStatusStarted,
73
>
) {
74
return
75
}
76
77
>
initialWorkerCount, workerCountSubscriptionCancelFn := s.options.WorkerCount(s.updateWorkerCount)
sequential_scheduler.go
78
>
s.workerCountSubscriptionCancelFn = workerCountSubscriptionCancelFn
79
>
s.updateWorkerCount(initialWorkerCount)
80
>
81
>
s.logger.Info("sequential scheduler started")
82
}
83