#Worker Pools are Blocking

118 messages · Page 1 of 1 (latest)

brittle heron
#

I have a go application which consists of a ticker which takes from a channel and passes it into another channel, which then is consumed within a worker pool, the result of which gets passed into another channel used in another worker pool.

    jobs := make(chan string, 1)
    jobs2 := make(chan string, 1)
    jobs3 := make(chan string, 100000000)

    // How do I free up this worker.
    for w := 1; w <= 5; w++ {
        go o.workerOne(w, jobs, payloads)
    }

    for w := 1; w <= 100; w++ {
        go o.workerTwo(w, payloads, results)
    }

    for w := 1; w <= 1000; w++ {
        go o.workerThree(w, results)
    }

    go func() {
        for {
            select {
            case <-o.Context.Done():
                return
            case _ = <-ticker.C:
                                jobs <- <-o.Queue
                        }
                }
        }

      <-sleep.Done()
    ticker.Stop()

However once jobs has been passed 5 items then no more work is done, the queue is just constantly appended to, how can I free the workers?

honest gorge
#

you don’t “free” them; they have to return themselves

#

this is often accomplished by closing the channel where you send to the channel and checking for the channel being closed within the worker

#

(if you use a for range loop on the channel, it checks for closing automatically, and the loop exits after the channel is closed)

high dagger
#

I'm guessin payloads is jobs2, and results is jobs3

brittle heron
#

Thanks, so I'm using range, this is the pattern I'm using, Ahh i missed an important detail...

# 
func (o *Orchestrator) workerThree(id int, unvalidatedLinks <-chan string, startingUrl *url.URL) {
    for job := range jobs {
              o.Queue <- "something"
         }
}
#

So workerOne pipes to workerTwo and workerThree, workerThree might result in extra items on the o.Queue

#

Which means there isn't a known end to these jobs, unless I timeout.

#

Unless I assume the maximum size 🤔 of the worker pool for job one

high dagger
#

so what's the question here?
if there is no end, there is no need for graceful shutdown of these goroutines

brittle heron
#

The question is, I currently have a worker pool set to 5

    for w := 1; w <= 5; w++ {
        go o.workerOne(w, jobs, payloads)
    }

When I have passed enough "jobs" to execute all of those instances, then no more jobs are processed, I just continuously append on to the jobs queue.

I want to be in a position where I can processes those remaining jobs.

#

Is the only way I can do that is making the pool high enough to be the maximum?

#

It would be good to know how I could structure this to that the program could run indefinitely.

#

In my mind, I should be able to have n number of worker pools which after finishing their work can become available to execute additional tasks, I am most likely misunderstanding something fundamental here, and thats fine becaue I'm learning.

honest gorge
#

the point of having five workers is that you only do up to five jobs concurrently, no…?

#

they should pick up the next job when they're done with their current one

brittle heron
#

the point of having five workers is that you only do up to five jobs concurrently, no…?
No and this is probably where I'm misunderstanding, in my mind I'm creating five available workers which can do work some point in the future but also be re-used for other work in the future - does that make sense?

honest gorge
#

i'm not sure what "other" means there

#

is it coming in on the same channel? i don't really follow how work is "other"

brittle heron
#

Gotcha, let me try and be clearer.

by "other" I mean that additional jobs that come from the ticker.

honest gorge
#

it's that you only do five jobs simultaneously

#

e.g. if you put six jobs on, five would run immediately, and once one of the jobs finishes the sixth would be picked up

brittle heron
#

"once one of the jobs finishes" - when is this triggered, is that when the worker returns?

honest gorge
#

no, the worker probably shouldn't return, else it wouldn't be able to run any more jobs 😁

brittle heron
#

👁️ 👁️

#

Could that be the reason hahaah

high dagger
#

with that code, the goroutines will run until the channels are closed, and we have not seen that code

honest gorge
#

well, if your workers are dying, there's nobody left to pick up the jobs…

high dagger
#

so it will never happen

brittle heron
#

well, if your workers are dying, there's nobody left to pick up the jobs…

#

That makes sense, and I am returning

#

This? would this exit the worker so it couldn't pick anymore work from the queue which is passed to the range?

Screenshot_2024-06-30_at_15.25.19.png
honest gorge
#

if that's in workerOne (and not in a nested function), yes, that would exit the worker

#

return does in fact return 😛

brittle heron
#

😛

hidden lotus
brittle heron
#

Okay let me give this a go, thanks 😅

#

is there any docs that explain this return behaviour?

honest gorge
#

…that return returns from a function?

brittle heron
#

So in the error case, I should use continue?

honest gorge
#

if you want to move onto the next job immediately, sure

brittle heron
#

…that return returns from a function?
😆 now when you say it like that, I guess it pushes that goroutine from being available?

honest gorge
#

it sounds like you think there's more complexity than there really is here. when you return, the function returns, it's not really any more complicated than that

#

when the function returns, you are no longer running the code in the current function to pull off the channel

brittle heron
#

Yeah, you're right

#

Its just a bit weird with that range channel

#

but it now makes sense

honest gorge
#

