[stm] optimizations #1456
[stm] optimizations #1456
Conversation
| TID.useIO { | ||
| case -1L => | ||
| TID.useNew { tid => | ||
| Tick.withCurrent( |
There was a problem hiding this comment.
I've renamed TID to Tick for clarity since after #1455, the transaction has two timestamps. I've also improved the API to avoid leaking the -1 case.
|
|
||
| private def commit[A, S](tid: Long, log: TRefLog, probe: Boolean = false)(using AllowUnsafe): Boolean = | ||
| // Thread-local cache for the commit buffer to avoid repeated allocations | ||
| private val bufferCache = new ThreadLocal[ArrayList[Any]] |
There was a problem hiding this comment.
cache the buffer used used to traverse the ref log to avoid allocations
| entry match | ||
| case _: TRefLog.Read[?] => | ||
| // Read-only: just validate, no locking needed | ||
| ref.validate(entry) |
There was a problem hiding this comment.
optimization for read only transactions with a single ref to avoid locking
| } | ||
|
|
||
| // Read-only transaction: already validated, no locking needed | ||
| if !hasWrites then boundary.break(true) |
There was a problem hiding this comment.
optimization for read only transactions with multiple refs, avoiding locking as well
| true | ||
| } | ||
| // No try/finally needed - boundary.break returns from the block, not the method | ||
| buffer.clear() |
There was a problem hiding this comment.
try/catch would be safer but there's some cost to installing the interrupt handler and the code is safe
| final private class TRefImpl[A] private[kyo] (initialState: Write[A]) | ||
| extends AtomicInteger(0) // Atomic super class to keep the lock state | ||
| final private class TRefImpl[A] private[kyo] (initEntry: Write[A]) | ||
| extends TRef.State.Owner |
There was a problem hiding this comment.
Instead of extending from AtomicInteger, extend from TRef.State.Owner, which then extends from AtomicLong to pack both the read tick and the lock state in a single long. The bit packing is abstracted away via TRef.State.
| import TRef.State.* | ||
|
|
||
| private[kyo] def state(using AllowUnsafe): Write[A] = currentState | ||
| @volatile private var _entry = initEntry |
There was a problem hiding this comment.
renamed the previous state methods/parameter to entry to avoid confusion
| // Early retry if the TRef is concurrently modified | ||
| Tick.withCurrent { tick => | ||
| val e = _entry | ||
| if e.tick.value > tick.value || getState().readTick > tick.value then |
There was a problem hiding this comment.
early retry optimization if the ref being written has a more recent read
| // Value-based fallback only for reads: if the same reference was written | ||
| // back, the read is still valid (reduces spurious aborts). Not safe for | ||
| // writes since two transactions computing the same value must not both commit. | ||
| case read: Read[?] => current.value.asInstanceOf[AnyRef].eq(read.value.asInstanceOf[AnyRef]) |
There was a problem hiding this comment.
if the entry is a read, it doesn't matter if the tick is different in case the value is the same. I'm using eq since == could be expensive and have to calculate the hashcode
| lockState == 0 && (super.compareAndSet(lockState, Int.MaxValue) || loop()) | ||
| end match | ||
| case _: Read[?] => s.acquireReader.exists(next => casState(s, next) || loop()) | ||
| case _: Write[?] => s.acquireWriter(tick.value).exists(next => casState(s, next) || loop()) |
There was a problem hiding this comment.
I love how Scala 3 is able to keep @tailrec with the lamba. This happens because Maybe.exists is inline
|
I'm planning to on the test flakiness in main and on benchmarking in a separate PR |
| var buffer = bufferCache.get() | ||
| if buffer == null then | ||
| buffer = new ArrayList[Any] | ||
| bufferCache.set(buffer) |
There was a problem hiding this comment.
It might be nice to push the getting/refilling this into a method in the ThreadLocal itself.
There was a problem hiding this comment.
There's ThreadLocal.withInitial on the JVM but JS doesn't have the stub for it
| package kyo | ||
|
|
||
| /** Monotonic tick value for STM conflict detection */ | ||
| private[kyo] opaque type Tick = Long |
There was a problem hiding this comment.
?
| private[kyo] opaque type Tick = Long | |
| private[kyo] opaque type Tick <: Long = Long |
There was a problem hiding this comment.
Good idea! I'm including in the next PR
| Maybe.when((self & LockMask) == 0 && self.readTick <= tick)((self & ~LockMask) | WriteLock) | ||
|
|
||
| // Display | ||
| inline def asString: String = |
There was a problem hiding this comment.
?
| inline def asString: String = | |
| inline def render: String = |
There was a problem hiding this comment.
I'm including this in a follow up PR
| // Write lock requires no existing locks | ||
| lockState == 0 && (super.compareAndSet(lockState, Int.MaxValue) || loop()) | ||
| end match | ||
| case _: Read[?] => s.acquireReader.exists(next => casState(s, next) || loop()) |
There was a problem hiding this comment.
Should the loop() be inside exists?
There was a problem hiding this comment.
Yes, this is the cas retry loop. If State allows the acquireReader transition by returning a Present but casState fails, which indicates a concurrent write, it loops
A few optimizations of the STM impl. Please check the comments for details.