-
Notifications
You must be signed in to change notification settings - Fork 1.9k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Ability to specify the number of permits to acquire and release #1553
Conversation
This implementation will not work properly. To see why, add tests that have multiple coroutines waiting a acquire a semaphore and one |
The bug is fixed and the new one test is added. |
This code will not work either. Add a test where there are zero remaining permits, then one coroutine is trying to acquire two (and so goes to the queue). Another one releases one permit. The first one should not be resumed (because it needs two permits), but will be resumed in this implementation. |
I had to rewrite the logic of the main loop, but I totally documented its stages. Looks like it is still lock free and fast. The new test is added. |
} | ||
|
||
private suspend fun addToQueueAndSuspend() = suspendAtomicCancellableCoroutine<Unit> sc@ { cont -> | ||
private suspend fun tryToAddToQueue(permits: Int) = suspendAtomicCancellableCoroutine<Unit> sc@{ cont -> |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't see how permits
is used by this method. I'm very surprised that tests pass. Seems like some tests are missing.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The previous commit didn't contain significant changes, sorry. Please, recheck.
This code would not work properly when multiple acquires are waiting, each one trying to acquire > 1 permit and we release permits one by one. There's no logic to check which ones should be resumed. |
Sorry, I didn't commit all the code, and this led to a misunderstanding. Now all changes are pushed. |
I have yet closely studied the correctness of this algorithm, but one thing I can tell in advance -- this implementation significantly increases overhead of a simple one-permit acquire/release case by allocating additional |
(cont as CancellableContinuation<Unit>).resume(Unit) | ||
internal fun tryToResumeFromQueue(permits: Int) { | ||
accumulator.getAndAdd(permits) // add thread permits to common accumulator | ||
var remain = accumulator.getAndSet(0) // try to take possession of all the accumulated permits at the moment |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
^ The pair of getAndAdd
and getAndAdd
is not efficient. Should be replaced with one atomic CAS-loop.
I replaced the The benchmark is executed with existed stress tests. Each test was run (N = 1 million) at the following configuration (Intel Q9550, 8Gb DDR2 800).
Also I noticed that teamcity build is failed due to changes in methods' signatures in the Semaphore interface ( |
…s-number # Conflicts: # kotlinx-coroutines-core/common/src/sync/Semaphore.kt
4a49830
to
aff8202
Compare
I fixed some merge conflicts and update public API via |
The feature seems useful, but so far I'm skeptical that this implementation can be made truly correct (linearizable). I am unable to prove its correctness by looking at the code and we still miss tests that would be checking Semaphore's correctness under a wide range of scenarios. As our experience shows, simple tests like the ones added here, are not enough for concurrent algorithms of such complexity. I'm waiting for #1737 to be delivered first. |
@khovanskiy we have added a lincheck test (see #1898), found a couple of bugs, and fixed them. Thus, I am almost sure that your current solution is non-linearizable since it is based on bugged implementation. Could you please rebase onto |
We are currently in a big rush in rewriting all the synchronization primitives (#2045) and are not ready to either guide new contributors through the whole new codebase (and also invest a lot of our effort to code style and proper review iterations. It's typically not a big deal for the rest of kotlinx.coroutines, but highly-polished and complex lock-free and linearizable code is a completely different story in terms of time and effort) or to merge it as is. Thanks for your effort and sorry for the inconvenience |
Bonus: This feature will help to emulate fair
ReadWriteLock
in coroutines: