Merge branch 'new_spawn' of https://github.com/Araq/Nimrod into new_spawn

2026-07-17 06:21:18 +00:00 · 2014-06-06 21:11:11 +02:00
parent 387593bcaa c8b5d6a634
commit 4220b1c81d
9 changed files with 305 additions and 212 deletions
--- a/lib/pure/concurrency/threadpool.nim
+++ b/lib/pure/concurrency/threadpool.nim
@@ -40,33 +40,44 @@ proc signal(cv: var CondVar) =
  release(cv.L)
  signal(cv.c)

+const CacheLineSize = 32 # true for most archs
+
 type
-  Barrier* {.compilerProc.} = object
+  Barrier {.compilerProc.} = object
    entered: int
-    cv: CondVar
-    cacheAlign: array[0..20, byte] # ensure 'left' is not on the same
-                                   # cache line as 'entered'
+    cv: CondVar # condvar takes 3 words at least
+    when sizeof(int) < 8:
+      cacheAlign: array[CacheLineSize-4*sizeof(int), byte] 
    left: int
+    cacheAlign2: array[CacheLineSize-sizeof(int), byte]
+    interest: bool ## wether the master is interested in the "all done" event

-proc barrierEnter*(b: ptr Barrier) {.compilerProc.} =
-  atomicInc b.entered
+proc barrierEnter(b: ptr Barrier) {.compilerProc, inline.} =
+  # due to the signaling between threads, it is ensured we are the only
+  # one with access to 'entered' so we don't need 'atomicInc' here:
+  inc b.entered
+  # also we need no 'fence' instructions here as soon 'nimArgsPassingDone'
+  # will be called which already will perform a fence for us.

-proc barrierLeave*(b: ptr Barrier) {.compilerProc.} =
+proc barrierLeave(b: ptr Barrier) {.compilerProc, inline.} =
  atomicInc b.left
-  # these can only be equal if 'closeBarrier' already signaled its interest
-  # in this event:
-  if b.left == b.entered: signal(b.cv)
+  when not defined(x86): fence()
+  if b.interest and b.left == b.entered: signal(b.cv)

-proc openBarrier*(b: ptr Barrier) {.compilerProc.} =
+proc openBarrier(b: ptr Barrier) {.compilerProc, inline.} =
  b.entered = 0
-  b.cv = createCondVar()
-  b.left = -1
+  b.left = 0
+  b.interest = false

-proc closeBarrier*(b: ptr Barrier) {.compilerProc.} =
-  # signal interest in the "all done" event:
-  atomicInc b.left
-  while b.left != b.entered: await(b.cv)
-  destroyCondVar(b.cv)
+proc closeBarrier(b: ptr Barrier) {.compilerProc.} =
+  fence()
+  if b.left != b.entered:
+    b.cv = createCondVar()
+    fence()
+    b.interest = true
+    fence()
+    while b.left != b.entered: await(b.cv)
+    destroyCondVar(b.cv)

 {.pop.}

@@ -79,25 +90,26 @@ type
    cv: CondVar
    idx: int

-  RawPromise* = ptr RawPromiseObj ## untyped base class for 'Promise[T]'
-  RawPromiseObj {.inheritable.} = object # \
-    # we allocate this with the thread local allocator; this
-    # is possible since we already need to do the GC_unref
-    # on the owning thread
+  RawFlowVar* = ref RawFlowVarObj ## untyped base class for 'FlowVar[T]'
+  RawFlowVarObj = object of TObject
    ready, usesCondVar: bool
    cv: CondVar #\
    # for 'awaitAny' support
    ai: ptr AwaitInfo
    idx: int
-    data: PObject  # we incRef and unref it to keep it alive
-    owner: ptr Worker
-    next: RawPromise
-    align: float64 # a float for proper alignment
+    data: pointer  # we incRef and unref it to keep it alive
+    owner: pointer # ptr Worker

-  Promise* {.compilerProc.} [T] = ptr object of RawPromiseObj
-    blob: T  ## the underlying value, if available. Note that usually
-             ## you should not access this field directly! However it can
-             ## sometimes be more efficient than getting the value via ``^``.
+  FlowVarObj[T] = object of RawFlowVarObj
+    blob: T
+
+  FlowVar*{.compilerProc.}[T] = ref FlowVarObj[T] ## a data flow variable
+
+  ToFreeQueue = object
+    len: int
+    lock: TLock
+    empty: TCond
+    data: array[512, pointer]

  WorkerProc = proc (thread, args: pointer) {.nimcall, gcsafe.}
  Worker = object
