Split channel into sender/receiver

.github/workflows/ci.yml

@@ -86,12 +86,6 @@ jobs:
             nim_branch: version-2-2
             builder: windows-2025
 
-          # The oldest line still built by nightlies, kept as the floor check.
-          - os: linux
-            cpu: amd64
-            nim_branch: version-2-0
-            builder: ubuntu-24.04
-
     name: '${{ matrix.target.os }}-${{ matrix.target.cpu }}-nim-${{ matrix.target.nim_branch }} (${{ matrix.branch }})'
     runs-on: ${{ matrix.target.builder }}
     steps:

.gitignore

@@ -3,3 +3,22 @@ nimblecache/
 htmldocs/
 tests/tsmartptrsleak
 tests/tchannels_simple
+/tests/tatomics
+/tests/tbarrier
+/tests/tchannels_close
+/tests/tchannels_closed
+/tests/tchannels_closed_drained
+/tests/tchannels_concurrent_close
+/tests/tchannels_cooperative
+/tests/tchannels_legacy
+/tests/tchannels_leak
+/tests/tchannels_singlebuf
+/tests/tonce
+/tests/tonce2
+/tests/trwlock
+/tests/trwlock1
+/tests/tsemaphore
+/tests/tsemaphore1
+/tests/tsmartptrs
+/tests/twaitgroups
+/tests/tchannels_queued_items

tests/nim.cfg

@@ -22,6 +22,7 @@
   --cc:clang
   @if not windows:
     --debugger:native
+    --define:tsan
     --passc:"-fsanitize=thread -fno-omit-frame-pointer -mno-omit-leaf-frame-pointer"
     --passl:"-fsanitize=thread -fno-omit-frame-pointer -mno-omit-leaf-frame-pointer"
   @end

tests/tchannels_close.nim

@@ -0,0 +1,81 @@
+## Tests the close semantics of channels: when the last sender is destroyed,
+## the channel is closed for sending and receivers drain the remaining
+## messages then start receiving default(T). Symmetrically, when the last
+## receiver is destroyed, the channel is closed for receiving and senders
+## (including blocked ones) are woken up and their sends fail.
+
+discard """
+  matrix: "--threads:on --gc:orc; --threads:on --gc:arc"
+  disabled: "freebsd"
+"""
+import threading/channels
+import std/[os, isolation]
+
+block drain_then_default:
+  # A receiver created after the last sender is gone drains the queue
+  # and then receives default(T).
+  var receiver: Receiver[int]
+  block:
+    var (sender, r) = newChan[int]()
+    discard sender.send(1)
+    discard sender.send(2)
+    receiver = r
+  # Both senders are gone here: the channel is closed for sending.
+  doAssert receiver.recv() == 1
+  doAssert receiver.recv() == 2
+  doAssert receiver.recv() == 0 # default(int)
+  doAssert receiver.recv() == 0 # still default(int)
+  var dst: int
+  doAssert not receiver.tryRecv(dst)
+  doAssert not receiver.recv(dst) # closed: returns false
+
+block last_sender_closes:
+  # Destroying one of two senders does not close the channel;
+  # destroying the last one does.
+  var receiver: Receiver[int]
+  block:
+    var (sender, r) = newChan[int]()
+    receiver = r
+    var s2 = sender # A second sender
+    discard s2.send(7)
+  doAssert receiver.recv() == 7
+  doAssert receiver.recv() == 0 # default(int)
+
+block close_wakes_blocked_receiver:
+  # A receiver blocked on recv is woken up when the last sender is
+  # destroyed while it is waiting, and receives default(T).
+  var receiver: Receiver[int]
+  var thr: Thread[Receiver[int]]
+  var got: int
+  proc worker(r: Receiver[int]) {.thread.} =
+    got = r.recv() # Blocks until the channel is closed
+  block:
+    var (sender, r) = newChan[int]()
+    receiver = r
+    createThread(thr, worker, receiver)
+    sleep(100) # Give the worker time to block on recv
+  # The last sender was destroyed above, closing the channel while the
+  # worker is blocked on recv.
+  thr.joinThread()
+  doAssert got == 0 # default(int)
+
+block close_wakes_blocked_sender:
+  # A sender blocked on send (full channel) is woken up when the last
+  # receiver is destroyed, and send returns false.
+  var sender: Sender[int]
+  var thr: Thread[Sender[int]]
+  var sent: bool
+  proc worker(s: Sender[int]) {.thread.} =
+    # The channel is full when the worker starts, so this blocks until
+    # the last receiver is destroyed.
+    sent = send(s, isolate(1))
+  block:
+    var (s, r) = newChan[int](elements = 1)
+    sender = s
+    discard s.send(0) # Fill the single-slot buffer
+    createThread(thr, worker, sender)
+    sleep(100) # Give the worker time to block on send
+  # The last receiver was destroyed above, closing the channel for
+  # receiving while the worker is blocked on send.
+  thr.joinThread()
+  doAssert not sent # send returned false: channel closed for receiving

tests/tchannels_closed.nim

@@ -0,0 +1,60 @@
+## Tests the `closed` procs: a channel reports closed for sending when the
+## last Sender is destroyed and closed for receiving when the last Receiver
+## is destroyed. Destroying one of several handles of a direction does not
+## close it, and once a direction is closed it stays closed.
+##
+## Handles are released early with `s = Sender[int]()` (assignment of a nil
+## handle), not `=destroy`: `=destroy` does not nil the handle's pointer, so
+## an explicit `=destroy` on a local would be run a second time by the
+## compiler at scope end and touch the channel again.
+
+discard """
+  matrix: "--threads:on --gc:orc; --threads:on --gc:arc"
+  disabled: "freebsd"
+"""
+import threading/channels
+import std/isolation
+
+block not_closed_initially:
+  let (s, r) = newChan[int]()
+  # Compile-time check that `closed` returns a bool
+  let sClosed: bool = s.closed()
+  let rClosed: bool = r.closed()
+  doAssert not sClosed
+  doAssert not rClosed
+
+block dup_keeps_direction_open:
+  var (s, r) = newChan[int]()
+  var s2 = s
+  var r2 = r
+  s = Sender[int]()
+  r = Receiver[int]()
+  # The copies keep both directions open
+  doAssert not s2.closed()
+  doAssert not r2.closed()
+  s2 = Sender[int]()
+  r2 = Receiver[int]()
+
+block closed_for_sending:
+  var (s, r) = newChan[int]()
+  var s2 = s
+  s = Sender[int]()
+  doAssert not s2.closed()  # s2 keeps the channel open for sending
+  s2 = Sender[int]()
+  doAssert r.closed()       # all senders are gone
+  # The receiver still works: an empty closed channel yields default(T)
+  doAssert r.recv() == 0
+  var dst: int
+  doAssert not r.recv(dst)
+  doAssert dst == 0
+  doAssert not r.tryRecv(dst)
+
+block closed_for_receiving:
+  var (s, r) = newChan[int]()
+  var r2 = r
+  r = Receiver[int]()
+  doAssert not r2.closed()  # r2 keeps the channel open for receiving
+  r2 = Receiver[int]()
+  doAssert s.closed()       # all receivers are gone
+  # Sending is refused once the channel is closed for receiving
+  doAssert not s.send(isolate(1))

tests/tchannels_closed_drained.nim

@@ -0,0 +1,40 @@
+## Regression test: once all senders are gone, the channel must not report
+## closed on the receiving side until the queued items have been drained:
+## readers must be able to read every sent item, and only then observe
+## `closed`.
+
+discard """
+  matrix: "--threads:on --gc:orc; --threads:on --gc:arc"
+  disabled: "freebsd"
+"""
+import threading/channels
+import std/isolation
+
+block closed_only_after_drained:
+  var (s, r) = newChan[int](elements = 10)
+  doAssert not r.closed()
+  for i in 1..5:
+    doAssert s.send(isolate(i))
+  # All senders are gone, but 5 items are still queued:
+  # the channel must not report closed yet
+  s = Sender[int]()
+  doAssert not r.closed()
+  var dst: int
+  for i in 1..5:
+    doAssert r.tryRecv(dst)
+    doAssert dst == i
+    doAssert r.closed() == (i == 5)  # closed only once fully drained
+  doAssert not r.tryRecv(dst)
+
+block readers_still_drain_all_items:
+  var (s, r) = newChan[int](elements = 10)
+  for i in 1..5:
+    doAssert s.send(isolate(i))
+  s = Sender[int]()
+  # Blocking recv yields every queued item in order, then default(T)
+  var dst: int
+  for i in 1..5:
+    doAssert r.recv(dst)
+    doAssert dst == i
+  doAssert not r.recv(dst)
+  doAssert dst == 0

