ported channels to ARC
This commit is contained in:
parent
fbb8052e3a
commit
9faf00598f
2 changed files with 144 additions and 122 deletions
|
|
@ -3764,7 +3764,7 @@ when not defined(JS): #and not defined(nimscript):
|
||||||
when hasAlloc: include "system/strmantle"
|
when hasAlloc: include "system/strmantle"
|
||||||
|
|
||||||
when hasThreadSupport:
|
when hasThreadSupport:
|
||||||
when hostOS != "standalone" and not usesDestructors: include "system/channels"
|
when hostOS != "standalone": include "system/channels"
|
||||||
|
|
||||||
when not defined(nimscript) and hasAlloc:
|
when not defined(nimscript) and hasAlloc:
|
||||||
when not usesDestructors:
|
when not usesDestructors:
|
||||||
|
|
|
||||||
|
|
@ -25,7 +25,7 @@ when not declared(ThisIsSystem):
|
||||||
{.error: "You must not import this module explicitly".}
|
{.error: "You must not import this module explicitly".}
|
||||||
|
|
||||||
type
|
type
|
||||||
pbytes = ptr array[0.. 0xffff, byte]
|
pbytes = ptr UncheckedArray[byte]
|
||||||
RawChannel {.pure, final.} = object ## msg queue for a thread
|
RawChannel {.pure, final.} = object ## msg queue for a thread
|
||||||
rd, wr, count, mask, maxItems: int
|
rd, wr, count, mask, maxItems: int
|
||||||
data: pbytes
|
data: pbytes
|
||||||
|
|
@ -33,6 +33,7 @@ type
|
||||||
cond: SysCond
|
cond: SysCond
|
||||||
elemType: PNimType
|
elemType: PNimType
|
||||||
ready: bool
|
ready: bool
|
||||||
|
when not usesDestructors:
|
||||||
region: MemRegion
|
region: MemRegion
|
||||||
PRawChannel = ptr RawChannel
|
PRawChannel = ptr RawChannel
|
||||||
LoadStoreMode = enum mStore, mLoad
|
LoadStoreMode = enum mStore, mLoad
|
||||||
|
|
@ -52,14 +53,19 @@ proc deinitRawChannel(p: pointer) =
|
||||||
# we need to grab the lock to be safe against sending threads!
|
# we need to grab the lock to be safe against sending threads!
|
||||||
acquireSys(c.lock)
|
acquireSys(c.lock)
|
||||||
c.mask = ChannelDeadMask
|
c.mask = ChannelDeadMask
|
||||||
|
when not usesDestructors:
|
||||||
deallocOsPages(c.region)
|
deallocOsPages(c.region)
|
||||||
|
else:
|
||||||
|
if c.data != nil: deallocShared(c.data)
|
||||||
deinitSys(c.lock)
|
deinitSys(c.lock)
|
||||||
deinitSysCond(c.cond)
|
deinitSysCond(c.cond)
|
||||||
|
|
||||||
proc storeAux(dest, src: pointer, mt: PNimType, t: PRawChannel,
|
when not usesDestructors:
|
||||||
|
|
||||||
|
proc storeAux(dest, src: pointer, mt: PNimType, t: PRawChannel,
|
||||||
mode: LoadStoreMode) {.benign.}
|
mode: LoadStoreMode) {.benign.}
|
||||||
|
|
||||||
proc storeAux(dest, src: pointer, n: ptr TNimNode, t: PRawChannel,
|
proc storeAux(dest, src: pointer, n: ptr TNimNode, t: PRawChannel,
|
||||||
mode: LoadStoreMode) {.benign.} =
|
mode: LoadStoreMode) {.benign.} =
|
||||||
var
|
var
|
||||||
d = cast[ByteAddress](dest)
|
d = cast[ByteAddress](dest)
|
||||||
|
|
@ -76,7 +82,7 @@ proc storeAux(dest, src: pointer, n: ptr TNimNode, t: PRawChannel,
|
||||||
if m != nil: storeAux(dest, src, m, t, mode)
|
if m != nil: storeAux(dest, src, m, t, mode)
|
||||||
of nkNone: sysAssert(false, "storeAux")
|
of nkNone: sysAssert(false, "storeAux")
|
||||||
|
|
||||||
proc storeAux(dest, src: pointer, mt: PNimType, t: PRawChannel,
|
proc storeAux(dest, src: pointer, mt: PNimType, t: PRawChannel,
|
||||||
mode: LoadStoreMode) =
|
mode: LoadStoreMode) =
|
||||||
template `+!`(p: pointer; x: int): pointer =
|
template `+!`(p: pointer; x: int): pointer =
|
||||||
cast[pointer](cast[int](p) +% x)
|
cast[pointer](cast[int](p) +% x)
|
||||||
|
|
@ -183,7 +189,10 @@ proc rawSend(q: PRawChannel, data: pointer, typ: PNimType) =
|
||||||
if q.count >= cap:
|
if q.count >= cap:
|
||||||
# start with capacity for 2 entries in the queue:
|
# start with capacity for 2 entries in the queue:
|
||||||
if cap == 0: cap = 1
|
if cap == 0: cap = 1
|
||||||
|
when not usesDestructors:
|
||||||
var n = cast[pbytes](alloc0(q.region, cap*2*typ.size))
|
var n = cast[pbytes](alloc0(q.region, cap*2*typ.size))
|
||||||
|
else:
|
||||||
|
var n = cast[pbytes](allocShared0(cap*2*typ.size))
|
||||||
var z = 0
|
var z = 0
|
||||||
var i = q.rd
|
var i = q.rd
|
||||||
var c = q.count
|
var c = q.count
|
||||||
|
|
@ -192,19 +201,29 @@ proc rawSend(q: PRawChannel, data: pointer, typ: PNimType) =
|
||||||
copyMem(addr(n[z*typ.size]), addr(q.data[i*typ.size]), typ.size)
|
copyMem(addr(n[z*typ.size]), addr(q.data[i*typ.size]), typ.size)
|
||||||
i = (i + 1) and q.mask
|
i = (i + 1) and q.mask
|
||||||
inc z
|
inc z
|
||||||
if q.data != nil: dealloc(q.region, q.data)
|
if q.data != nil:
|
||||||
|
when not usesDestructors:
|
||||||
|
dealloc(q.region, q.data)
|
||||||
|
else:
|
||||||
|
deallocShared(q.data)
|
||||||
q.data = n
|
q.data = n
|
||||||
q.mask = cap*2 - 1
|
q.mask = cap*2 - 1
|
||||||
q.wr = q.count
|
q.wr = q.count
|
||||||
q.rd = 0
|
q.rd = 0
|
||||||
|
when not usesDestructors:
|
||||||
storeAux(addr(q.data[q.wr * typ.size]), data, typ, q, mStore)
|
storeAux(addr(q.data[q.wr * typ.size]), data, typ, q, mStore)
|
||||||
|
else:
|
||||||
|
copyMem(addr(q.data[q.wr * typ.size]), data, typ.size)
|
||||||
inc q.count
|
inc q.count
|
||||||
q.wr = (q.wr + 1) and q.mask
|
q.wr = (q.wr + 1) and q.mask
|
||||||
|
|
||||||
proc rawRecv(q: PRawChannel, data: pointer, typ: PNimType) =
|
proc rawRecv(q: PRawChannel, data: pointer, typ: PNimType) =
|
||||||
sysAssert q.count > 0, "rawRecv"
|
sysAssert q.count > 0, "rawRecv"
|
||||||
dec q.count
|
dec q.count
|
||||||
|
when not usesDestructors:
|
||||||
storeAux(data, addr(q.data[q.rd * typ.size]), typ, q, mLoad)
|
storeAux(data, addr(q.data[q.rd * typ.size]), typ, q, mLoad)
|
||||||
|
else:
|
||||||
|
copyMem(data, addr(q.data[q.rd * typ.size]), typ.size)
|
||||||
q.rd = (q.rd + 1) and q.mask
|
q.rd = (q.rd + 1) and q.mask
|
||||||
|
|
||||||
template lockChannel(q, action): untyped =
|
template lockChannel(q, action): untyped =
|
||||||
|
|
@ -231,18 +250,21 @@ proc sendImpl(q: PRawChannel, typ: PNimType, msg: pointer, noBlock: bool): bool
|
||||||
signalSysCond(q.cond)
|
signalSysCond(q.cond)
|
||||||
result = true
|
result = true
|
||||||
|
|
||||||
proc send*[TMsg](c: var Channel[TMsg], msg: TMsg) {.inline.} =
|
proc send*[TMsg](c: var Channel[TMsg], msg: sink TMsg) {.inline.} =
|
||||||
## Sends a message to a thread. `msg` is deeply copied.
|
## Sends a message to a thread. `msg` is deeply copied.
|
||||||
discard sendImpl(cast[PRawChannel](addr c), cast[PNimType](getTypeInfo(msg)), unsafeAddr(msg), false)
|
discard sendImpl(cast[PRawChannel](addr c), cast[PNimType](getTypeInfo(msg)), unsafeAddr(msg), false)
|
||||||
|
wasMoved(msg)
|
||||||
|
|
||||||
proc trySend*[TMsg](c: var Channel[TMsg], msg: TMsg): bool {.inline.} =
|
proc trySend*[TMsg](c: var Channel[TMsg], msg: sink TMsg): bool {.inline.} =
|
||||||
## Tries to send a message to a thread.
|
## Tries to send a message to a thread.
|
||||||
##
|
##
|
||||||
## `msg` is deeply copied. Doesn't block.
|
## `msg` is deeply copied. Doesn't block.
|
||||||
##
|
##
|
||||||
## Returns `false` if the message was not sent because number of pending items
|
## Returns `false` if the message was not sent because number of pending items
|
||||||
## in the channel exceeded `maxItems`.
|
## in the channel exceeded `maxItems`.
|
||||||
sendImpl(cast[PRawChannel](addr c), cast[PNimType](getTypeInfo(msg)), unsafeAddr(msg), true)
|
result = sendImpl(cast[PRawChannel](addr c), cast[PNimType](getTypeInfo(msg)), unsafeAddr(msg), true)
|
||||||
|
if result:
|
||||||
|
wasMoved(msg)
|
||||||
|
|
||||||
proc llRecv(q: PRawChannel, res: pointer, typ: PNimType) =
|
proc llRecv(q: PRawChannel, res: pointer, typ: PNimType) =
|
||||||
q.ready = true
|
q.ready = true
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue