Alloc: Pool MemRegion, split foreign queues

lib/system.nim

@@ -1154,7 +1154,19 @@ template sysAssert(cond: bool, msg: string) =
       cstderr.rawWrite "\n"
       rawQuit 1
 
-const hasAlloc = (hostOS != "standalone" or not defined(nogc)) and not defined(nimscript)
+const
+  hasAlloc = (hostOS != "standalone" or not defined(nogc)) and not defined(nimscript)
+  hasDefaultAllocator =
+    hasAlloc and
+    not (defined(useNimRtl) or defined(useMalloc) or defined(gcRegions) or
+         defined(nogc) or defined(boehmgc) or defined(gogc))
+  hasThreadLocalAllocator =
+    hasDefaultAllocator and hasThreadSupport and defined(gcDestructors)
+
+when hasThreadLocalAllocator:
+  # threadimpl is included before mmdisp provides these implementations.
+  proc initThreadAllocator() {.gcsafe, raises: [].}
+  proc releaseThreadAllocator() {.gcsafe, raises: [].}
 
 when notJSnotNims and hasAlloc and not defined(nimSeqsV2):
   proc addChar(s: NimString, c: char): NimString {.compilerproc, gcsafe.}
@@ -2425,6 +2437,8 @@ when notJSnotNims and hasAlloc:
   {.push profiler: off.}
   include "system/mmdisp"
   {.pop.}
+  when hasThreadLocalAllocator:
+    initThreadAllocator()
   {.push stackTrace: off, profiler: off.}
   when not defined(nimSeqsV2):
     include "system/sysstr"

lib/system/arc.nim

@@ -312,13 +312,17 @@ when not (defined(gcOrc) or defined(gcYrc)):
     ## Forces a full garbage collection pass. With `--mm:arc` a nop.
     discard
 
-template setupForeignThreadGc* =
-  ## With `--mm:arc` a nop.
-  discard
-
-template tearDownForeignThreadGc* =
-  ## With `--mm:arc` a nop.
-  discard
+when not hasThreadSupport:
+  template setupForeignThreadGc* = discard
+  template tearDownForeignThreadGc* = discard
+elif emulatedThreadVars:
+  template setupForeignThreadGc* =
+    {.error: "setupForeignThreadGc is available only when ``--threads:on`` and ``--tlsEmulation:off`` are used".}
+  template tearDownForeignThreadGc* =
+    {.error: "tearDownForeignThreadGc is available only when ``--threads:on`` and ``--tlsEmulation:off`` are used".}
+elif not hasThreadLocalAllocator:
+  template setupForeignThreadGc* = discard
+  template tearDownForeignThreadGc* = discard
 
 proc isObjDisplayCheck(source: PNimTypeV2, targetDepth: int16, token: uint32): bool {.compilerRtl, inl.} =
   result = targetDepth <= source.depth and source.display[targetDepth] == token

lib/system/threadimpl.nim

@@ -19,6 +19,13 @@ when not defined(useNimRtl):
 
       threadType = ThreadType.NimThread
 
+when hasThreadLocalAllocator and not emulatedThreadVars:
+  proc setupForeignThreadGc*() {.gcsafe, raises: [].} =
+    initThreadAllocator()
+
+  proc tearDownForeignThreadGc*() {.gcsafe, raises: [].} =
+    releaseThreadAllocator()
+
 when defined(gcDestructors):
   proc deallocThreadStorage(p: pointer) = c_free(p)
 else:
@@ -83,6 +90,8 @@ else:
         deallocThreadStorage(thrd.rawStack)
 
 proc threadProcWrapStackFrame[TArg](thrd: ptr Thread[TArg]) {.raises: [].} =
+  when hasThreadLocalAllocator:
+    initThreadAllocator()
   when defined(boehmgc):
     boehmGC_call_with_stack_base(threadProcWrapDispatch[TArg], thrd)
   elif not defined(nogc) and not defined(gogc) and not defined(gcRegions) and not usesDestructors:
@@ -97,6 +106,8 @@ proc threadProcWrapStackFrame[TArg](thrd: ptr Thread[TArg]) {.raises: [].} =
     when declared(deallocOsPages): deallocOsPages()
   else:
     threadProcWrapDispatch(thrd)
+    when hasThreadLocalAllocator:
+      releaseThreadAllocator()
 
 template nimThreadProcWrapperBody*(closure: untyped): untyped =
   var thrd = cast[ptr Thread[TArg]](closure)

tests/threads/tthreadallocatorforeignpool.nim

@@ -0,0 +1,59 @@
+discard """
+  matrix: "--mm:arc --threads:on --tlsEmulation:off; --mm:orc --threads:on --tlsEmulation:off"
+  disabled: "windows"
+  output: "ok"
+  timeout: "30"
+"""
+
+import std/posix
+
+var
+  escaped: pointer
+  reused: pointer
+
+proc allocateOnForeignThread(_: pointer): pointer {.noconv.} =
+  setupForeignThreadGc()
+  escaped = allocShared(96)
+  cast[ptr int](escaped)[] = 73
+  tearDownForeignThreadGc()
+  result = nil
+
+proc reuseOnForeignThread(_: pointer): pointer {.noconv.} =
+  setupForeignThreadGc()
+  doAssert cast[ptr int](escaped)[] == 73
+  deallocShared(escaped)
+  reused = allocShared(96)
+  doAssert reused == escaped
+  deallocShared(reused)
+  tearDownForeignThreadGc()
+  result = nil
+
+proc consumeDeferredFree(_: pointer): pointer {.noconv.} =
+  setupForeignThreadGc()
+  let first = allocShared(96)
+  let second = allocShared(96)
+  # The first allocation advances the active chunk and collects its deferred
+  # foreign frees. The next allocation reuses the remotely returned cell.
+  doAssert second == escaped
+  deallocShared(first)
+  deallocShared(second)
+  tearDownForeignThreadGc()
+  result = nil
+
+proc run(worker: proc(_: pointer): pointer {.noconv.}) =
+  var thread: Pthread
+  doAssert pthread_create(addr thread, nil, worker, nil) == 0
+  doAssert pthread_join(thread, nil) == 0
+
+# setup/teardown is the checkout/return boundary. A distinct native thread can
+# safely inherit the allocator even while one of its allocations is still live.
+run(allocateOnForeignThread)
+run(reuseOnForeignThread)
+
+# A free that arrives while the allocator is idle is queued on its handle and
+# consumed after that allocator is handed to another foreign thread.
+run(allocateOnForeignThread)
+deallocShared(escaped)
+run(consumeDeferredFree)
+
+echo "ok"

tests/threads/tthreadallocatorhandoffrace.nim

@@ -0,0 +1,64 @@
+discard """
+  matrix: "--mm:arc --threads:on; --mm:orc --threads:on"
+  output: "ok"
+  timeout: "30"
+"""
+
+import std/[atomics, typedthreads]
+
+const
+  pointerCount = 512
+  drainCount = 2048
+  iterations {.intdefine.} = 200
+  sizes = [16, 64, 4000, 4096, 8192]
+
+var
+  pointers: array[pointerCount, pointer]
+  mayExit: Atomic[bool]
+
+proc owner() {.thread.} =
+  for i in 0..<pointers.len:
+    let size = sizes[i mod sizes.len]
+    pointers[i] = allocShared(size)
+    cast[ptr byte](pointers[i])[] = byte(i)
+
+proc borrower() {.thread.} =
+  while not mayExit.load(moAcquire):
+    discard
+
+proc drain() {.thread.} =
+  var drained: array[drainCount, pointer]
+  for i in 0..<drained.len:
+    drained[i] = allocShared(sizes[i mod sizes.len])
+  for p in drained:
+    deallocShared(p)
+  let occupied = getOccupiedMem()
+  doAssert occupied == 0, "allocator retained " & $occupied & " bytes"
+
+for _ in 0..<iterations:
+  # The owner retires with live small and big allocations. The borrower checks
+  # out that region while this already-running main thread returns the cells.
+  # This races foreign queue publication against both directions of the
+  # MemRegion handoff without creating an unbounded number of regions.
+  block:
+    var thread: Thread[void]
+    createThread(thread, owner)
+    joinThread(thread)
+
+  mayExit.store(false, moRelaxed)
+  var borrowerThread: Thread[void]
+  createThread(borrowerThread, borrower)
+  for i, p in pointers:
+    if i == pointers.len div 2:
+      # Let the borrower tear the allocator down while the second half of the
+      # foreign publications are still in flight.
+      mayExit.store(true, moRelease)
+    deallocShared(p)
+  joinThread(borrowerThread)
+
+  block:
+    var thread: Thread[void]
+    createThread(thread, drain)
+    joinThread(thread)
+
+echo "ok"

tests/threads/tthreadallocatorpool.nim

