Skip to content

Function containing atomic operations should be private #4041

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.

Sign up for GitHub

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

Merged
merged 3 commits into from
Mar 11, 2024
Merged

Function containing atomic operations should be private #4041

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
be39731
Calling a function containing atomic operations from a different file…
mvicsokolova Feb 9, 2024
3f60c28
fixup
mvicsokolova Feb 14, 2024
f29bb49
Dropped a comment
mvicsokolova Mar 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 23 additions & 2 deletions kotlinx-coroutines-core/common/src/channels/BufferedChannel.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -241,7 +241,7 @@ internal open class BufferedChannel<E>(
/**
* Abstract send implementation.
*/
protected inline fun <R> sendImpl(
private inline fun <R> sendImpl(
/* The element to be sent. */
element: E,
/* The waiter to be stored in case of suspension,
Expand Down Expand Up @@ -350,6 +350,27 @@ internal open class BufferedChannel<E>(
}
}

protected fun trySendDropOldest(element: E): ChannelResult<Unit> =
sendImpl( // <-- this is an inline function
element = element,
// Put the element into the logical buffer even
// if this channel is already full, the `onSuspend`
// callback below extract the first (oldest) element.
waiter = BUFFERED,
// Finish successfully when a rendezvous has happened
// or the element has been buffered.
onRendezvousOrBuffered = { return success(Unit) },
// In case the algorithm decided to suspend, the element
// was added to the buffer. However, as the buffer is now
// overflowed, the first (oldest) element has to be extracted.
onSuspend = { segm, i ->
dropFirstElementUntilTheSpecifiedCellIsInTheBuffer(segm.id * SEGMENT_SIZE + i)
return success(Unit)
},
// If the channel is closed, return the corresponding result.
onClosed = { return closed(sendException) }
)

private inline fun sendImplOnNoWaiter(
/* The working cell is specified by
the segment and the index in it. */
Expand Down Expand Up @@ -1587,7 +1608,7 @@ internal open class BufferedChannel<E>(
* It is nulled-out on both completion and cancellation paths that
* could happen concurrently.
*/
@BenignDataRace
//@BenignDataRace
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What did this poor thing do to deserve being commented out?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed, I did not mean it)

private var continuation: CancellableContinuationImpl<Boolean>? = null

// `hasNext()` is just a special receive operation.
Expand Down
21 changes: 0 additions & 21 deletions kotlinx-coroutines-core/common/src/channels/ConflatedBufferedChannel.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,27 +72,6 @@ internal open class ConflatedBufferedChannel<E>(
return success(Unit)
}

private fun trySendDropOldest(element: E): ChannelResult<Unit> =
sendImpl( // <-- this is an inline function
element = element,
// Put the element into the logical buffer even
// if this channel is already full, the `onSuspend`
// callback below extract the first (oldest) element.
waiter = BUFFERED,
// Finish successfully when a rendezvous has happened
// or the element has been buffered.
onRendezvousOrBuffered = { return success(Unit) },
// In case the algorithm decided to suspend, the element
// was added to the buffer. However, as the buffer is now
// overflowed, the first (oldest) element has to be extracted.
onSuspend = { segm, i ->
dropFirstElementUntilTheSpecifiedCellIsInTheBuffer(segm.id * SEGMENT_SIZE + i)
return success(Unit)
},
// If the channel is closed, return the corresponding result.
onClosed = { return closed(sendException) }
)

@Suppress("UNCHECKED_CAST")
override fun registerSelectForSend(select: SelectInstance<*>, element: Any?) {
// The plain `send(..)` operation never suspends. Thus, either this
Expand Down