From 5d654d44ac10c9f92d8d610b58b6fd739dd0764d Mon Sep 17 00:00:00 2001 From: Jacek Sieka Date: Fri, 18 Sep 2026 12:12:21 +0200 Subject: [PATCH 1/2] Split channel into sender/receiver By separating the handles responsible for the two sides of a channel, we get to support an intuitive api for closing channels that is safe by construction, ie there's no risk that senders write into a channel closed for sending and vice versa. Both receivers and senders can be copied and moved around so that they can each be shared by multiple threads. The principle of operation is simple: * senders and receivers have separate refcounts on the raw channel * with all senders gone, it means no more items will be pushed to the channel - all queued receivers are woken up and told it's over (once they've drained the channel) * with all receivers gone, it means no more messages can be pushed into the channel - we can free the buffer and sending becomes a no-op * with both gone, we release the channel resources. On the sending side, we return a `bool` for "no more receivers" - on the receiving side, we return `default(T)` or a `bool` for the var-taking version - both sides can check if the channel was closed to disambiguate. --- .gitignore | 19 ++ tests/nim.cfg | 1 + tests/tchannels_close.nim | 81 ++++++ tests/tchannels_closed.nim | 60 ++++ tests/tchannels_closed_drained.nim | 40 +++ tests/tchannels_concurrent_close.nim | 58 ++++ tests/tchannels_cooperative.nim | 29 +- tests/tchannels_leak.nim | 46 +++ tests/tchannels_legacy.nim | 63 ++++ tests/tchannels_queued_items.nim | 31 ++ tests/tchannels_simple.nim | 30 +- tests/tchannels_singlebuf.nim | 29 +- threading.nimble | 4 +- threading/channels.nim | 421 ++++++++++++++++++++------- 14 files changed, 761 insertions(+), 151 deletions(-) create mode 100644 tests/tchannels_close.nim create mode 100644 tests/tchannels_closed.nim create mode 100644 tests/tchannels_closed_drained.nim create mode 100644 tests/tchannels_concurrent_close.nim create mode 100644 tests/tchannels_leak.nim create mode 100644 tests/tchannels_legacy.nim create mode 100644 tests/tchannels_queued_items.nim diff --git a/.gitignore b/.gitignore index dc17cc5..9b4d107 100644 --- a/.gitignore +++ b/.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 diff --git a/tests/nim.cfg b/tests/nim.cfg index 4bc5d2d..d0aa7d2 100644 --- a/tests/nim.cfg +++ b/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 diff --git a/tests/tchannels_close.nim b/tests/tchannels_close.nim new file mode 100644 index 0000000..cae2b6d --- /dev/null +++ b/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 diff --git a/tests/tchannels_closed.nim b/tests/tchannels_closed.nim new file mode 100644 index 0000000..76a0bac --- /dev/null +++ b/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)) diff --git a/tests/tchannels_closed_drained.nim b/tests/tchannels_closed_drained.nim new file mode 100644 index 0000000..2d2bf2c --- /dev/null +++ b/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 diff --git a/tests/tchannels_concurrent_close.nim b/tests/tchannels_concurrent_close.nim new file mode 100644 index 0000000..1ebef31 --- /dev/null +++ b/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.. 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..= 2 block: - let chan0 = newChan[int]() - let chan1 = chan0 + let (s0, r0) = newChan[int]() + let s1 = s0 block: - let chan3 = chan0 - let chan4 = chan0 + let s2 = s0 + let s3 = s0 diff --git a/tests/tchannels_singlebuf.nim b/tests/tchannels_singlebuf.nim index c000841..012588e 100644 --- a/tests/tchannels_singlebuf.nim +++ b/tests/tchannels_singlebuf.nim @@ -8,27 +8,27 @@ const Message = "Hello" block trySend_recv: var attempts = 0 - proc test(chan: Chan[string]) {.thread.} = + proc test(s: Sender[string]) {.thread.} = var notSent = true let msg = Message while notSent: - notSent = not chan.trySend(msg) + notSent = not s.trySend(msg) if notSent: atomicInc(attempts) - var chan = newChan[string](elements = 1) + var (s, r) = newChan[string](elements = 1) # Fill the channel before spawning the thread - chan.send("Dummy message") + discard s.send("Dummy message") - var thread: Thread[Chan[string]] - createThread(thread, test, chan) + var thread: Thread[Sender[string]] + createThread(thread, test, s) sleep 10 # Receive the dummy message to make room for the real message - discard chan.recv() + discard r.recv() var dest: string - chan.recv(dest) + discard r.recv(dest) doAssert dest == Message thread.joinThread() @@ -38,24 +38,23 @@ block trySend_recv: block send_tryRecv: var attempts = 0 - proc test(chan: Chan[string]) {.thread.} = + proc test(r: Receiver[string]) {.thread.} = var notReceived = true var msg: string while notReceived: - notReceived = not chan.tryRecv(msg) + notReceived = not r.tryRecv(msg) if notReceived: atomicInc(attempts) doAssert msg == Message - var chan = newChan[string](elements = 1) + var (s, r) = newChan[string](elements = 1) - var thread: Thread[Chan[string]] - createThread(thread, test, chan) + var thread: Thread[Receiver[string]] + createThread(thread, test, r) sleep 10 let src = Message - chan.send(src) + discard s.send(src) thread.joinThread() doAssert attempts > 0, "tryRecv should have been attempted multiple times" - diff --git a/threading.nimble b/threading.nimble index 6c9d89e..375a214 100644 --- a/threading.nimble +++ b/threading.nimble @@ -1,7 +1,7 @@ # Package -version = "0.2.1" -author = "Araq" +version = "0.3.0" +author = "Threading contributors" description = "New threading primitives for --mm:arc/orc: atomics, channels, smart pointers and wait groups." license = "MIT" diff --git a/threading/channels.nim b/threading/channels.nim index f9d7529..af0b1b4 100644 --- a/threading/channels.nim +++ b/threading/channels.nim @@ -19,17 +19,33 @@ ## ## .. warning:: This module is experimental and its interface may change. ## -## This module implements multi-producer multi-consumer channels - a concurrency -## primitive with a high-level interface intended for communication and -## synchronization between threads. It allows sending and receiving typed, isolated +## This module implements multi-producer multi-consumer bounded channels - a +## concurrency primitive with a high-level interface intended for communication +## and synchronization between threads. It allows sending and receiving typed, isolated ## data, enabling safe and efficient concurrency. ## -## The `Chan` type represents a generic fixed-size channel object that internally manages -## the underlying resources and synchronization. It has to be initialized using -## the `newChan` proc. Sending and receiving operations are provided by the -## blocking `send` and `recv` procs, and non-blocking `trySend` and `tryRecv` -## procs. Send operations add messages to the channel, receiving operations -## remove them. +## A channel is created as a pair of a `Sender` and a `Receiver` using the +## `newChan` proc. +## +## Both are reference-counted handles to the same underlying fixed-size +## channel, and they can be copied and shared between threads. +## +## Sending operations are provided by the blocking `send` proc and the +## non-blocking `trySend` and `tryTake` procs on the `Sender`. +## +## Receiving operations are provided by the blocking `recv` proc and the +## non-blocking `tryRecv` proc on the `Receiver`. +## +## When the last `Sender` is destroyed, the channel is closed for sending: +## `Receiver`s drain any messages still in the queue and then start receiving +## `default(T)` values instead of sent messages. +## +## When the last `Receiver` is destroyed, the channel is closed for receiving: +## `send` calls on the remaining `Sender`s return `false` and blocked senders +## are woken up. +## +## The deprecated `Channel` API at the bottom of this module emulates +## the legacy single-handle channel on top of a `Sender`/`Receiver` pair. ## ## See also: ## * [std/isolation](https://nim-lang.org/docs/isolation.html) @@ -40,24 +56,24 @@ runnableExamples("--threads:on --gc:orc"): import std/os - # In this example a channel is declared at module scope. + # In this example a channel pair is declared at module scope. # Channels are generic, and they include support for passing objects between # threads. # Note that isolated data passed through channels is moved around. - var chan = newChan[string]() + var (sender, receiver) = newChan[string]() block example_blocking: # This proc will be run in another thread. - proc basicWorker() = - chan.send("Hello World!") + proc basicWorker(s: Sender[string]) = + discard s.send("Hello World!") # Launch the worker. - var worker: Thread[void] - createThread(worker, basicWorker) + var worker: Thread[Sender[string]] + createThread(worker, basicWorker, sender) # Block until the message arrives, then print it out. var dest = "" - dest = chan.recv() + discard receiver.recv(dest) assert dest == "Hello World!" # Wait for the thread to exit before moving on to the next example. @@ -66,14 +82,14 @@ runnableExamples("--threads:on --gc:orc"): block example_non_blocking: # This is another proc to run in a background thread. This proc takes a while # to send the message since it first sleeps for some time. - proc slowWorker(delay: Natural) = - # `delay` is a period in milliseconds - sleep(delay) - chan.send("Another message") + proc slowWorker(args: tuple[s: Sender[string], d: Natural]) = + # `args.d` is a delay period in milliseconds + sleep(args.d) + discard args.s.send("Another message") # Launch the worker with a delay set to 2 seconds (2000 ms). - var worker: Thread[Natural] - createThread(worker, slowWorker, 2000) + var worker: Thread[tuple[s: Sender[string], d: Natural]] + createThread(worker, slowWorker, (sender, 2000.Natural)) # This time, use a non-blocking approach with tryRecv. # Since the main thread is not blocked, it could be used to perform other @@ -81,7 +97,7 @@ runnableExamples("--threads:on --gc:orc"): var messages: seq[string] while true: var msg = "" - if chan.tryRecv(msg): + if receiver.tryRecv(msg): messages.add msg # "Another message" break messages.add "Pretend I'm doing useful work..." @@ -114,7 +130,9 @@ type head: Atomic[int] ## Write/enqueue/send index tail: Atomic[int] ## Read/dequeue/receive index buffer: ptr UncheckedArray[byte] - atomicCounter: Atomic[int] + refCount: Atomic[int] ## Total number of live Sender and Receiver handles + senders: Atomic[int] ## Number of live Sender handles + receivers: Atomic[int] ## Number of live Receiver handles # ------------------------------------------------------------------------------ @@ -130,9 +148,6 @@ proc setTail(chan: ChannelRaw, value: int, order: MemoryOrder = moRelaxed) {.inl proc setHead(chan: ChannelRaw, value: int, order: MemoryOrder = moRelaxed) {.inline.} = chan.head.store(value, order) -proc setAtomicCounter(chan: ChannelRaw, value: int, order: MemoryOrder = moRelaxed) {.inline.} = - chan.atomicCounter.store(value, order) - proc numItems(chan: ChannelRaw): int {.inline.} = result = chan.getHead() - chan.getTail() if result < 0: @@ -146,6 +161,12 @@ template isFull(chan: ChannelRaw): bool = template isEmpty(chan: ChannelRaw): bool = chan.getHead() == chan.getTail() +template closedForSending(chan: ChannelRaw): bool = + chan.senders.load(moRelaxed) == 0 + +template closedForReceiving(chan: ChannelRaw): bool = + chan.receivers.load(moRelaxed) == 0 + # Channels memory ops # ------------------------------------------------------------------------------ @@ -162,21 +183,41 @@ proc allocChannel(size, n: int): ChannelRaw = result.slots = n result.setHead(0) result.setTail(0) - result.setAtomicCounter(0) + result.refCount.store(0, moRelaxed) + result.senders.store(0, moRelaxed) + result.receivers.store(0, moRelaxed) proc freeChannel(chan: ChannelRaw) = - if chan.isNil: - return - - if not chan.buffer.isNil: - deallocShared(chan.buffer) - deinitLock(chan.lock) deinitCond(chan.spaceAvailableCV) deinitCond(chan.dataAvailableCV) deallocShared(chan) +proc freeSenders(chan: ChannelRaw) = + acquire(chan.lock) + broadcast(chan.dataAvailableCV) + release(chan.lock) + +proc freeReceivers[T](chan: ChannelRaw) = + acquire(chan.lock) + + # Because there are no more readers, there will be no more sending and no more + # reading - we can destroy the pending items and free the channel buffer + if not chan.buffer.isNil: + let n = chan.numItems() + for k in 0..`_. +template trySend*[T](s: Sender[T], src: T): bool = + ## Helper template for `trySend <#trySend,Sender[T],sinkIsolated[T]>`_. ## ## .. warning:: For repeated sends of the same value, consider using the - ## `tryTake <#tryTake,Chan[T],Isolated[T]>`_ proc with a pre-isolated + ## `tryTake <#tryTake,Sender[T],Isolated[T]>`_ proc with a pre-isolated ## value to avoid unnecessary copying. mixin isolate - trySend(c, isolate(src)) + trySend(s, isolate(src)) -proc tryTake*[T](c: Chan[T], src: var Isolated[T]): bool {.inline.} = - ## Tries to send the message `src` to the channel `c`. +proc tryTake*[T](s: Sender[T], src: var Isolated[T]): bool {.inline.} = + ## Tries to send the message `src` to the channel `s`. ## ## The memory of `src` is moved directly. Be careful not to reuse `src` afterwards. ## This proc is suitable when `src` cannot be copied. @@ -325,69 +432,173 @@ proc tryTake*[T](c: Chan[T], src: var Isolated[T]): bool {.inline.} = ## operations. ## ## Returns `false` if the message was not sent because the number of pending - ## messages in the channel exceeded its capacity. - result = channelSend(c.d, src.addr, sizeof(T), false) + ## messages in the channel exceeded its capacity, or because the channel is + ## closed for receiving (all receivers are gone). + result = channelSend(s.d, src.addr, sizeof(T), false) if result: wasMoved(src) -proc tryRecv*[T](c: Chan[T], dst: var T): bool {.inline.} = - ## Tries to receive a message from the channel `c` and fill `dst` with its value. - ## - ## Doesn't block waiting for messages in the channel to become available. - ## Instead returns after an attempt to receive a message was made. - ## - ## .. warning:: In high-concurrency situations, consider using an exponential - ## backoff strategy to reduce contention and improve the success rate of - ## operations. - ## - ## Returns `false` and does not change `dist` if no message was received. - channelReceive(c.d, dst.addr, sizeof(T), false) - -proc send*[T](c: Chan[T], src: sink Isolated[T]) {.inline.} = - ## Sends the message `src` to the channel `c`. +proc send*[T](s: Sender[T], src: sink Isolated[T]): bool {.inline.} = + ## Sends the message `src` to the channel `s`. ## This blocks the sending thread until `src` was successfully sent. ## - ## The memory of `src` is moved, not copied. + ## The memory of `src` is moved, not copied, even if the send fails. ## ## If the channel is already full with messages this will block the thread until ## messages from the channel are removed. + ## + ## Returns `false` if the channel is closed for receiving (all receivers are + ## gone) and the message was not sent. when defined(gcOrc) and defined(nimSafeOrcSend): GC_runOrc() - discard channelSend(c.d, src.addr, sizeof(T), true) + result = channelSend(s.d, src.addr, sizeof(T), true) wasMoved(src) -template send*[T](c: Chan[T]; src: T) = - ## Helper template for `send`. +template send*[T](s: Sender[T]; src: T): bool = + ## Helper template for `send <#send,Sender[T],sinkIsolated[T]>`_. + ## + ## Returns `false` if the channel is closed for receiving (all receivers are + ## gone) and the message was not sent. mixin isolate - send(c, isolate(src)) + send(s, isolate(src)) + +proc closed*[T](s: Sender[T]): bool {.inline.} = + ## Returns true if all receivers are gone and the channel is closed for + ## receiving. + ## + ## Once the channel is closed in either direction, that direction cannot be + ## reopened. + s.d.closedForReceiving() + -proc recv*[T](c: Chan[T], dst: var T) {.inline.} = - ## Receives a message from the channel `c` and fill `dst` with its value. +# Receiver operations +# ------------------------------------------------------------------------------ + +proc tryRecv*[T](r: Receiver[T], dst: var T): bool {.inline.} = + ## Tries to receive a message from the channel `r` and fill `dst` with its value. ## - ## This blocks the receiving thread until a message was successfully received. + ## Doesn't block waiting for messages in the channel to become available. + ## Instead returns after an attempt to receive a message was made. ## - ## If the channel does not contain any messages this will block the thread until - ## a message get sent to the channel. - discard channelReceive(c.d, dst.addr, sizeof(T), true) + ## .. warning:: In high-concurrency situations, consider using an exponential + ## backoff strategy to reduce contention and improve the success rate of + ## operations. + ## + ## Returns `false` and does not change `dst` if no message was received, + ## including when the channel is closed and fully drained. + channelReceive(r.d, dst.addr, sizeof(T), false) -proc recv*[T](c: Chan[T]): T {.inline.} = - ## Receives a message from the channel. +proc recv*[T](r: Receiver[T], dst: var T): bool {.inline.} = + ## Receives a message from the channel `r` and fill `dst` with its value. + ## + ## This blocks the receiving thread until a message was successfully received + ## or the channel is closed for sending. + ## + ## If the channel does not contain any messages this will block the thread until + ## a message gets sent to the channel. + ## + ## Returns `true` if a message was received. If the channel is closed for + ## sending (all senders are gone) and fully drained, `dst` is filled with + ## `default(T)` and `false` is returned. + result = channelReceive(r.d, dst.addr, sizeof(T), true) + if not result: + dst = default(T) + +proc recv*[T](r: Receiver[T]): T {.inline.} = + ## Receives a message from the channel `r`. ## A version of `recv`_ that returns the message. - discard channelReceive(c.d, result.addr, sizeof(T), true) + ## + ## If the channel is closed (all senders are gone) and fully drained, + ## `default(T)` is returned. + if not channelReceive(r.d, result.addr, sizeof(T), true): + result = default(T) -proc recvIso*[T](c: Chan[T]): Isolated[T] {.inline.} = - ## Receives a message from the channel. +proc recvIso*[T](r: Receiver[T]): Isolated[T] {.inline.} = + ## Receives a message from the channel `r`. ## A version of `recv`_ that returns the message and isolates it. - discard channelReceive(c.d, result.addr, sizeof(T), true) + ## + ## If the channel is closed (all senders are gone) and fully drained, + ## `default(T)` is returned. + if not channelReceive(r.d, result.addr, sizeof(T), true): + result = isolate(default(T)) + +proc closed*[T](s: Receiver[T]): bool {.inline.} = + ## Returns true if all senders are gone and the channel is closed for + ## sending, and all messages that were sent have been received (the queue + ## is fully drained). + ## + ## Once the channel is closed in either direction, that direction cannot be + ## reopened. + s.d.closedForSending() and s.d.isEmpty() -proc peek*[T](c: Chan[T]): int {.inline.} = - ## Returns an estimation of the current number of messages held by the channel. - numItems(c.d) +# Channel pair +# ------------------------------------------------------------------------------ -proc newChan*[T](elements: Positive = 30): Chan[T] = - ## An initialization procedure, necessary for acquiring resources and - ## initializing internal state of the channel. +type ChannelPair*[T] = tuple[sender: Sender[T], receiver: Receiver[T]] + +proc newChan*[T](elements: Positive = 30): ChannelPair[T] = + ## Creates a new channel: a pair of a `Sender` and a `Receiver` connected to + ## the same underlying fixed-size channel. ## ## `elements` is the capacity of the channel and thus how many messages it can hold ## before it refuses to accept any further messages. - result = Chan[T](d: allocChannel(sizeof(T), elements)) + ## + ## When the last `Sender` is destroyed, the channel is closed for sending: + ## `Receiver`s drain any messages still in the queue and then start receiving + ## `default(T)` values instead of sent messages. Symmetrically, when the last + ## `Receiver` is destroyed, the channel is closed for receiving: `send` calls + ## on the remaining `Sender`s return `false` and blocked senders are woken up. + let raw = allocChannel(sizeof(T), elements) + raw.refCount.store(2, moRelaxed) + raw.senders.store(1, moRelaxed) + raw.receivers.store(1, moRelaxed) + result = (Sender[T](d: raw), Receiver[T](d: raw)) + +# Legacy API (deprecated) +# ------------------------------------------------------------------------------ + +{.push warning[Deprecated]: false.} +type Channel*[T] {.deprecated.} = ChannelPair[T] + +proc send*[T](c: Channel[T], src: sink Isolated[T]) {.deprecated, inline.} = + ## Deprecated in 0.3.0: use `send`_ on the `sender` half of the channel instead. + discard send(c.sender, src) + +template send*[T](c: Channel[T]; src: T) {.deprecated.} = + ## Deprecated in 0.3.0: use `send`_ on the `sender` half of the channel instead. + mixin isolate + discard c.sender.send(src) + +proc trySend*[T](c: Channel[T], src: sink Isolated[T]): bool {.deprecated, inline.} = + ## Deprecated in 0.3.0: use `trysend`_ on the `sender` half of the channel instead. + trySend(c.sender, src) + +template trySend*[T](c: Channel[T], src: T): bool {.deprecated.} = + ## Deprecated in 0.3.0: use `trysend`_ on the `sender` half of the channel instead. + mixin isolate + trySend(c.sender, isolate(src)) + +proc tryTake*[T](c: Channel[T], src: var Isolated[T]): bool {.deprecated, inline.} = + ## Deprecated in 0.3.0: use `trytake`_ on the `sender` half of the channel instead. + c.sender.tryTake(src) + +proc tryRecv*[T](c: Channel[T], dst: var T): bool {.deprecated, inline.} = + ## Deprecated in 0.3.0: use `tryrecv`_ on the `receiver` half of the channel instead. + c.receiver.tryRecv(dst) + +proc recv*[T](c: Channel[T], dst: var T) {.deprecated, inline.} = + ## Deprecated in 0.3.0: use `recv`_ on the `receiver` half of the channel instead. + discard c.receiver.recv(dst) + +proc recv*[T](c: Channel[T]): T {.deprecated, inline.} = + ## Deprecated in 0.3.0: use `recv`_ on the `receiver` half of the channel instead. + c.receiver.recv() + +proc recvIso*[T](c: Channel[T]): Isolated[T] {.deprecated, inline.} = + ## Deprecated in 0.3.0: use `recviso`_ on the `receiver` half of the channel instead. + c.receiver.recvIso() + +proc peek*[T](c: Channel[T]): int {.deprecated, inline.} = + ## Deprecated in 0.3.0: no replacement + c.sender.d.numItems() +{.pop.} From 8dbd1a99a32768b7a1e68617eead1821312afa11 Mon Sep 17 00:00:00 2001 From: Jacek Sieka Date: Mon, 19 Oct 2026 09:36:01 +0200 Subject: [PATCH 2/2] avoid memory leak with destructor call --- .github/workflows/ci.yml | 6 ---- tests/tchannels_leak_read.nim | 53 ++++++++++++++++++++++++++++++++++ threading.nimble | 3 +- threading/channels.nim | 54 ++++++++++++++++++++++++++++------- 4 files changed, 99 insertions(+), 17 deletions(-) create mode 100644 tests/tchannels_leak_read.nim diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7b8038d..da097c3 100644 --- a/.github/workflows/ci.yml +++ b/.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: diff --git a/tests/tchannels_leak_read.nim b/tests/tchannels_leak_read.nim new file mode 100644 index 0000000..60da67e --- /dev/null +++ b/tests/tchannels_leak_read.nim @@ -0,0 +1,53 @@ +## Regression test: receiving into a reused `dst` must release its previous +## value: `recv`/`tryRecv` overwrite `dst` with the received message, and the +## old contents must be destroyed (their finalizers must run), not silently +## dropped. +## +## The destructions are counted with a destructor on the payload type, which +## is deterministic and independent of the GC (works under arc and orc alike). + +discard """ + matrix: "--threads:on --gc:orc; --threads:on --gc:arc" +""" +import threading/channels +import std/isolation + +type Item = object + id: int + +var destroyedCount = 0 +proc `=destroy`(i: var Item) = + if i.id != 0: + inc(destroyedCount) + +const N = 10_000 + +block recv: + var (s, r) = newChan[Item]() + var v: Item + for i in 1..N: + doAssert s.send(isolate(Item(id: i))) + doAssert r.recv(v) + doAssert v.id == i + # Every overwrite of a live `v` destroyed the previous item (the first + # overwrite replaced the default value, which is not counted). + doAssert destroyedCount == N - 1, $destroyedCount & " " & $N + +block tryRecv: + # The `v` of the previous block was destroyed at the end of the block. + doAssert destroyedCount == N, $destroyedCount & " " & $N + var (s, r) = newChan[Item]() + var v = Item(id: 42) + # A failed tryRecv must leave `dst` untouched (in particular, not destroy it). + doAssert not r.tryRecv(v) + doAssert v.id == 42 + doAssert destroyedCount == N, $destroyedCount & " " & $N + for i in 1..N: + doAssert s.send(isolate(Item(id: i))) + doAssert r.tryRecv(v) + doAssert v.id == i + doAssert destroyedCount == 2 * N, $destroyedCount & " " & $N + +# Plus the scope-end destruction of the live `v` of this block (item N): +# the total is 2N + 1. +doAssert destroyedCount == 2 * N + 1, $destroyedCount & " " & $N diff --git a/threading.nimble b/threading.nimble index 375a214..3260a58 100644 --- a/threading.nimble +++ b/threading.nimble @@ -7,4 +7,5 @@ license = "MIT" # Dependencies -requires "nim >= 2.0.0" +# `createShared` unusable in earlier releases: https://github.com/nim-lang/Nim/issues/26014 +requires "nim >= 2.2.12" diff --git a/threading/channels.nim b/threading/channels.nim index af0b1b4..1814373 100644 --- a/threading/channels.nim +++ b/threading/channels.nim @@ -116,7 +116,7 @@ runnableExamples("--threads:on --gc:orc"): when not (defined(gcArc) or defined(gcOrc) or defined(gcAtomicArc) or defined(nimdoc)): {.error: "This module requires one of --mm:arc / --mm:atomicArc / --mm:orc compilation flags".} -import std/[locks, isolation, atomics] +import std/[locks, isolation, atomics, typetraits] # Channel # ------------------------------------------------------------------------------ @@ -170,6 +170,12 @@ template closedForReceiving(chan: ChannelRaw): bool = # Channels memory ops # ------------------------------------------------------------------------------ +type + # Only ever called with a callback that runs `=destroy` on the value, which + # does not trigger a GC cycle under the arc/orc/atomicArc memory management + # modes this module supports, so the indirect call is GC-safe. + DestroyCallback = proc(p: pointer) {.gcsafe.} ## Destroys the value at `p` + proc allocChannel(size, n: int): ChannelRaw = result = cast[ChannelRaw](allocShared(sizeof(ChannelObj))) @@ -260,7 +266,13 @@ proc channelSend(chan: ChannelRaw, data: pointer, size: int, blocking: static bo release(chan.lock) result = true -proc channelReceive(chan: ChannelRaw, data: pointer, size: int, blocking: static bool): bool = +proc channelReceive( + chan: ChannelRaw, + data: pointer, + size: int, + blocking: static bool, + destroyDst: DestroyCallback, +): bool = assert not chan.isNil assert not data.isNil @@ -290,6 +302,11 @@ proc channelReceive(chan: ChannelRaw, data: pointer, size: int, blocking: static else: chan.getTail() - chan.slots + # Destroy the previous value of `data` before overwriting it with the + # received message. + if destroyDst != nil: + destroyDst(data) + copyMem(data, chan.buffer[readIdx * size].addr, size) atomicInc(chan.tail) @@ -485,8 +502,16 @@ proc tryRecv*[T](r: Receiver[T], dst: var T): bool {.inline.} = ## operations. ## ## Returns `false` and does not change `dst` if no message was received, - ## including when the channel is closed and fully drained. - channelReceive(r.d, dst.addr, sizeof(T), false) + ## including when the channel is closed and fully drained. When a message is + ## received, the previous value of `dst` is destroyed before being + ## overwritten. + when supportsCopyMem(T): + channelReceive(r.d, dst.addr, sizeof(T), false, nil) + else: + proc destroy(p: pointer) = + `=destroy`(cast[ptr T](p)[]) + + channelReceive(r.d, dst.addr, sizeof(T), false, destroy) proc recv*[T](r: Receiver[T], dst: var T): bool {.inline.} = ## Receives a message from the channel `r` and fill `dst` with its value. @@ -500,18 +525,25 @@ proc recv*[T](r: Receiver[T], dst: var T): bool {.inline.} = ## Returns `true` if a message was received. If the channel is closed for ## sending (all senders are gone) and fully drained, `dst` is filled with ## `default(T)` and `false` is returned. - result = channelReceive(r.d, dst.addr, sizeof(T), true) + + # Because we unconditionally return an item, we can destroy the previous value + # of `dst` before overwriting it with the received message. + `=destroy`(dst) + result = channelReceive(r.d, dst.addr, sizeof(T), true, nil) if not result: - dst = default(T) + const tmp = default(T) + copyMem(result.addr, tmp.addr, sizeof(T)) -proc recv*[T](r: Receiver[T]): T {.inline.} = +proc recv*[T](r: Receiver[T]): T {.inline, noinit.} = ## Receives a message from the channel `r`. ## A version of `recv`_ that returns the message. ## ## If the channel is closed (all senders are gone) and fully drained, ## `default(T)` is returned. - if not channelReceive(r.d, result.addr, sizeof(T), true): - result = default(T) + # `result` is default-initialized, so no destructor callback is needed. + if not channelReceive(r.d, result.addr, sizeof(T), true, nil): + const tmp = default(T) + copyMem(result.addr, tmp.addr, sizeof(T)) proc recvIso*[T](r: Receiver[T]): Isolated[T] {.inline.} = ## Receives a message from the channel `r`. @@ -519,7 +551,8 @@ proc recvIso*[T](r: Receiver[T]): Isolated[T] {.inline.} = ## ## If the channel is closed (all senders are gone) and fully drained, ## `default(T)` is returned. - if not channelReceive(r.d, result.addr, sizeof(T), true): + # `result` is default-initialized, so no destructor callback is needed. + if not channelReceive(r.d, result.addr, sizeof(T), true, nil): result = isolate(default(T)) proc closed*[T](s: Receiver[T]): bool {.inline.} = @@ -601,4 +634,5 @@ proc recvIso*[T](c: Channel[T]): Isolated[T] {.deprecated, inline.} = proc peek*[T](c: Channel[T]): int {.deprecated, inline.} = ## Deprecated in 0.3.0: no replacement c.sender.d.numItems() + {.pop.}