Strange delay with sleeping tasks

[isavcic]

Hey @alitto,

I'm trying to execute a simple test in order to grasp how pond works and I'm experiencing a strange delay when submitting 100 tasks which sleep 1 second each:

package main

import (
    "fmt"
    "time"

    "github.com/alitto/pond"
)

func main() {
    pool := pond.New(100, 1000)
    defer pool.StopAndWait()

    for i := 0; i < 100; i++ {
        n := i
        pool.Submit(func() {
            time.Sleep(1 * time.Second)
            fmt.Printf("Task #%d done\n", n)
        })
    }
}

I was expecting a single, 1 second delay at the end and printout of all "Task # done" at the same time, but I'm seeing tasks executing 1 per second instead. What am I doing wrong? Thanks!

Ivan

[alitto]

Hi @isavcic!

The reason for that behaviour is that worker pools are dynamic by default, which means they create more workers as needed depending on the workload. In your example, the worker pool starts out with 1 worker goroutine and can scale up to 100 if needed, but it never gets to launch more than 2 goroutines because its waiting queue (pool's max. capacity) is never full. In your case, the max. capacity is set to 1000 and only 100 tasks are submitted to the pool. In order to ensure all 100 tasks are processed concurrently by 100 workers, you have 2 alternatives:

1. Reduce the waiting queue size to make sure backpressure is detected sooner:

   pool := pond.New(100, 10)

2. Use a fixed-size pool by configuring the initial no. of workers:

   pool := pond.New(100, 1000, pond.MinWorkers(100))

   Both options will cause the pool to launch more goroutines, which should end up producing the result you were expecting. Please let me know if that works out for you. Have a nice day!

[isavcic]

Thanks for the quick reply! That makes sense. So, maxCapacity doesn't represent the maximum simultaneous, active tasks, but rather a (soft) limit of sorts on the task ingestion buffer? What are the scenarios when that is the bottleneck? I'm still trying to grasp the logic behind it, sorry. :)

By setting the pool as pool := pond.New(100, 0) I'm getting the expected behaviour: 100 tasks maximum running at the same time. Nice!

[alitto]

Thanks for the quick reply! That makes sense. So, maxCapacity doesn't represent the maximum simultaneous, active tasks, but rather as a (soft) limit of sorts on the task ingestion buffer?

That's correct, internally, each pool has 2 buffered channels tasks and dispatchedTasks. When a task is submitted, it's always sent to the tasks channel first. Each pool also has a dispatcher goroutine that continuously reads from the tasks channeel and forwards all items to the dispatchedTasks channel, which is the one used to send the actual tasks to worker goroutines. The size of the first channel (tasks) is controlled by the maxCapacity option, and maxWorkers, on the other hand, controls the size of the second channel (dispatchedTasks), but it also represents the max no. of workers. So basically, maxCapacity determines whether the pool.Submit(task) call is blocking (maxCapacity = 0) or non-blocking (maxCapacity > 0).

What are the scenarios when that is the bottleneck?

The maxCapacity option is there to absorb large bursts of tasks from client goroutines (e.g. a spike in HTTP requests) and avoid blocking on the call to Submit when all workers are busy, because when this happens, the dispatcher also blocks waiting until a worker becomes available.

That said, maybe the scenario when one needs to tune the maxCapacity option is not very frequent, which would mean that option could just default to 0 :thinking:

[alitto]

Hey @isavcic!, I just wanted to let you know that, based on this issue and our discussion around it, I released a new version of the library with changes to give a better control over the pool resizing strategy and also revamped the default strategy to avoid the behaviour you were experiencing. Any feedback you have would be very appreciated :slightly_smiling_face:

[isavcic]

Hey, great! I'm looking over the doc now. If I undestand correctly, if I want immediate task execution (by spawning a new goroutine/worker), I should use the Eager strategy? Currently I ended up doing this by spawning the pool with pool := pond.New(512, 1)

[alitto]

Yes, that's right, Eager would give you that behaviour (immediate task execution unless the maxWorkers limit is reached). Doing pool := pond.New(512, 1) also gives you something similar but because of the maxCapacity option set to 1, client goroutines might spend some time blocked in the Submit() call when more than 1 goroutines attempt to submit tasks simultaneously. Now that the resizing strategy is a different config option, you can safely increase the size of the buffer via maxCapacity and still have immediate execution (this was not possible before). Thanks for your feedback!, it was key to discover this limitation :slightly_smiling_face:

[ioannist]

Any guarantees on the ordering of dispatchedTasks? Completion order is obviously dependant on the worker, but it would help to know if dispatchedTasks are fifo to other.

[alitto]

Hey @ioannist, sorry for the delay, I just came back from vacations. Currently, there's no ordering guarantee when submitting tasks to worker pools. This library uses a buffered channel internally to dispatch tasks to workers, and these do not guarantee delivery order (even though data itself is ordered in the channel). More info:

• https://go.dev/doc/effective_go#channels
• https://www.ardanlabs.com/blog/2014/02/the-nature-of-channels-in-go.html