@@ -109,109 +121,117 @@ type
    ready: bool # put it here for correct alignment!
    initialized: bool # whether it has even been initialized
    shutdown: bool # the pool requests to shut down this worker thread
-    promiseLock: TLock
-    head: RawPromise
+    q: ToFreeQueue

-proc finished*(prom: RawPromise) =
-  ## This MUST be called for every created promise to free its associated
-  ## resources. Note that the default reading operation ``^`` is destructive
-  ## and calls ``finished``.
-  doAssert prom.ai.isNil, "promise is still attached to an 'awaitAny'"
-  assert prom.next == nil
-  let w = prom.owner
-  acquire(w.promiseLock)
-  prom.next = w.head
-  w.head = prom
-  release(w.promiseLock)
+proc await*(fv: RawFlowVar) =
+  ## waits until the value for the flowVar arrives. Usually it is not necessary
+  ## to call this explicitly.
+  if fv.usesCondVar:
+    fv.usesCondVar = false
+    await(fv.cv)
+    destroyCondVar(fv.cv)

-proc cleanPromises(w: ptr Worker) =
-  var it = w.head
-  acquire(w.promiseLock)
-  while it != nil:
-    let nxt = it.next
-    if it.usesCondVar: destroyCondVar(it.cv)
-    if it.data != nil: GC_unref(it.data)
-    dealloc(it)
-    it = nxt
-  w.head = nil
-  release(w.promiseLock)
+proc finished(fv: RawFlowVar) =
+  doAssert fv.ai.isNil, "flowVar is still attached to an 'awaitAny'"
+  # we have to protect against the rare cases where the owner of the flowVar
+  # simply disregards the flowVar and yet the "flowVarr" has not yet written
+  # anything to it:
+  await(fv)
+  if fv.data.isNil: return
+  let owner = cast[ptr Worker](fv.owner)
+  let q = addr(owner.q)
+  var waited = false
+  while true:
+    acquire(q.lock)
+    if q.len < q.data.len:
+      q.data[q.len] = fv.data
+      inc q.len
+      release(q.lock)
+      break
+    else:
+      # the queue is exhausted! We block until it has been cleaned:
+      release(q.lock)
+      wait(q.empty, q.lock)
+      waited = true
+  fv.data = nil
+  # wakeup other potentially waiting threads:
+  if waited: signal(q.empty)

-proc nimCreatePromise(owner: pointer; blobSize: int): RawPromise {.
-                     compilerProc.} =
-  result = cast[RawPromise](alloc0(RawPromiseObj.sizeof + blobSize))
-  result.owner = cast[ptr Worker](owner)
+proc cleanFlowVars(w: ptr Worker) =
+  let q = addr(w.q)
+  acquire(q.lock)
+  for i in 0 .. <q.len:
+    GC_unref(cast[PObject](q.data[i]))
+  q.len = 0
+  release(q.lock)
+  signal(q.empty)

-proc nimPromiseCreateCondVar(prom: RawPromise) {.compilerProc.} =
-  prom.cv = createCondVar()
-  prom.usesCondVar = true
+proc fvFinalizer[T](fv: FlowVar[T]) = finished(fv)

-proc nimPromiseSignal(prom: RawPromise) {.compilerProc.} =
-  if prom.ai != nil:
-    acquire(prom.ai.cv.L)
-    prom.ai.idx = prom.idx
-    inc prom.ai.cv.counter
-    release(prom.ai.cv.L)
-    signal(prom.ai.cv.c)
-  if prom.usesCondVar: signal(prom.cv)
+proc nimCreateFlowVar[T](): FlowVar[T] {.compilerProc.} =
+  new(result, fvFinalizer)

-proc await*[T](prom: Promise[T]) =
-  ## waits until the value for the promise arrives.
-  if prom.usesCondVar: await(prom.cv)
+proc nimFlowVarCreateCondVar(fv: RawFlowVar) {.compilerProc.} =
+  fv.cv = createCondVar()
+  fv.usesCondVar = true

-proc awaitAndThen*[T](prom: Promise[T]; action: proc (x: T) {.closure.}) =
-  ## blocks until the ``prom`` is available and then passes its value
+proc nimFlowVarSignal(fv: RawFlowVar) {.compilerProc.} =
+  if fv.ai != nil:
+    acquire(fv.ai.cv.L)
+    fv.ai.idx = fv.idx
+    inc fv.ai.cv.counter
+    release(fv.ai.cv.L)
+    signal(fv.ai.cv.c)
+  if fv.usesCondVar: signal(fv.cv)
+
+proc awaitAndThen*[T](fv: FlowVar[T]; action: proc (x: T) {.closure.}) =
+  ## blocks until the ``fv`` is available and then passes its value
  ## to ``action``. Note that due to Nimrod's parameter passing semantics this
  ## means that ``T`` doesn't need to be copied and so ``awaitAndThen`` can
  ## sometimes be more efficient than ``^``.
-  if prom.usesCondVar: await(prom)
+  await(fv)
  when T is string or T is seq:
-    action(cast[T](prom.data))
+    action(cast[T](fv.data))
  elif T is ref:
-    {.error: "'awaitAndThen' not available for Promise[ref]".}
+    {.error: "'awaitAndThen' not available for FlowVar[ref]".}
  else:
-    action(prom.blob)
-  finished(prom)
+    action(fv.blob)
+  finished(fv)

-proc `^`*[T](prom: Promise[ref T]): foreign ptr T =
-  ## blocks until the value is available and then returns this value. Note
-  ## this reading is destructive for reasons of efficiency and convenience.
-  ## This calls ``finished(prom)``.
-  if prom.usesCondVar: await(prom)
-  result = cast[foreign ptr T](prom.data)
-  finished(prom)
+proc `^`*[T](fv: FlowVar[ref T]): foreign ptr T =
+  ## blocks until the value is available and then returns this value.
+  await(fv)
+  result = cast[foreign ptr T](fv.data)

-proc `^`*[T](prom: Promise[T]): T =
-  ## blocks until the value is available and then returns this value. Note
-  ## this reading is destructive for reasons of efficiency and convenience.
-  ## This calls ``finished(prom)``.
-  if prom.usesCondVar: await(prom)
+proc `^`*[T](fv: FlowVar[T]): T =
+  ## blocks until the value is available and then returns this value.
+  await(fv)
  when T is string or T is seq:
-    result = cast[T](prom.data)
+    result = cast[T](fv.data)
  else:
-    result = prom.blob
-  finished(prom)
+    result = fv.blob

-proc awaitAny*(promises: openArray[RawPromise]): int =
-  # awaits any of the given promises. Returns the index of one promise for which
-  ## a value arrived. A promise only supports one call to 'awaitAny' at the
-  ## same time. That means if you await([a,b]) and await([b,c]) the second
-  ## call will only await 'c'. If there is no promise left to be able to wait
+proc awaitAny*(flowVars: openArray[RawFlowVar]): int =
+  ## awaits any of the given flowVars. Returns the index of one flowVar for
+  ## which a value arrived. A flowVar only supports one call to 'awaitAny' at
+  ## the same time. That means if you await([a,b]) and await([b,c]) the second
+  ## call will only await 'c'. If there is no flowVar left to be able to wait
  ## on, -1 is returned.
  ## **Note**: This results in non-deterministic behaviour and so should be
  ## avoided.
  var ai: AwaitInfo
  ai.cv = createCondVar()
  var conflicts = 0
-  for i in 0 .. promises.high:
-    if cas(addr promises[i].ai, nil, addr ai):
-      promises[i].idx = i
+  for i in 0 .. flowVars.high:
+    if cas(addr flowVars[i].ai, nil, addr ai):
+      flowVars[i].idx = i
    else:
      inc conflicts
-  if conflicts < promises.len:
+  if conflicts < flowVars.len:
    await(ai.cv)
    result = ai.idx
-    for i in 0 .. promises.high:
-      discard cas(addr promises[i].ai, addr ai, nil)
+    for i in 0 .. flowVars.high:
+      discard cas(addr flowVars[i].ai, addr ai, nil)
  else:
    result = -1
  destroyCondVar(ai.cv)
@@ -239,7 +259,7 @@ proc slave(w: ptr Worker) {.thread.} =
    await(w.taskArrived)
    assert(not w.ready)
    w.f(w, w.data)
-    if w.head != nil: w.cleanPromises
+    if w.q.len != 0: w.cleanFlowVars
    if w.shutdown:
      w.shutdown = false
      atomicDec currentPoolSize
@@ -260,8 +280,9 @@ var
 proc activateThread(i: int) {.noinline.} =
  workersData[i].taskArrived = createCondVar()
  workersData[i].taskStarted = createCondVar()
-  initLock workersData[i].promiseLock
  workersData[i].initialized = true
+  initCond(workersData[i].q.empty)
+  initLock(workersData[i].q.lock)
  createThread(workers[i], slave, addr(workersData[i]))

 proc setup() =
@@ -278,14 +299,16 @@ proc preferSpawn*(): bool =
 proc spawn*(call: expr): expr {.magic: "Spawn".}
  ## always spawns a new task, so that the 'call' is never executed on
  ## the calling thread. 'call' has to be proc call 'p(...)' where 'p'