if it's the top-level function in the goroutine, the goroutine also exits, but that's not really important here

#

the goroutine could still be alive but just running some other function and the issue would still exist

#

btw, i’m not sure if you were just flailing while debugging, but you almost certainly do not want a channel with a buffer of one hundred million

#

i wouldn’t have a buffer at all unless you actually know what it’s doing, which that buffer size indicates that you might not 😛

brittle heron
#

debugging, but you almost certainly do not want a channel with a buffer of one hundred million
Very much flailing, and experimenting

#

I will restrict the size of the buffer to basically the max total size of the requests I'm planning on making, but will learn to see if theres a better number.

#

Thanks for spotting that, and helping

honest gorge
#

try unbuffered to begin with

#

make it buffered if you identify an issue with that

brittle heron
#

The reason why I was using buffered vs unbuffered was that I was aware of the size (before I started playing with it), is that a typical reason or not?

#

But Buffered Channel will also block the goroutine either when it is empty and waiting to be filled or it's on its full-capacity and there's a statement that want to fill the channel.
To me this was the reason I was using it, to avoid having too much of a big buffer, or atleast having a buffer of a defined size.

honest gorge
#

an unbuffered channel is effectively a buffered channel of size 0

#

so it's just about as much "avoid having too much of a big buffer" as you can get

#

i'd ask yourself why you really need a buffered channel before you use one

#

they can help to decouple the processing speeds between stages of the pipeline, to help smooth out transient changes, due to e.g. a network blip

#

they won't change anything in the steady-state, though

#

the downside of a buffered channel is that the entire buffer is allocated the entire time

brittle heron
#

i'd ask yourself why you really need a buffered channel before you use one
I think there are blocking behaviour differences no?

honest gorge
#

could you describe those to us?

brittle heron
#

Hmm, not well but I'll try.

Unbuffered channels, will block until the reciever recieves that value, buffered channels will not block in that case unless they are full?

honest gorge
#

correct, but you have to ask yourself whether that's necessarily a bad thing

#

e.g. if the last step in a pipeline is consistently slower than the first few steps, it might make sense to have those first few steps slow down by blocking

#

that's called "back-pressure"

#

(the last step is pushing back on the rest of the pipeline)

#

a buffer is only a fix up to a certain point, because the buffer can always fill up

brittle heron
#

Gotcha

honest gorge
#

e.g. if the last step is always half the speed of the first step, no buffer will help in the steady-state, because any buffer size will always eventually fill up

#

it's complicated, as you're now trying to think about the behavior of multiple individual components in a complex system. not to say that they have no legitimate use, but i find that buffers are often used as a band-aid fix that allow you to continue to not really understand what's happening, though.

brittle heron
#

Thats fair to say, I will try and have a think if its required - I did use it as a band-aid because at the moment I am dealing with some blocking.

honest gorge
#

there can also be concerns like: if you are concerned about the latency of a single "item" progressing through the pipeline, having large buffers in the middle can mean that the "acceptance"/first stage of the pipeline seems completely fast, but it takes a long time for something to progress through

#

if the pipeline is going to be slow, it might be better to know that upfront rather than having it covered up by a buffer in the middle

brittle heron
#

I think its blocking because there is nothing to receive the value, is that 100% the only way a unbuffered channel will block, it only happens after sometime of the application running 🤔

honest gorge
#

yes, a send to an unbuffered channel will block until a corresponding receive

brittle heron
#

I guess that means the workers are busy and have not been "free", because I would think the code inside the worker

