I'm trying to do some task scheduler and worker. Tasks are added to the queue from which worker retrieves and processes them. Tasks from one queue must be executed sequentially. Tasks from different queues are executed in parallel. When the worker has completed all the tasks in the queue, he waits for a while and finishes his work. The queue must be removed from the schedule
I did something, but I think it isn't the best solution. I think there may be problems when worker shuts down and closes the channel.
Is my solution concurrency-safe?
// Schedule struct
type Schedule struct {
sync.RWMutex
queues map[int]chan Task
idle byte
}
// Scheduler pushes task to the queue
func (s *Schedule) Scheduler(t Task, i int) {
var queue chan Task
var ok bool
s.RLock()
if queue, ok = s.queues[i]; !ok {
s.RUnlock()
s.Lock()
if queue, ok = s.queues[i]; !ok {
queue = make(chan Task)
s.queues[i] = queue
go s.worker(queue, i)
}
s.Unlock()
} else {
s.RUnlock()
}
queue <- t
}
// Worker retrieves task from the queue and process
func (s *Schedule) worker(c chan Task, i int) {
timeout := time.After(s.idle * time.Second)
done := false
for !done {
select {
case task := <-c:
task.Execute()
timeout = time.After(s.idle * time.Second)
case <-timeout:
s.Lock()
close(c)
delete(s.queues, i)
s.Unlock()
done = true
default:
time.Sleep(10 * time.Millisecond)
}
}
}
Is my solution concurrency-safe?
You need to write tests that tests the various behaviors of the algorithm lifetime and run them with the-race
flag to enable the race detector. The code should not need those locks. You should be able to handle events with an async for loop select with an exclusive access to shared variables. Your code does not implement a mechanism to close the processing. I suggest you review those elements before going further. \$\endgroup\$