Created
February 12, 2022 22:22
-
-
Save djspiewak/8a15a3ee115afe726f2637d907781c06 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
private[effect] final class UnsafeBounded[A](bound: Int) { | |
private[this] val buffer = new Array[AnyRef](bound) | |
private[this] val filled = new Array[Boolean](bound) // can't bitpack this because of word tearing | |
private[this] val first = new AtomicInteger(0) | |
private[this] val last = new AtomicInteger(0) | |
private[this] val length = new AtomicInteger(0) | |
def debug(): String = buffer.mkString("[", ", ", "]") | |
def size(): Int = length.get() | |
@tailrec | |
def put(data: A): Unit = { | |
val oldLast = last.get() | |
if (length.get() >= bound) { | |
throw FailureSignal | |
} else { | |
if (!filled(oldLast) && last.compareAndSet(oldLast, (oldLast + 1) % bound)) { | |
buffer(oldLast) = data.asInstanceOf[AnyRef] | |
filled(oldLast) = true | |
// we're already exclusive with other puts, and take can only *decrease* length, so we don't gate | |
// this also forms a write barrier for buffer | |
length.incrementAndGet() | |
() | |
} else { | |
put(data) | |
} | |
} | |
} | |
@tailrec | |
def take(): A = { | |
val oldFirst = first.get() | |
if (length.get() <= 0) { // read barrier and check that boundary puts have completed | |
throw FailureSignal | |
} else { | |
if (filled(oldFirst) && first.compareAndSet(oldFirst, (oldFirst + 1) % bound)) { | |
val back = buffer(oldFirst).asInstanceOf[A] | |
filled(oldFirst) = false | |
// we're already exclusive with other takes, and put can only *increase* length, so we don't gate | |
// doing this *after* we read from the buffer ensures we don't read a new value when full | |
// specifically, it guarantees that puts will fail even if first has already advanced | |
length.decrementAndGet() | |
buffer(oldFirst) = null // prevent memory leaks (no need to eagerly publish) | |
back | |
} else { | |
take() | |
} | |
} | |
} | |
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment