Poll for messages using TaskExecutor
#178
Conversation
We currently sleep for `pollInterval` when no new messages have been polled from the cluster. This leads to unnecessary slowness of the client. Instead of doing that, we now break up the polling of messages into two distinct approaches: 1. Attempt to poll synchronously: if there a message is polled, we return it. If there is no message, we immediately go to step 2. 2. We create a `DispatchQueue` and run the `consumerPoll` on it using `withTaskExecutorPreference`. We make the `consumerPoll` call wait for up to `pollInterval` before bailing. This prevents us from sleeping on the running thread, and frees up cycles to do other work if required. Resolves swift-server#165
Whoops.
Can't use the current implementation before Swift 6.
mimischi
force-pushed
the
issue-165-nonblocking-poll-task-executor
branch
from
9c4b7dd to
a84f1c0
Compare
|
Haven't done any benchmarks just yet. Would like to do some, before we go about merging this. |
|
The benchmark looks better, but the benchmark suite says all differences are negative—and it's setting a positive difference as the threshold, so while this PR is better, the benchmark ends up failing? |
Sources/Kafka/KafkaConsumer.swift
Outdated
| @@ -12,6 +12,7 @@ | |||
| // | |||
| //===----------------------------------------------------------------------===// | |||
|
|
|||
| import Dispatch | |||
There was a problem hiding this comment.
Don't think we need this import here
There was a problem hiding this comment.
Whoops. Left over from a refactor. Is there a linter that can help point out unused imports?
| #if swift(>=6.0) | ||
| // Wait on a separate thread for the next message. | ||
| return try await withTaskExecutorPreference(queue) { | ||
| try client.consumerPoll(for: Int32(self.pollInterval.inMilliseconds)) |
There was a problem hiding this comment.
What happens after the time out?
There was a problem hiding this comment.
We attempt to retrieve a message for self.pollInterval. If there's still no message, we return nil—the same behavior as in the above if let. I'd expect we get caught up in the while-loop on line 100 until we do receive a message eventually.
| public func enqueue(_ _job: consuming ExecutorJob) { | ||
| let job = UnownedJob(_job) | ||
| queue.async { | ||
| job.runSynchronously( |
There was a problem hiding this comment.
@ktoso Should we call this runSynchronously or the one that also takes isolatedTo?
Yes the benchmarks improved significantly! Negative numbers means fewer allocations. |
It's just funny that the benchmarking suite thinks the benchmark failed, because we are below the positive threshold with our negative numbers :) |
|
Also, should this PR—once approved—update the benchmark baseline as well? |
Yes. We intentionally fail the benchmarks when we improve so we set a new thresholds to make sure we don't regress again. |
There are different return codes so CI can choose how to handle that - for an expected improvement one would check in new baselines manually anyway - for an unexpected improvement something may be wrong with eg. The benchmark setup. But super nice improvements here 👍🏻 |
|
@hassila Oh, fair enough. I've not thought about the unexpected improvement situation. Thanks for bringing that up! |
| let job = UnownedJob(_job) | ||
| queue.async { | ||
| job.runSynchronously( | ||
| on: self.asUnownedTaskExecutor() |
There was a problem hiding this comment.
if you're able to use the
public func runSynchronously(isolatedTo serialExecutor: UnownedSerialExecutor,
taskExecutor: UnownedTaskExecutor) {
here and pass the queue's UnownedSerialExecutor that would be preferable as it would AFAIR be more correct in tracking where this is isolated to.
If this is a pain because the queue does not conform to SerialExecutor on some platforms still... then perhaps conditionalize it to platforms where it is? Or leave as is and let me know and we'll chase fixing the conformance.
There was a problem hiding this comment.
Got you. Since we own the underlying queue here is owned by us nothing should be isolated to it.
We currently sleep for
pollIntervalwhen no new messages have been polled from the cluster. This leads to unnecessary slowness of the client. Instead of doing that, we now break up the polling of messages into two distinct approaches:DispatchQueueand run theconsumerPollon it usingwithTaskExecutorPreference. We make theconsumerPollcall wait for up topollIntervalbefore bailing.This prevents us from sleeping on the running thread, and frees up cycles to do other work if required.
Resolves #165