tests/tchannels_concurrent_close.nim

@@ -0,0 +1,58 @@
+## Stress test: destroying the last Sender and the last Receiver concurrently
+## must not use the channel after it has been freed.
+##
+## Regression: the refcount decrement and the free used to happen outside the
+## channel lock, so one thread could free the channel while the other was
+## about to take the lock for the close broadcast (use-after-free, heap
+## corruption). The handles are passed to the threads as pointers so that
+## exactly one Sender and one Receiver exist at a time.
+
+discard """
+  matrix: "--threads:on --gc:orc; --threads:on --gc:arc"
+  disabled: "freebsd"
+"""
+import threading/channels
+import std/atomics
+
+when defined(tsan):
+  # TSan is slow and reports the race within the first iterations anyway
+  const Iterations = 20_000
+else:
+  const Iterations = 200_000
+
+var globalPair = newChan[int]()
+var startA: Atomic[int]
+var startB: Atomic[int]
+
+proc destroySender(p: ptr Sender[int]) {.thread.} =
+  while startA.load(moAcquire) == 0:
+    for k in 0..3: discard k
+  p[] = Sender[int]()   # =copy with a nil source -> =destroy(p[])
+
+proc destroyReceiver(p: ptr Receiver[int]) {.thread.} =
+  while startB.load(moAcquire) == 0:
+    for k in 0..3: discard k
+  p[] = Receiver[int]()
+
+var a: Thread[ptr Sender[int]]
+var b: Thread[ptr Receiver[int]]
+for i in 0..<Iterations:
+  startA.store(0, moRelaxed)
+  startB.store(0, moRelaxed)
+  createThread(a, destroySender, addr(globalPair.sender))
+  createThread(b, destroyReceiver, addr(globalPair.receiver))
+  # Give both threads time to reach their spin
+  for j in 0..<50:
+    if (j and 7) == 0: discard j
+  startA.store(1, moRelease)
+  # Start the receiver destructor with a variable phase offset so the two
+  # destructors sweep across each other's refcount/lock window
+  for j in 0..<(i and 199):
+    if (j and 3) == 0: discard j
+  startB.store(1, moRelease)
+  joinThread(a)
+  joinThread(b)
+  # The workers have nilled both handles (`p[] = Sender[int]()`), so
+  # assign a fresh channel (the compiler elides the temporary, leaving
+  # exactly one Sender and one Receiver)
+  globalPair = newChan[int]()

tests/tchannels_cooperative.nim

@@ -10,50 +10,51 @@ const
   sentmsg = "task sent"
 
 type
-  Payload = tuple[chan: Chan[int16], idx: int16]
+  Payload = tuple[sender: Sender[int16], idx: int16]
 
 var
   sentmessages = newSeqOfCap[string](NTasks)
   receivedmessages = newSeqOfCap[int16](NTasks)
 
 # A prototype of a task executing thread
-proc runner(tasksCh: Chan[Payload]) {.thread.} =
+proc runner(tasksCh: Receiver[Payload]) {.thread.} =
   var p: Payload
   while true:
-    tasksCh.recv(p) # Get a message from the main thread
+    discard tasksCh.recv(p) # Get a message from the main thread
     if p.idx == -1: break # Check for an ad hoc stop signal
     else:
       sleep(SleepDurationMS) # Hard work
-      p.chan.send(p.idx) # Notify a consumer
+      discard p.sender.send(p.idx) # Notify a consumer
 
 # A single thread receiving result from runner threads
-proc consumer(args: tuple[resultsCh: Chan[int16], tasks: int16]) {.thread.} =
+proc consumer(args: tuple[resultsCh: Receiver[int16], tasks: int16]) {.thread.} =
   var idx: int16
   for _ in 0..<args.tasks: # We know the number of tasks and wait for them all
-    args.resultsCh.recv(idx)
+    discard args.resultsCh.recv(idx)
     {.gcsafe.}: # Don't do this. Here we know it's an exclusive access
       receivedmessages.add(idx) # Store which task was completed
 
 proc main(chanSize: Natural) =
   sentmessages.setLen(0)
   receivedmessages.setLen(0)
   var
-    taskThreads = newSeq[Thread[Chan[Payload]]](countProcessors())
-    tasksCh = newChan[Payload](chanSize)
-    consumerTh: Thread[(Chan[int16], int16)]
-    resultsCh = newChan[int16](chanSize)
+    taskThreads = newSeq[Thread[Receiver[Payload]]](countProcessors())
+    consumerTh: Thread[tuple[resultsCh: Receiver[int16], tasks: int16]]
+  let
+    (tasksSender, tasksReceiver) = newChan[Payload](chanSize)
+    (resultsSender, resultsReceiver) = newChan[int16](chanSize)
 
   # Consumer must be ready first to not block
-  createThread(consumerTh, consumer, (resultsCh, NTasks))
+  createThread(consumerTh, consumer, (resultsReceiver, NTasks))
   # Start runner threads
-  for i in 0..high(taskThreads): createThread(taskThreads[i], runner, tasksCh)
+  for i in 0..high(taskThreads): createThread(taskThreads[i], runner, tasksReceiver)
   # Loop iterating fake data
   for idx in 0'i16..<NTasks:
-    tasksCh.send((resultsCh, idx))
+    discard tasksSender.send((resultsSender, idx))
     sentmessages.add(sentmsg)
 
   for _ in taskThreads: # Stopping worker threads
-    tasksCh.send((resultsCh, -1'i16)) # A thread can't get more than 1 stop signal
+    discard tasksSender.send((resultsSender, -1'i16)) # A thread can't get more than 1 stop signal
   joinThreads(taskThreads)
   joinThread(consumerTh)
 

tests/tchannels_leak.nim

@@ -0,0 +1,46 @@
+## Regression test: destroying all handles of a channel (including copies)
+## must free the underlying channel. Measured via the process RSS growth:
+## a leaked channel keeps its buffer alive, so the growth over many
+## create/destroy cycles is proportional to the number of leaked channels.
+##
+## Handles are released with `s = Sender[T]()` (assignment of a nil handle),
+## not `=destroy`: `=destroy` does not nil the handle's pointer, so an
+## explicit `=destroy` on a local would be run a second time by the compiler
+## at scope end and touch the channel again.
+
+discard """
+  matrix: "--threads:on --gc:orc; --threads:on --gc:arc"
+  disabled: "freebsd, windows"
+"""
+import threading/channels
+import std/strutils
+
+when defined(linux):
+  proc rssKb(): int64 =
+    for line in lines("/proc/self/status"):
+      if line.len > 5 and line[0..4] == "VmRSS":
+        return parseInt(line.split(":")[1].strip().split()[0])
+
+  when defined(tsan):
+    const N = 20_000
+  else:
+    const N = 100_000
+
+  # Allow 1 kB of growth per channel: the old leak cost ~2.2 kB per channel
+  # (so it fails with a 2x margin), a leak-free build grows by a few MB
+  # total (so it passes with a wide margin).
+  const MaxGrowthKb = N
+
+  let before = rssKb()
+  for i in 0..<N:
+    var (s, r) = newChan[string](elements = 100)
+    var s2 = s
+    var r2 = r
+    s2.reset()
+    r2.reset()
+    s.reset()
+    r.reset()
+  let after = rssKb()
+  doAssert after - before < MaxGrowthKb,
+    "channel leak: RSS grew by " & $((after - before) * 1024) &
+    " bytes over " & $N & " channels"