CODEWITHSUNDEEP
go
orcaman
orcaman

Reputation: 6561

Golang: limit concurrency levels of a blocking operation

I have the following scenario: I am receiving a message on a channel telling me to upload a file. The upload is made by the blocking function uploadToServer. The zipGen channel may receive several messages per second, and I want to upload maximum 5 files simultaneously (not more, but possibly less - depending on how many messages are sent on zipGen by a third worker that is out of the scope of this question).

The listenToZips function runs inside a go routine (go listenToZips() on the file's init function):

func listenToZips() {
    for {
        select {
        case zip := <-zipGen:
          uploadToServer(zip) // this is blocking
        }
    }
}

If I launch go uploadToServer(zip) instead of just uploadToServer(zip) - I get too much concurrency (so for example my program will try to upload 10 files at the same time, but I want a maximum of 5).

On the other hand, without go uploadToServer(zip) (just using uploadToServer(zip) like in the above function), I only upload one file at a time (since the uploadToServer(zip) is blocking).

How can I achieve this level of control to allow me a max upload of 5 files simultaneously?

Thanks!

Upvotes: 2

Views: 1459

Answers (2)

korovkin
korovkin

Reputation: 21

try this: https://github.com/korovkin/limiter

 limiter := NewConcurrencyLimiter(10)
 limiter.Execute(func() {
        uploadToServer()
 })
 limiter.Wait()

Upvotes: 1

Not_a_Golfer
Not_a_Golfer

Reputation: 49195

The simplest option - prespawn N goroutines that take input from the channel, and upload it, in a loop. In each goroutine's context the operation will be blocking, but N goroutines do this. Only one goroutine will receive each message, of course.

func listenToZips(concurrent int) {

    for i:=0; i < concurrent; i++ {

      // spawn a listener goroutine
      go func() {

         for {
            select {
            case zip := <-zipGen:
               uploadToServer(zip) // this is blocking
            }
         }
      }()

   }

}

Of course you can then add stop condition, probably using a different channel, but the basic idea is just the same.

Upvotes: 3

Related Questions