Senior
// The client passes an object (Task) with data for executing a task, // in our example, we will use an empty structure. // // The handler can process no more than N tasks simultaneously, // and no more than X tasks can be queued for processing. // If there is no space in the queue, immediately return an error to the client. // // A task is taken for processing if there are free handlers available. // We simulate processing duration with time.Sleep(5*time.Second). // As soon as one task is completed, we take the next task from the queue. // If the queue is empty, we wait for new tasks from clients. // // With a star: extend the Task structure and get task statuses through a separate request package scheduler type Scheduler interface { AddTask(t Task) error Close() } type scheduler struct { queue chan Task wg sync.WaitGroup } func New(workers, queue int) Scheduler { } type Task interface { Do() }
sobes.tech AI
Answer from AI
Below is an example implementation of Scheduler in Go, which limits the number of concurrently processed tasks (workers) and the queue size. If the queue is full, an error is returned. Tasks are processed in parallel, and after a task is completed, the next one from the queue is taken.
package scheduler
import (
"errors"
"sync"
"time"
)
type Task interface {
Do()
}
type scheduler struct {
queue chan Task
workers int
wg sync.WaitGroup
quit chan struct{}
}
func New(workers, queue int) Scheduler {
s := &scheduler{
queue: make(chan Task, queue),
workers: workers,
quit: make(chan struct{}),
}
for i := 0; i < workers; i++ {
go s.worker()
}
return s
}
func (s *scheduler) worker() {
for {
select {
case task := <-s.queue:
s.wg.Add(1)
task.Do()
s.wg.Done()
case <-s.quit:
return
}
}
}
func (s *scheduler) AddTask(t Task) error {
select {
case s.queue <- t:
return nil
default:
return errors.New("queue is full")
}
}
func (s *scheduler) Close() {
close(s.quit)
// Wait for all tasks to complete
s.wg.Wait()
}
// Example implementation of Task
// type MyTask struct{}
// func (t MyTask) Do() {
// time.Sleep(5 * time.Second) // simulate work
// }
Explanations:
queueis a buffered channel for tasks, with sizequeue.- If the channel is full,
AddTaskreturns an error. workersis the number of handler goroutines.- Each goroutine reads tasks from the channel and executes
Do(). Closestops the handlers and waits for all tasks to finish.
To implement task statuses, you can extend the Task interface with a structure containing ID and status, and add a map for storing statuses with synchronization.