so i watched this video

Question

so i watched this video and i tried to recreate what he was doing with downloading with an uploader i have uploadWorkers inputchannel with files output channel with uploadResponse an uploadResultHandler which listenes to the uploadResponseChannel and a main method which combines the workers with handler the problem i have now how do i write a main method for that so how does the resulthandler know that he is done listening to uploadresults i can t close the response channel in the worker basically i am confused slightly smiling face

bitkid · Answer

i came up with something like this

bitkid · Answer

```object Uploader JvmStatic fun main args Array lt String gt val files = Channel lt File gt GlobalScope launch uploader files files sendBlocking File a ```

bitkid · Answer

but this only waits until the file is sent not until the response has been processed

bitkid · Answer

i mean what i can do is do uploading and processing the files in one step then i know when all sendBlocking are done i am done

bitkid · Answer

is there maybe a channel implementation which expects a certain amount elements

reline · Answer

Use `runBlocking` e g ``` fun main args Array lt String gt = runBlocking ```

bitkid · Answer

that doesn t really solve the problem so basically i have workloads a finite amount i submit it to a worker pool through a channel which processes the work and and then it submits a work done message to a different channel this channel is then consumed by a different coroutine how can i make sure that this coroutine got a message for each workload submitted to the pool when i close the channel after all tasks have been submitted to the workers i might loose messages on the consuming side

bitkid · Answer

i mean i could use something like a countdownlatch or cyclicbarrier but i thought maybe coroutines has something OOTB

reline · Answer

I would simply move things around ``` object Uploader JvmStatic fun main args Array lt String gt = runBlocking val files = Channel lt File gt GlobalScope launch files send File a uploader files ``` ~now your main function waits for `uploader ` to finish~ I just realized your `uploader ` is an extension of `CoroutineScope` correct

bitkid · Answer

y

bitkid · Answer

maybe using in channels in this context is not the best abstraction

bitkid · Answer

basically i want to upload a lot files concurrently with having control over the concurrency basically nr of threads when using channels i can controll the concurrency by controlling the numbers of consumers of the channel but i guess i could also just use a fixedThreadPool asDispatcher and withContext pool launch

bitkid · Answer

i am also wondering when i submit lets say 100 items to a channel and then call channel close how can i make sure that the items have been processed on the reading side

bitkid · Answer

the approach i used here does what it should do with a countdown latch but i have the feeling somehow that s an anti pattern <https gist github com bitkid 68fd50575f35ab456f5b03a7746437f2>

reline · Answer

yeah remove your latch I whipped something up ``` val files = 1 10 map File $it val channel = Channel lt File gt launch files forEach channel send it channel close val time = measureTimeMillis coroutineScope uploader channel println time ```

reline · Answer

then add `responses close ` inside `uploadWorker ` after your for loop

reline · Answer

I m not 100 sure if using `coroutineScope` like this here is the right way to go about it but the only alternative I can think of is making `uploader ` suspending ``` suspend fun uploader files ReceiveChannel lt File gt nrOfWorkers Int = 4 = coroutineScope ```

Brad Murray · Answer

I have a similar structure in my system and using Channels Basically have a single actor managing state in memory or persisted Send the Batch of items to it to manage have it track what is outstanding per batch and also ensure each completed item is processed by it Essentially a reconciliation issue It fans out to a limited number of workers co routines to control concurrency rather than just spinning up new co routines per item

bitkid · Answer

but when i call responses close after running the loop how can i make sure uploadResultHandler has everything processed already plus when i have multiple workers the worker potentially is still working and wants to sent his last result to responses when another worker already received the last element

bitkid · Answer

what i mean is even if you are the one getting the last file it doesnt mean that you are the last to finish processing

bitkid · Answer

so uploadworker 1 gets element n 1 it takes 10 seconds to process uploadworker 2 gets element n but it only takes 1 second to process and it closes the channel so uploadworker 1 can t send the remaining result

bitkid · Answer

so i need something external to synchronize the results