func (o *Orchestrator) requester(id int, requests <-chan Request, results chan<- io.ReadCloser) {
    for request := range requests {
for request := range requests {

would pick it up.

honest gorge
#

i agree, yes

#

in that case, it's working precisely as intended

#

again, i'd consider whether that's actually an "issue" per se

brittle heron
#

Interesting, how do you mean? There is a job which needs to be processed but it can't becuase the workers are "busy" (not freed), it sounds like to me I need to track down where this blockage is 🤔

honest gorge
#

if you only spawn five workers, you're only doing five jobs simultaneously, that's a conscious choice

#

sure, if workers are getting stuck that's one thing, but just not having something accepted immediately is expected; if you didn't want that you'd be spawning an unbounded number of goroutines, but that comes with its own issues

#

(e.g. spawning a goroutine inherently takes time away from all the other goroutines, so it can't go on forever)

brittle heron
#

but just not having something accepted immediately is expected;
yeah this is fine, thats expected but its almost as if all five workers get blocked somehow.

#

if you didn't want that you'd be spawning an unbounded number of goroutines, but that comes with its own issues
yes I want to avoid this, I don't want an unbounded number which will take up much more memory.

timber lagoon
#

What carson is alluding to (I think) is that flow control is useful between producer and consumers.

So a queue of say 2xN work items for N work items is probably the max you need for reasonable incoming rate without huge bursts (means that while 5 workers are working, you can enqueue 10 more todo before blocking on producing side)

tulip escarp
#

workerOne pipes to workerTwo and workerThree, workerThree might result in extra items on the o.Queue

Can this be the cause of a deadlock? If worker 3 is trying to push back into the original channel and that channel is full (it has buffer size 1), then worker 3 will wait and stop consuming from its own queue. Eventually you might get all worker 3's waiting which will result in all worker 2's getting stuck because worker 3's are not cleaning up their own output channels. And then worker 1's will get stuck too.

Hard to tell without seeing the entire code and debugging so I could be wrong.

brittle heron
#

I think you're right @tulip escarp.

Though to be more specific there is a ticker, which takes from o.Queue and creates a job for workerOne, then that returns 1 result to workerTwo, and then workerTwo will return N number of results back to o.Queue.

It looks like workerThree gets stuck because there is nothing collecting the value for o.Queue, and then because of that workerOne gets into a deadlock.

I would like to understand how I could run this indefinitely, to do that I will need to avoid this deadlock, but this data needs to go somewhere to be processsed - I could use a buffered channel but then once it has filled I will be blocked.

Is there an alternative place I could store this data so I don't get blocked?

#

I could add some "buffer" to protect against reaching the max, basically to ensure I never send too many that fills the queue and therefore blocks.

high dagger
#

I don't think the problem is the storage of the data, but the lack of synchronization between the concurrent workers. This might be causing the deadlock. The code does not show clearly whether the third worker produces into the channel consumed by the first worker, and we just don't have enough code to determine the problem. Moreover, I still don't understand what you are trying to accomplish, so I can't suggest a more viable approach.
I get you have a ticker, a goroutine consuming the ticker and producing to a queue, and then we get three sets of goroutines that use channels to pass work along in cascade, but seem unrelated to that queue.

tulip escarp
#

Channels are great as thread-safe mechanism to pass data between goroutines, but they do come with a limited capacity by design and they block (the published and reader) when capacity is reached. There is no unlimited channel in Go. You can implement your own jobs queue as a linked list with a mutex to make it thread safe. That will never block but you are risking a memory overload if you don't consume fast enough from it. You could add some sort of guard against that.

I am also not sure there is a deadlock like I said. It should be possible to not deadlock in this pattern. Again, hard to be sure without reading the full code. To avoid deadlock, you want workers1 to read from queue1 as quickly as possible, then push the job to queue2 in a new goroutine. This guarantees that worker1s will not block. The goroutines will push to queue2 when/if it has capacity. Repeat the same logic for workers2 and workers3. Push to the next queue in a separate goroutine.

high dagger
#

Oh, well, I see now! The third worker is producing to the queue, and the ticker is consuming from the queue and producing to the first channel. Sorry about prev statement.

timber lagoon
#

a channel is a thread safe circular buffer optimized implementation, doing a queue yourself won’t be better and memory is the limit in all cases, which is why flow control is needed

high dagger
#

So there is clearly a cycle in there. Every message that you put in any of those channels, it will go around and around forever, limited only by the speed of the ticker.

#

Whatever code you have somewhere else sending values to those channels, they will eventually clog all the buffers of the channels.

tulip escarp
#

Agreed. Implementing own channel is only different if using linked list so then the memory is not prealloacted. Otherwise, it will be slower than channel.

high dagger
#

Hence @tulip escarp was right.

#

how I could run this indefinitely
The messages are in a cycle now, when/where do you want them to exit the infinite loop?

timber lagoon
#

you probably need 2 channels if your consumers also produce

tulip escarp
#

This pattern reminds me a bit of a web crawler. You crawl a page, find links, then put the new links back on the original queue. Of course this needs to be finite. Can't just infinitely loop and add more links. You might want to read about that pattern to give you some additional perspective.

high dagger
#

Yeah, it fits that pattern pretty well. It just seems to never complete the processing of each page.

timber lagoon
#

breadth first vs depth first... you want to finish something before putting more stuff (though a FIFO queue is supposed to be breadth first as long as you don't have concurrent producers)

high dagger
#

Apache nutch just takes from the hbase "database", processes pages in parallel, then saves links back into hbase creating more pages to crawl in the next loop iterations; you don't need to have all those pages in a channel or even in memory.

timber lagoon
#

of note that yes you could externalize your queue to a database to effectively make it 'infinite'

high dagger
#

Actually, when you have these huge datasets, neither breadth-first nor depth-first are usually good enough, you want to take into account other properties, like number of pages linking to the page url, the TTL of the cacheable resource, and so on; oh, well, I don't want to digress or hijack the OP topic.

timber lagoon
#

I meant purely about order of processing so you don't have "current page" in the stack forever - and you have to deal with cycles too

brittle heron
brittle heron
#

Yeah I want to avoid the o.Queue growing indefinitenly, I do want it to be bounded.

#

Which does mean I might miss some links, but that to me is less of an issue.

high dagger
#

You are not having any control on the number of links or tasks to be processed. You may expect it to stay low but it can actually grow beyond the capacity of the buffered channels. The solution seems easy now: take control of that in your design, handle gracefully both the shutdown when tasks stop spawning more tasks and an excess of tasks. I do not know what you want to do with the excess, but you have stated already one possibility: discard them.