first steps to explicit channels for thread communication; added mainThreadId
This commit is contained in:
parent
fe5df368c1
commit
42e6130b2c
16 changed files with 155 additions and 71 deletions
|
|
@ -145,17 +145,17 @@ proc rope*(i: BiggestInt): PRope {.rtl, extern: "nro$1BiggestInt".} =
|
|||
proc rope*(f: BiggestFloat): PRope {.rtl, extern: "nro$1BiggestFloat".} =
|
||||
## Converts a float to a rope.
|
||||
result = rope($f)
|
||||
|
||||
proc disableCache*() {.rtl, extern: "nro$1".} =
|
||||
## the cache is discarded and disabled. The GC will reuse its used memory.
|
||||
cache = nil
|
||||
cacheEnabled = false
|
||||
|
||||
proc enableCache*() {.rtl, extern: "nro$1".} =
|
||||
## Enables the caching of leaves. This reduces the memory footprint at
|
||||
## the cost of runtime efficiency.
|
||||
cacheEnabled = true
|
||||
|
||||
proc disableCache*() {.rtl, extern: "nro$1".} =
|
||||
## the cache is discarded and disabled. The GC will reuse its used memory.
|
||||
cache = nil
|
||||
cacheEnabled = false
|
||||
|
||||
proc `&`*(a, b: PRope): PRope {.rtl, extern: "nroConcRopeRope".} =
|
||||
## the concatenation operator for ropes.
|
||||
if a == nil:
|
||||
|
|
|
|||
|
|
@ -783,7 +783,7 @@ const
|
|||
hasThreadSupport = compileOption("threads")
|
||||
hasSharedHeap = defined(boehmgc) # don't share heaps; every thread has its own
|
||||
|
||||
when hasThreadSupport and not hasSharedHeap:
|
||||
when hasThreadSupport:
|
||||
{.pragma: rtlThreadVar, threadvar.}
|
||||
else:
|
||||
{.pragma: rtlThreadVar.}
|
||||
|
|
@ -835,7 +835,7 @@ proc insert*[T](x: var seq[T], item: T, i = 0) {.noSideEffect.} =
|
|||
setLen(x, xl+1)
|
||||
var j = xl-1
|
||||
while j >= i:
|
||||
x[j+1] = x[j]
|
||||
shallowCopy(x[j+1], x[j])
|
||||
dec(j)
|
||||
x[i] = item
|
||||
|
||||
|
|
@ -1411,7 +1411,8 @@ var
|
|||
## writes an error message and terminates the program. `outOfMemHook` can
|
||||
## be used to raise an exception in case of OOM like so:
|
||||
##
|
||||
## code-block:: nimrod
|
||||
## .. code-block:: nimrod
|
||||
##
|
||||
## var gOutOfMem: ref EOutOfMemory
|
||||
## new(gOutOfMem) # need to be allocated *before* OOM really happened!
|
||||
## gOutOfMem.msg = "out of memory"
|
||||
|
|
|
|||
|
|
@ -205,9 +205,8 @@ proc send*[TMsg](receiver: TThreadId[TMsg], msg: TMsg) =
|
|||
var q = cast[PInbox](getInBoxMem(receiver[]))
|
||||
sendImpl(q)
|
||||
|
||||
proc llRecv(res: pointer, typ: PNimType) =
|
||||
proc llRecv(q: PInbox, res: pointer, typ: PNimType) =
|
||||
# to save space, the generic is as small as possible
|
||||
var q = cast[PInbox](getInBoxMem())
|
||||
acquireSys(q.lock)
|
||||
q.ready = true
|
||||
while q.count <= 0:
|
||||
|
|
@ -222,7 +221,8 @@ proc llRecv(res: pointer, typ: PNimType) =
|
|||
proc recv*[TMsg](): TMsg =
|
||||
## receives a message from its internal message queue. This blocks until
|
||||
## a message has arrived! You may use ``peek`` to avoid the blocking.
|
||||
llRecv(addr(result), cast[PNimType](getTypeInfo(result)))
|
||||
var q = cast[PInbox](getInBoxMem())
|
||||
llRecv(q, addr(result), cast[PNimType](getTypeInfo(result)))
|
||||
|
||||
proc peek*(): int =
|
||||
## returns the current number of messages in the inbox.
|
||||
|
|
@ -242,3 +242,56 @@ proc ready*[TMsg](t: var TThread[TMsg]): bool =
|
|||
var q = cast[PInbox](getInBoxMem(t))
|
||||
result = q.ready
|
||||
|
||||
# ---------------------- channel support -------------------------------------
|
||||
|
||||
type
|
||||
TChannel*[TMsg] = TInbox ## a channel for thread communication
|
||||
TChannelId*[TMsg] = ptr TChannel[TMsg] ## the current implementation uses
|
||||
## a pointer as a channel ID.
|
||||
|
||||
proc open*[TMsg](c: var TChannel[TMsg]) =
|
||||
## opens a channel `c` for inter thread communication.
|
||||
initInbox(addr(c))
|
||||
|
||||
proc close*[TMsg](c: var TChannel[TMsg]) =
|
||||
## closes a channel `c` and frees its associated resources.
|
||||
freeInbox(addr(c))
|
||||
|
||||
proc channelId*[TMsg](c: var TChannel[TMsg]): TChannelId[TMsg] {.inline.} =
|
||||
## returns the channel ID of `c`.
|
||||
result = addr(c)
|
||||
|
||||
proc send*[TMsg](c: var TChannel[TMsg], msg: TMsg) =
|
||||
## sends a message to a channel. `msg` is deeply copied.
|
||||
var q = cast[PInbox](addr(c))
|
||||
sendImpl(q)
|
||||
|
||||
proc send*[TMsg](c: TChannelId[TMsg], msg: TMsg) =
|
||||
## sends a message to a thread. `msg` is deeply copied.
|
||||
var q = cast[PInbox](c)
|
||||
sendImpl(q)
|
||||
|
||||
proc peek*[TMsg](c: var TChannel[TMsg]): int =
|
||||
## returns the current number of messages in the channel `c`.
|
||||
var q = cast[PInbox](addr(c))
|
||||
lockInbox(q):
|
||||
result = q.count
|
||||
|
||||
proc peek*[TMsg](c: TChannelId[TMsg]): int =
|
||||
## returns the current number of messages in the channel `c`.
|
||||
var q = cast[PInbox](c)
|
||||
lockInbox(q):
|
||||
result = q.count
|
||||
|
||||
proc recv*[TMsg](c: TChannelId[TMsg]): TMsg =
|
||||
## receives a message from the channel `c`. This blocks until
|
||||
## a message has arrived! You may use ``peek`` to avoid the blocking.
|
||||
var q = cast[PInbox](c)
|
||||
llRecv(q, addr(result), cast[PNimType](getTypeInfo(result)))
|
||||
|
||||
proc recv*[TMsg](c: var TChannel[TMsg]): TMsg =
|
||||
## receives a message from the channel `c`. This blocks until
|
||||
## a message has arrived! You may use ``peek`` to avoid the blocking.
|
||||
var q = cast[PInbox](addr(c))
|
||||
llRecv(q, addr(result), cast[PNimType](getTypeInfo(result)))
|
||||
|
||||
|
|
|
|||
|
|
@ -133,6 +133,26 @@ when defined(boehmgc):
|
|||
proc asgnRefNoCycle(dest: ppointer, src: pointer) {.compilerproc, inline.} =
|
||||
dest[] = src
|
||||
|
||||
type
|
||||
TMemRegion = object {.final, pure.}
|
||||
|
||||
var
|
||||
dummy {.rtlThreadVar.}: int
|
||||
|
||||
proc rawAlloc(r: var TMemRegion, size: int): pointer =
|
||||
result = boehmAlloc(size)
|
||||
if result == nil: raiseOutOfMem()
|
||||
proc rawAlloc0(r: var TMemRegion, size: int): pointer =
|
||||
result = alloc(size)
|
||||
zeroMem(result, size)
|
||||
proc realloc(r: var TMemRegion, p: Pointer, newsize: int): pointer =
|
||||
result = boehmRealloc(p, newsize)
|
||||
if result == nil: raiseOutOfMem()
|
||||
proc rawDealloc(r: var TMemRegion, p: Pointer) = boehmDealloc(p)
|
||||
|
||||
proc deallocOsPages(r: var TMemRegion) {.inline.} = nil
|
||||
proc deallocOsPages() {.inline.} = nil
|
||||
|
||||
include "system/cellsets"
|
||||
elif defined(nogc):
|
||||
# Even though we don't want the GC, we cannot simply use C's memory manager
|
||||
|
|
|
|||
|
|
@ -266,7 +266,10 @@ proc initInbox(p: pointer)
|
|||
proc freeInbox(p: pointer)
|
||||
when not defined(boehmgc) and not hasSharedHeap:
|
||||
proc deallocOsPages()
|
||||
|
||||
|
||||
when defined(mainThread):
|
||||
initInbox(addr(mainThread.inbox))
|
||||
|
||||
template ThreadProcWrapperBody(closure: expr) =
|
||||
ThreadVarSetValue(globalsSlot, closure)
|
||||
var t = cast[ptr TThread[TMsg]](closure)
|
||||
|
|
@ -379,6 +382,10 @@ proc myThreadId*[TMsg](): TThreadId[TMsg] =
|
|||
## returns the thread ID of the thread that calls this proc.
|
||||
result = cast[TThreadId[TMsg]](ThreadVarGetValue(globalsSlot))
|
||||
|
||||
proc mainThreadId*[TMsg](): TThreadId[TMsg] =
|
||||
## returns the thread ID of the main thread.
|
||||
result = cast[TThreadId[TMsg]](addr(mainThread))
|
||||
|
||||
when useStackMaskHack:
|
||||
proc runMain(tp: proc () {.thread.}) {.compilerproc.} =
|
||||
var mainThread: TThread[pointer]
|
||||
|
|
@ -388,7 +395,8 @@ when useStackMaskHack:
|
|||
# --------------------------- lock handling ----------------------------------
|
||||
|
||||
type
|
||||
TLock* = TSysLock ## Nimrod lock; not re-entrant!
|
||||
TLock* = TSysLock ## Nimrod lock; whether this is re-entrant
|
||||
## or not is unspecified!
|
||||
|
||||
const
|
||||
noDeadlocks = false # compileOption("deadlockPrevention")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue