2
\$\begingroup\$

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)
        }
    }
}
\$\endgroup\$
2
  • \$\begingroup\$ besides a code review, how is the scheduling implemented in this snippet ? This snippet implements something that does look like a job queuer. 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\$
    – mh-cbon
    Nov 4, 2019 at 11:54
  • \$\begingroup\$ I need to process jobs from the queue sequentially, in the same order in which they were added. But jobs from different queues must be processed asynchronously. When the queue is empty, the worker waits for a while, then closes and removes the queue from map. This is what I am trying to implement in this code \$\endgroup\$
    – buildok
    Nov 4, 2019 at 12:07

1 Answer 1

1
\$\begingroup\$
  1. There is no need for the default clause in Schedule.worker.
  2. Memory side, it is better to create a time.Timer object and call Timer.Reset & Timer.Stop on it.
  3. No need for the done variable, we can just return when we are done.
func (s *Schedule) worker(c chan Task, i int) {
    t := time.NewTimer(s.idle * time.Second)
    for {
        select {
        case task := <-c:
            task.Execute()
            if !t.Stop() {
                <-t.C
            }
            t.Reset(s.idle * time.Second)
        case <-t.C:
            s.Lock()
            close(c)
            delete(s.queues, i)
            s.Unlock()
            return
         }
    }
}
  1. It is not advisable to close a channel from a receiver. Better to close the channel when you know that there are no more tasks and have the worker wait for that.
func (s *Schedule) worker(c chan Task, i int) {
    for task := range c {
        task.Execute()
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.