@@ -0,0 +1,92 @@
+discard """
+  matrix: "--mm:arc --threads:on; --mm:orc --threads:on"
+  output: "ok"
+  timeout: "30"
+"""
+
+import std/[atomics, typedthreads]
+
+const concurrentThreads = 4
+
+var
+  escaped: pointer
+  reused: pointer
+  bigEscaped: pointer
+  roundAddresses: array[2, array[concurrentThreads, pointer]]
+  ready: Atomic[int]
+  mayExit: Atomic[bool]
+
+proc allocateEscaped() {.thread.} =
+  escaped = allocShared(64)
+  cast[ptr int](escaped)[] = 42
+
+proc consumeAfterHandoff() {.thread.} =
+  doAssert cast[ptr int](escaped)[] == 42
+  deallocShared(escaped)
+  reused = allocShared(64)
+  doAssert reused == escaped
+  deallocShared(reused)
+
+# A live allocation can outlast its original thread. The next thread receives
+# the same allocator and its stable handle makes the deallocation local again.
+block:
+  var thread: Thread[void]
+  createThread(thread, allocateEscaped)
+  joinThread(thread)
+  createThread(thread, consumeAfterHandoff)
+  joinThread(thread)
+
+proc allocateBigEscaped() {.thread.} =
+  bigEscaped = allocShared(8192)
+  cast[ptr int](bigEscaped)[] = 91
+
+proc consumeBigAfterHandoff() {.thread.} =
+  doAssert cast[ptr int](bigEscaped)[] == 91
+  deallocShared(bigEscaped)
+  let p = allocShared(8192)
+  doAssert p == bigEscaped
+  deallocShared(p)
+
+# Big chunks use a separate deferred-free queue on the stable handle.
+block:
+  var thread: Thread[void]
+  createThread(thread, allocateBigEscaped)
+  joinThread(thread)
+  createThread(thread, consumeBigAfterHandoff)
+  joinThread(thread)
+
+proc allocateConcurrently(arg: tuple[round, index: int]) {.thread.} =
+  let p = allocShared(80)
+  roundAddresses[arg.round][arg.index] = p
+  deallocShared(p)
+  discard ready.fetchAdd(1, moRelease)
+  while not mayExit.load(moAcquire):
+    discard
+
+proc runConcurrentRound(round: int) =
+  var threads: array[concurrentThreads, Thread[tuple[round, index: int]]]
+  ready.store(0, moRelaxed)
+  mayExit.store(false, moRelaxed)
+  for i in 0..<threads.len:
+    createThread(threads[i], allocateConcurrently, (round, i))
+  while ready.load(moAcquire) != concurrentThreads:
+    discard
+  mayExit.store(true, moRelease)
+  for thread in threads.mitems:
+    joinThread(thread)
+
+# The first round establishes the peak number of simultaneous allocators. The
+# following rounds must reuse those regions instead of reserving one region per
+# new thread.
+runConcurrentRound(0)
+for _ in 0..<32:
+  runConcurrentRound(1)
+  for p in roundAddresses[1]:
+    var found = false
+    for old in roundAddresses[0]:
+      if p == old:
+        found = true
+        break
+    doAssert found
+
+echo "ok"

tests/threads/tthreadallocatorpoolrace.nim

@@ -0,0 +1,56 @@
+discard """
+  matrix: "--mm:arc --threads:on; --mm:orc --threads:on"
+  output: "ok"
+  timeout: "30"
+"""
+
+import std/[atomics, typedthreads]
+
+const
+  pointerCount = 1024
+  iterations = 100
+
+var
+  pointers: array[pointerCount, pointer]
+  drainPointers: array[pointerCount, pointer]
+  ready: Atomic[bool]
+  start: Atomic[bool]
+  remoteDone: Atomic[bool]
+
+proc owner() {.thread.} =
+  for i in 0..<pointers.len:
+    pointers[i] = allocShared(16 + (i mod 8) * 16)
+  ready.store(true, moRelease)
+  while not start.load(moAcquire):
+    discard
+
+  # Race allocator activity against remote frees. The owner can consume cells
+  # while the remote thread is still publishing entries to its handle.
+  while not remoteDone.load(moAcquire):
+    for i in 0..<8:
+      let p = allocShared(16 + i * 16)
+      deallocShared(p)
+
+  # Exhaust local free lists so all remaining remote lists are consumed before
+  # this allocator is returned to the pool.
+  for i in 0..<drainPointers.len:
+    drainPointers[i] = allocShared(16 + (i mod 8) * 16)
+  for p in drainPointers:
+    deallocShared(p)
+  doAssert getOccupiedMem() == 0
+
+for _ in 0..<iterations:
+  ready.store(false, moRelaxed)
+  start.store(false, moRelaxed)
+  remoteDone.store(false, moRelaxed)
+  var thread: Thread[void]
+  createThread(thread, owner)
+  while not ready.load(moAcquire):
+    discard
+  start.store(true, moRelease)
+  for p in pointers:
+    deallocShared(p)
+  remoteDone.store(true, moRelease)
+  joinThread(thread)
+
+echo "ok"