Parallelize stdout and clean up PersistentWorker #501
Conversation
jeffzoch
requested review from
cgruber,
jin,
restingbull and
timpeut
as code owners
|
|
||
| @ExperimentalCoroutinesApi | ||
| override fun start(execute: Work) = WorkerContext.run { | ||
| //Use channel to serialize writing output | ||
| val writeChannel = Channel<WorkerProtocol.WorkResponse>(UNLIMITED) |
There was a problem hiding this comment.
Main idea is to launch the producer in its own coroutine and have the collector run to completion synchronously. The producer does the compilation in parallel and emits results across a channel which is serially read from. Flows are serial by default so we don't need to synchronize access on the consumer side.
| * @param f - suspending function | ||
| */ | ||
| fun <A, B> Flow<A>.suspendingParallelMap(scope: CoroutineScope, f: suspend (A) -> B): Flow<B> { | ||
| return flowOn(Dispatchers.IO) |
There was a problem hiding this comment.
I'm guessing compilation is CPU-bound, so Dispatchers.DEFAULT or similar might be more appropriate.
| info { "received req: ${request.requestId}" } | ||
| async { | ||
| doTask("request ${request.requestId}") { ctx -> | ||
| .asFlow() |
There was a problem hiding this comment.
Is this flow necessary? Would it be simpler to do something like:
generateSequence { WorkRequest.parseDelimitedFrom(io.input) }.forEach {
launch(Dispatchers.DEFAULT) {
// do the work
writeChannel.send(response)
}
}
There was a problem hiding this comment.
@jdai8 yeah that would totally work too - I didn't want to rock the boat too much but I can change it to that as its easier to reason about.
There was a problem hiding this comment.
Just went ahead and changed to just use coroutines (very similar to your thread implementation now just using kotlin primitives I guess. I'd think we would need to be compiling thousands of files to maybe even see a perf difference)
|
Much cleaner, thank you. Also educational. 👍 |
For discussion in #494