scoped memory management (#14790)
* fixes the regressions * closes #13936 * scope based memory management implemented * enabled tcontrolflow.nim test case * final cleanups
This commit is contained in:
parent
695154970d
commit
1854d29781
17 changed files with 611 additions and 477 deletions
154
tests/arc/tweave.nim
Normal file
154
tests/arc/tweave.nim
Normal file
|
|
@ -0,0 +1,154 @@
|
|||
discard """
|
||||
outputsub: '''Success'''
|
||||
cmd: '''nim c --gc:arc --threads:on $file'''
|
||||
disabled: "bsd"
|
||||
"""
|
||||
|
||||
# bug #13936
|
||||
|
||||
import std/atomics
|
||||
|
||||
const MemBlockSize = 256
|
||||
|
||||
type
|
||||
ChannelSPSCSingle* = object
|
||||
full{.align: 128.}: Atomic[bool]
|
||||
itemSize*: uint8
|
||||
buffer*{.align: 8.}: UncheckedArray[byte]
|
||||
|
||||
proc `=`(
|
||||
dest: var ChannelSPSCSingle,
|
||||
source: ChannelSPSCSingle
|
||||
) {.error: "A channel cannot be copied".}
|
||||
|
||||
proc initialize*(chan: var ChannelSPSCSingle, itemsize: SomeInteger) {.inline.} =
|
||||
## If ChannelSPSCSingle is used intrusive another data structure
|
||||
## be aware that it should be the last part due to ending by UncheckedArray
|
||||
## Also due to 128 bytes padding, it automatically takes half
|
||||
## of the default MemBlockSize
|
||||
assert itemsize.int in 0 .. int high(uint8)
|
||||
assert itemSize.int +
|
||||
sizeof(chan.itemsize) +
|
||||
sizeof(chan.full) < MemBlockSize
|
||||
|
||||
chan.itemSize = uint8 itemsize
|
||||
chan.full.store(false, moRelaxed)
|
||||
|
||||
func isEmpty*(chan: var ChannelSPSCSingle): bool {.inline.} =
|
||||
not chan.full.load(moAcquire)
|
||||
|
||||
func tryRecv*[T](chan: var ChannelSPSCSingle, dst: var T): bool {.inline.} =
|
||||
## Try receiving the item buffered in the channel
|
||||
## Returns true if successful (channel was not empty)
|
||||
##
|
||||
## ⚠ Use only in the consumer thread that reads from the channel.
|
||||
assert (sizeof(T) == chan.itemsize.int) or
|
||||
# Support dummy object
|
||||
(sizeof(T) == 0 and chan.itemsize == 1)
|
||||
|
||||
let full = chan.full.load(moAcquire)
|
||||
if not full:
|
||||
return false
|
||||
dst = cast[ptr T](chan.buffer.addr)[]
|
||||
chan.full.store(false, moRelease)
|
||||
return true
|
||||
|
||||
func trySend*[T](chan: var ChannelSPSCSingle, src: sink T): bool {.inline.} =
|
||||
## Try sending an item into the channel
|
||||
## Reurns true if successful (channel was empty)
|
||||
##
|
||||
## ⚠ Use only in the producer thread that writes from the channel.
|
||||
assert (sizeof(T) == chan.itemsize.int) or
|
||||
# Support dummy object
|
||||
(sizeof(T) == 0 and chan.itemsize == 1)
|
||||
|
||||
let full = chan.full.load(moAcquire)
|
||||
if full:
|
||||
return false
|
||||
cast[ptr T](chan.buffer.addr)[] = src
|
||||
chan.full.store(true, moRelease)
|
||||
return true
|
||||
|
||||
# Sanity checks
|
||||
# ------------------------------------------------------------------------------
|
||||
when isMainModule:
|
||||
|
||||
when not compileOption("threads"):
|
||||
{.error: "This requires --threads:on compilation flag".}
|
||||
|
||||
template sendLoop[T](chan: var ChannelSPSCSingle,
|
||||
data: sink T,
|
||||
body: untyped): untyped =
|
||||
while not chan.trySend(data):
|
||||
body
|
||||
|
||||
template recvLoop[T](chan: var ChannelSPSCSingle,
|
||||
data: var T,
|
||||
body: untyped): untyped =
|
||||
while not chan.tryRecv(data):
|
||||
body
|
||||
|
||||
type
|
||||
ThreadArgs = object
|
||||
ID: WorkerKind
|
||||
chan: ptr ChannelSPSCSingle
|
||||
|
||||
WorkerKind = enum
|
||||
Sender
|
||||
Receiver
|
||||
|
||||
template Worker(id: WorkerKind, body: untyped): untyped {.dirty.} =
|
||||
if args.ID == id:
|
||||
body
|
||||
|
||||
proc thread_func(args: ThreadArgs) =
|
||||
|
||||
# Worker RECEIVER:
|
||||
# ---------
|
||||
# <- chan
|
||||
# <- chan
|
||||
# <- chan
|
||||
#
|
||||
# Worker SENDER:
|
||||
# ---------
|
||||
# chan <- 42
|
||||
# chan <- 53
|
||||
# chan <- 64
|
||||
Worker(Receiver):
|
||||
var val: int
|
||||
for j in 0 ..< 10:
|
||||
args.chan[].recvLoop(val):
|
||||
# Busy loop, in prod we might want to yield the core/thread timeslice
|
||||
discard
|
||||
echo " Receiver got: ", val
|
||||
doAssert val == 42 + j*11
|
||||
|
||||
Worker(Sender):
|
||||
doAssert args.chan.full.load(moRelaxed) == false
|
||||
for j in 0 ..< 10:
|
||||
let val = 42 + j*11
|
||||
args.chan[].sendLoop(val):
|
||||
# Busy loop, in prod we might want to yield the core/thread timeslice
|
||||
discard
|
||||
echo "Sender sent: ", val
|
||||
|
||||
proc main() =
|
||||
echo "Testing if 2 threads can send data"
|
||||
echo "-----------------------------------"
|
||||
var threads: array[2, Thread[ThreadArgs]]
|
||||
|
||||
var chan = cast[ptr ChannelSPSCSingle](allocShared(MemBlockSize))
|
||||
chan[].initialize(itemSize = sizeof(int))
|
||||
|
||||
createThread(threads[0], thread_func, ThreadArgs(ID: Receiver, chan: chan))
|
||||
createThread(threads[1], thread_func, ThreadArgs(ID: Sender, chan: chan))
|
||||
|
||||
joinThread(threads[0])
|
||||
joinThread(threads[1])
|
||||
|
||||
freeShared(chan)
|
||||
|
||||
echo "-----------------------------------"
|
||||
echo "Success"
|
||||
|
||||
main()
|
||||
Loading…
Add table
Add a link
Reference in a new issue