@@ -23,13 +23,15 @@ internal open class ArrayChannel<E>(
2323 /* *
2424 * Buffer capacity.
2525 */
26- val capacity : Int
26+ private val capacity : Int ,
27+ private val onBufferOverflow : BufferOverflow
2728) : AbstractChannel<E>() {
2829 init {
2930 require(capacity >= 1 ) { " ArrayChannel capacity must be at least 1, but $capacity was specified"
3031 }
3132
3233 private val lock = ReentrantLock ()
34+
3335 /*
3436 * Guarded by lock.
3537 * Allocate minimum of capacity and 16 to avoid excess memory pressure for large channels when it's not necessary.
@@ -41,7 +43,7 @@ internal open class ArrayChannel<E>(
4143 protected final override val isBufferAlwaysEmpty: Boolean get() = false
4244 protected final override val isBufferEmpty: Boolean get() = size.value == 0
4345 protected final override val isBufferAlwaysFull: Boolean get() = false
44- protected final override val isBufferFull: Boolean get() = size.value == capacity
46+ protected final override val isBufferFull: Boolean get() = size.value == capacity && onBufferOverflow == BufferOverflow . SUSPEND
4547
4648 override val isFull: Boolean get() = lock.withLock { isFullImpl }
4749 override val isEmpty: Boolean get() = lock.withLock { isEmptyImpl }
@@ -53,31 +55,26 @@ internal open class ArrayChannel<E>(
5355 lock.withLock {
5456 val size = this .size.value
5557 closedForSend?.let { return it }
56- if (size < capacity) {
57- // tentatively put element to buffer
58- this .size.value = size + 1 // update size before checking queue (!!!)
59- // check for receivers that were waiting on empty queue
60- if (size == 0 ) {
61- loop@ while (true ) {
62- receive = takeFirstReceiveOrPeekClosed() ? : break @loop // break when no receivers queued
63- if (receive is Closed ) {
64- this .size.value = size // restore size
65- return receive!!
66- }
67- val token = receive!! .tryResumeReceive(element, null )
68- if (token != null ) {
69- assert { token == = RESUME_TOKEN }
70- this .size.value = size // restore size
71- return @withLock
72- }
58+ // update size before checking queue (!!!)
59+ updateBufferSize(size)?.let { return it }
60+ // check for receivers that were waiting on empty queue
61+ if (size == 0 ) {
62+ loop@ while (true ) {
63+ receive = takeFirstReceiveOrPeekClosed() ? : break @loop // break when no receivers queued
64+ if (receive is Closed ) {
65+ this .size.value = size // restore size
66+ return receive!!
67+ }
68+ val token = receive!! .tryResumeReceive(element, null )
69+ if (token != null ) {
70+ assert { token == = RESUME_TOKEN }
71+ this .size.value = size // restore size
72+ return @withLock
7373 }
7474 }
75- ensureCapacity(size)
76- buffer[(head + size) % buffer.size] = element // actually queue element
77- return OFFER_SUCCESS
7875 }
79- // size == capacity: full
80- return OFFER_FAILED
76+ enqueueElement( size, element)
77+ return OFFER_SUCCESS
8178 }
8279 // breaks here if offer meets receiver
8380 receive!! .completeResumeReceive(element)
@@ -90,41 +87,36 @@ internal open class ArrayChannel<E>(
9087 lock.withLock {
9188 val size = this .size.value
9289 closedForSend?.let { return it }
93- if (size < capacity) {
94- // tentatively put element to buffer
95- this .size.value = size + 1 // update size before checking queue (!!!)
96- // check for receivers that were waiting on empty queue
97- if (size == 0 ) {
98- loop@ while (true ) {
99- val offerOp = describeTryOffer(element)
100- val failure = select.performAtomicTrySelect(offerOp)
101- when {
102- failure == null -> { // offered successfully
103- this .size.value = size // restore size
104- receive = offerOp.result
105- return @withLock
106- }
107- failure == = OFFER_FAILED -> break @loop // cannot offer -> Ok to queue to buffer
108- failure == = RETRY_ATOMIC -> {} // retry
109- failure == = ALREADY_SELECTED || failure is Closed <* > -> {
110- this .size.value = size // restore size
111- return failure
112- }
113- else -> error(" performAtomicTrySelect(describeTryOffer) returned $failure "
90+ // update size before checking queue (!!!)
91+ updateBufferSize(size)?.let { return it }
92+ // check for receivers that were waiting on empty queue
93+ if (size == 0 ) {
94+ loop@ while (true ) {
95+ val offerOp = describeTryOffer(element)
96+ val failure = select.performAtomicTrySelect(offerOp)
97+ when {
98+ failure == null -> { // offered successfully
99+ this .size.value = size // restore size
100+ receive = offerOp.result
101+ return @withLock
102+ }
103+ failure == = OFFER_FAILED -> break @loop // cannot offer -> Ok to queue to buffer
104+ failure == = RETRY_ATOMIC -> {} // retry
105+ failure == = ALREADY_SELECTED || failure is Closed <* > -> {
106+ this .size.value = size // restore size
107+ return failure
114108 }
109+ else -> error(" performAtomicTrySelect(describeTryOffer) returned $failure "
115110 }
116111 }
117- // let's try to select sending this element to buffer
118- if (! select.trySelect()) { // :todo: move trySelect completion outside of lock
119- this .size.value = size // restore size
120- return ALREADY_SELECTED
121- }
122- ensureCapacity(size)
123- buffer[(head + size) % buffer.size] = element // actually queue element
124- return OFFER_SUCCESS
125112 }
126- // size == capacity: full
127- return OFFER_FAILED
113+ // let's try to select sending this element to buffer
114+ if (! select.trySelect()) { // :todo: move trySelect completion outside of lock
115+ this .size.value = size // restore size
116+ return ALREADY_SELECTED
117+ }
118+ enqueueElement(size, element)
119+ return OFFER_SUCCESS
128120 }
129121 // breaks here if offer meets receiver
130122 receive!! .completeResumeReceive(element)
@@ -135,6 +127,35 @@ internal open class ArrayChannel<E>(
135127 super .enqueueSend(send)
136128 }
137129
130+ // Guarded by lock
131+ // Result is `OFFER_SUCCESS | OFFER_FAILED | null`
132+ private fun updateBufferSize (currentSize : Int ): Symbol ? {
133+ if (currentSize < capacity) {
134+ size.value = currentSize + 1 // tentatively put it into the buffer
135+ return null // proceed
136+ }
137+ // buffer is full
138+ return when (onBufferOverflow) {
139+ BufferOverflow .SUSPEND -> OFFER_FAILED
140+ BufferOverflow .DROP_LATEST -> OFFER_SUCCESS
141+ BufferOverflow .DROP_OLDEST -> null // proceed, will drop oldest in enqueueElement
142+ }
143+ }
144+
145+ // Guarded by lock
146+ private fun enqueueElement (currentSize : Int , element : E ) {
147+ if (currentSize < capacity) {
148+ ensureCapacity(currentSize)
149+ buffer[(head + currentSize) % buffer.size] = element // actually queue element
150+ } else {
151+ // buffer is full
152+ assert { onBufferOverflow == BufferOverflow .DROP_OLDEST } // the only way we can get here
153+ buffer[head % buffer.size] = null // drop oldest element
154+ buffer[(head + currentSize) % buffer.size] = element // actually queue element
155+ head = (head + 1 ) % buffer.size
156+ }
157+ }
158+
138159 // Guarded by lock
139160 private fun ensureCapacity (currentSize : Int ) {
140161 if (currentSize >= buffer.size) {
@@ -212,7 +233,8 @@ internal open class ArrayChannel<E>(
212233 break @loop
213234 }
214235 failure == = POLL_FAILED -> break @loop // cannot poll -> Ok to take from buffer
215- failure == = RETRY_ATOMIC -> {} // retry
236+ failure == = RETRY_ATOMIC -> {
237+ } // retry
216238 failure == = ALREADY_SELECTED -> {
217239 this .size.value = size // restore size
218240 buffer[head] = result // restore head
0 commit comments