Introduction of timeout in passes #5
Conversation
| } | ||
| val duration = | ||
| if timeout == -1 then Duration.Inf else Duration(stopAt - currentTimeInMs, TimeUnit.MILLISECONDS) | ||
| Try(Await.result(future, duration)) match { |
There was a problem hiding this comment.
You have taken care of handling the overall timeout by honouring the remaining timeout at the thread level as well.
We need to use this place to introduce the thread-level timeout.
There was a problem hiding this comment.
This is really not parallelising the processing.
stream.forEach creates one Future (at line no 272) for a given part and then the current main thread will wait for that future to finish (at line no 279). So In principle, you are processing the part sequentially. Which is as good as not using the Future
There was a problem hiding this comment.
In order to implement the parallelism you need to iterate over all parts and start all the Futures and collect them in collection and apply timeout on Seqeunce of Futures.
Refer this code snippet as sample - https://github.com/Privado-Inc/privado-core/blob/main/src/main/scala/ai/privado/utility/ConcurrentProcessor.scala#L94-L103
No description provided.