This commit is contained in:
Eugene Kabanov 2017-03-23 17:13:38 +02:00 • committed by Andreas Rumpf
commit 254fbcc548
4 changed files with 84 additions and 19 deletions

View file

@ -763,6 +763,8 @@ proc getOccupiedMem(a: MemRegion): int {.inline.} =
# ---------------------- thread memory region ------------------------------- # ---------------------- thread memory region -------------------------------
template instantiateForRegion(allocator: untyped) = template instantiateForRegion(allocator: untyped) =
{.push stackTrace: off.}
when defined(fulldebug): when defined(fulldebug):
proc interiorAllocatedPtr*(p: pointer): pointer = proc interiorAllocatedPtr*(p: pointer): pointer =
result = interiorAllocatedPtr(allocator, p) result = interiorAllocatedPtr(allocator, p)
@ -850,5 +852,6 @@ template instantiateForRegion(allocator: untyped) =
proc getOccupiedSharedMem(): int = proc getOccupiedSharedMem(): int =
sharedMemStatsShared(sharedHeap.currMem - sharedHeap.freeMem) sharedMemStatsShared(sharedHeap.currMem - sharedHeap.freeMem)
{.pop.}
{.pop.} {.pop.}

View file

@ -242,7 +242,6 @@ type
PGcThread = ptr GcThread PGcThread = ptr GcThread
GcThread {.pure, inheritable.} = object GcThread {.pure, inheritable.} = object
sys: SysThread
when emulatedThreadVars and not useStackMaskHack: when emulatedThreadVars and not useStackMaskHack:
tls: ThreadLocalStorage tls: ThreadLocalStorage
else: else:
@ -345,18 +344,16 @@ when not defined(useNimRtl):
# use ``stdcall`` since it is mapped to ``noconv`` on UNIX anyway. # use ``stdcall`` since it is mapped to ``noconv`` on UNIX anyway.
type type
Thread* {.pure, final.}[TArg] = Thread* {.pure, final.}[TArg] = object
object of GcThread ## Nim thread. A thread is a heavy object (~14K) core: PGcThread
## that **must not** be part of a message! Use sys: SysThread
## a ``ThreadId`` for that.
when TArg is void: when TArg is void:
dataFn: proc () {.nimcall, gcsafe.} dataFn: proc () {.nimcall, gcsafe.}
else: else:
dataFn: proc (m: TArg) {.nimcall, gcsafe.} dataFn: proc (m: TArg) {.nimcall, gcsafe.}
data: TArg data: TArg
ThreadId*[TArg] = ptr Thread[TArg] ## the current implementation uses
## a pointer as a thread ID. {.deprecated: [TThread: Thread].}
{.deprecated: [TThread: Thread, TThreadId: ThreadId].}
var var
threadDestructionHandlers {.rtlThreadVar.}: seq[proc () {.closure, gcsafe.}] threadDestructionHandlers {.rtlThreadVar.}: seq[proc () {.closure, gcsafe.}]
@ -423,19 +420,20 @@ proc threadProcWrapStackFrame[TArg](thrd: ptr Thread[TArg]) =
when declared(threadType): when declared(threadType):
threadType = ThreadType.NimThread threadType = ThreadType.NimThread
when declared(registerThread): when declared(registerThread):
thrd.stackBottom = addr(thrd) thrd.core.stackBottom = addr(thrd)
registerThread(thrd) registerThread(thrd.core)
p(thrd) p(thrd)
when declared(registerThread): unregisterThread(thrd) when declared(registerThread): unregisterThread(thrd.core)
when declared(deallocOsPages): deallocOsPages() when declared(deallocOsPages): deallocOsPages()
else: else:
threadProcWrapDispatch(thrd) threadProcWrapDispatch(thrd)
template threadProcWrapperBody(closure: expr) {.immediate.} = template threadProcWrapperBody(closure: expr) {.immediate.} =
when declared(globalsSlot): threadVarSetValue(globalsSlot, closure) var thrd = cast[ptr Thread[TArg]](closure)
var core = thrd.core
when declared(globalsSlot): threadVarSetValue(globalsSlot, thrd.core)
when declared(initAllocator): when declared(initAllocator):
initAllocator() initAllocator()
var thrd = cast[ptr Thread[TArg]](closure)
threadProcWrapStackFrame(thrd) threadProcWrapStackFrame(thrd)
# Since an unhandled exception terminates the whole process (!), there is # Since an unhandled exception terminates the whole process (!), there is
# no need for a ``try finally`` here, nor would it be correct: The current # no need for a ``try finally`` here, nor would it be correct: The current
@ -444,7 +442,9 @@ template threadProcWrapperBody(closure: expr) {.immediate.} =
# page! # page!
# mark as not running anymore: # mark as not running anymore:
thrd.core = nil
thrd.dataFn = nil thrd.dataFn = nil
deallocShared(cast[pointer](core))
{.push stack_trace:off.} {.push stack_trace:off.}
when defined(windows): when defined(windows):
@ -502,6 +502,10 @@ when false:
discard pthread_cancel(t.sys) discard pthread_cancel(t.sys)
when declared(registerThread): unregisterThread(addr(t)) when declared(registerThread): unregisterThread(addr(t))
t.dataFn = nil t.dataFn = nil
## if thread `t` already exited, `t.core` will be `null`.
if not isNil(t.core):
deallocShared(t.core)
t.core = nil
when hostOS == "windows": when hostOS == "windows":
proc createThread*[TArg](t: var Thread[TArg], proc createThread*[TArg](t: var Thread[TArg],
@ -510,9 +514,11 @@ when hostOS == "windows":
## creates a new thread `t` and starts its execution. Entry point is the ## creates a new thread `t` and starts its execution. Entry point is the
## proc `tp`. `param` is passed to `tp`. `TArg` can be ``void`` if you ## proc `tp`. `param` is passed to `tp`. `TArg` can be ``void`` if you
## don't need to pass any data to the thread. ## don't need to pass any data to the thread.
t.core = cast[PGcThread](allocShared0(sizeof(GcThread)))
when TArg isnot void: t.data = param when TArg isnot void: t.data = param
t.dataFn = tp t.dataFn = tp
when hasSharedHeap: t.stackSize = ThreadStackSize when hasSharedHeap: t.core.stackSize = ThreadStackSize
var dummyThreadId: int32 var dummyThreadId: int32
t.sys = createThread(nil, ThreadStackSize, threadProcWrapper[TArg], t.sys = createThread(nil, ThreadStackSize, threadProcWrapper[TArg],
addr(t), 0'i32, dummyThreadId) addr(t), 0'i32, dummyThreadId)
@ -532,9 +538,11 @@ else:
## creates a new thread `t` and starts its execution. Entry point is the ## creates a new thread `t` and starts its execution. Entry point is the
## proc `tp`. `param` is passed to `tp`. `TArg` can be ``void`` if you ## proc `tp`. `param` is passed to `tp`. `TArg` can be ``void`` if you
## don't need to pass any data to the thread. ## don't need to pass any data to the thread.
t.core = cast[PGcThread](allocShared0(sizeof(GcThread)))
when TArg isnot void: t.data = param when TArg isnot void: t.data = param
t.dataFn = tp t.dataFn = tp
when hasSharedHeap: t.stackSize = ThreadStackSize when hasSharedHeap: t.core.stackSize = ThreadStackSize
var a {.noinit.}: PthreadAttr var a {.noinit.}: PthreadAttr
pthread_attr_init(a) pthread_attr_init(a)
pthread_attr_setstacksize(a, ThreadStackSize) pthread_attr_setstacksize(a, ThreadStackSize)
@ -554,10 +562,6 @@ else:
proc createThread*(t: var Thread[void], tp: proc () {.thread, nimcall.}) = proc createThread*(t: var Thread[void], tp: proc () {.thread, nimcall.}) =
createThread[void](t, tp) createThread[void](t, tp)
proc threadId*[TArg](t: var Thread[TArg]): ThreadId[TArg] {.inline.} =
## returns the thread ID of `t`.
result = addr(t)
when false: when false:
proc mainThreadId*[TArg](): ThreadId[TArg] = proc mainThreadId*[TArg](): ThreadId[TArg] =
## returns the thread ID of the main thread. ## returns the thread ID of the main thread.

View file

@ -0,0 +1,30 @@
discard """
outputsub: "129"
"""
import os, locks
type
MarkerObj = object
lock: Lock
counter: int
Marker = ptr MarkerObj
const
ThreadsCount = 129
SleepTime = 1000
proc worker(p: Marker) {.thread.} =
acquire(p.lock)
inc(p.counter)
release(p.lock)
sleep(SleepTime)
var p = cast[Marker](allocShared0(sizeof(MarkerObj)))
initLock(p.lock)
var ts = newSeq[Thread[Marker]](ThreadsCount)
for i in 0..<ts.len:
createThread(ts[i], worker, p)
joinThreads(ts)
echo p.counter

View file

@ -0,0 +1,28 @@
discard """
outputsub: "129"
"""
import os, locks
type
MarkerObj = object
lock: Lock
counter: int
Marker = ptr MarkerObj
const
ThreadsCount = 129
proc worker(p: Marker) {.thread.} =
acquire(p.lock)
inc(p.counter)
release(p.lock)
var p = cast[Marker](allocShared0(sizeof(MarkerObj)))
initLock(p.lock)
for i in 0..(ThreadsCount - 1):
var thread: Thread[Marker]
createThread(thread, worker, p)
joinThread(thread)
echo p.counter