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/.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..3260a58 100644 --- a/threading.nimble +++ b/threading.nimble @@ -1,10 +1,11 @@ # 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" # 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 f9d7529..1814373 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..." @@ -100,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 # ------------------------------------------------------------------------------ @@ -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,9 +161,21 @@ 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 # ------------------------------------------------------------------------------ +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))) @@ -162,21 +189,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 +449,190 @@ 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() + + +# Receiver operations +# ------------------------------------------------------------------------------ -proc recv*[T](c: Chan[T], dst: var T) {.inline.} = - ## Receives a message from the channel `c` and fill `dst` with its value. +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. 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)[]) -proc recv*[T](c: Chan[T]): T {.inline.} = - ## Receives a message from the channel. - ## A version of `recv`_ that returns the message. - discard channelReceive(c.d, result.addr, sizeof(T), true) + channelReceive(r.d, dst.addr, sizeof(T), false, destroy) -proc recvIso*[T](c: Chan[T]): Isolated[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. + + # 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: + const tmp = default(T) + copyMem(result.addr, tmp.addr, sizeof(T)) + +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. + # `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`. ## 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. + # `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.} = + ## 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 +# ------------------------------------------------------------------------------ + +type ChannelPair*[T] = tuple[sender: Sender[T], receiver: Receiver[T]] -proc newChan*[T](elements: Positive = 30): Chan[T] = - ## An initialization procedure, necessary for acquiring resources and - ## initializing internal state of the channel. +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.}