-  ## is gcsafe and has 'void' as the return type.
+  ## is gcsafe and has a return type that is either 'void' or compatible
+  ## with ``FlowVar[T]``.

 template spawnX*(call: expr): expr =
  ## spawns a new task if a CPU core is ready, otherwise executes the
  ## call in the calling thread. Usually it is advised to
  ## use 'spawn' in order to not block the producer for an unknown
  ## amount of time. 'call' has to be proc call 'p(...)' where 'p'
-  ## is gcsafe and has 'void' as the return type.
+  ## is gcsafe and has a return type that is either 'void' or compatible
+  ## with ``FlowVar[T]``.
  (if preferSpawn(): spawn call else: call)

 proc parallel*(body: stmt) {.magic: "Parallel".}
--- a/lib/system/atomics.nim
+++ b/lib/system/atomics.nim
@@ -10,7 +10,9 @@
 ## Atomic operations for Nimrod.
 {.push stackTrace:off.}

-when (defined(gcc) or defined(llvm_gcc)) and hasThreadSupport:
+const someGcc = defined(gcc) or defined(llvm_gcc) or defined(clang)
+
+when someGcc and hasThreadSupport:
  type 
    AtomMemModel* = enum
      ATOMIC_RELAXED,  ## No barriers or synchronization. 
@@ -153,41 +155,16 @@ when (defined(gcc) or defined(llvm_gcc)) and hasThreadSupport:
    ## A value of 0 indicates typical alignment should be used. The compiler may also 
    ## ignore this parameter.

+  template fence*() = atomicThreadFence(ATOMIC_SEQ_CST)
 elif defined(vcc) and hasThreadSupport:
  proc addAndFetch*(p: ptr int, val: int): int {.
    importc: "NimXadd", nodecl.}
+
 else:
  proc addAndFetch*(p: ptr int, val: int): int {.inline.} =
    inc(p[], val)
    result = p[]

-# atomic compare and swap (CAS) funcitons to implement lock-free algorithms  
-      
-#if defined(windows) and not defined(gcc) and hasThreadSupport:
-#    proc InterlockedCompareExchangePointer(mem: ptr pointer,
-#      newValue: pointer, comparand: pointer) : pointer {.nodecl, 
-#        importc: "InterlockedCompareExchangePointer", header:"windows.h".}
-
-#    proc compareAndSwap*[T](mem: ptr T, 
-#      expected: T, newValue: T): bool {.inline.}=
-#      ## Returns true if successfully set value at mem to newValue when value
-#      ## at mem == expected
-#      return InterlockedCompareExchangePointer(addr(mem), 
-#        addr(newValue), addr(expected))[] == expected
-    
-#elif not hasThreadSupport:
-#  proc compareAndSwap*[T](mem: ptr T, 
-#                          expected: T, newValue: T): bool {.inline.} =
-#      ## Returns true if successfully set value at mem to newValue when value
-#      ## at mem == expected
-#      var oldval = mem[]
-#      if oldval == expected:
-#        mem[] = newValue
-#        return true
-#      return false
-
-
-# Some convenient functions 
 proc atomicInc*(memLoc: var int, x: int = 1): int =
  when defined(gcc) and hasThreadSupport:
    result = atomic_add_fetch(memLoc.addr, x, ATOMIC_RELAXED)
@@ -205,7 +182,7 @@ proc atomicDec*(memLoc: var int, x: int = 1): int =
    dec(memLoc, x)
    result = memLoc

-when defined(windows) and not defined(gcc):
+when defined(windows) and not someGcc:
  proc interlockedCompareExchange(p: pointer; exchange, comparand: int32): int32
    {.importc: "InterlockedCompareExchange", header: "<windows.h>", cdecl.}

@@ -219,7 +196,7 @@ else:
  # XXX is this valid for 'int'?


-when (defined(x86) or defined(amd64)) and defined(gcc):
+when (defined(x86) or defined(amd64)) and (defined(gcc) or defined(llvm_gcc)):
  proc cpuRelax {.inline.} =
    {.emit: """asm volatile("pause" ::: "memory");""".}
 elif (defined(x86) or defined(amd64)) and defined(vcc):
@@ -231,4 +208,10 @@ elif false:

  proc cpuRelax {.inline.} = os.sleep(1)

+when not defined(fence) and hasThreadSupport:
+  # XXX fixme
+  proc fence*() {.inline.} =
+    var dummy: bool
+    discard cas(addr dummy, false, true)
+
 {.pop.}