broken attempt to fix queue exhaustion

lib/pure/concurrency/threadpool.nim

@@ -100,7 +100,7 @@ type
     # for 'awaitAny' support
     ai: ptr AwaitInfo
     idx: int
-    data: pointer  # we incRef and unref it to keep it alive
+    data: RootRef  # we incRef and unref it to keep it alive
     owner: pointer # ptr Worker
 
   FlowVarObj[T] = object of FlowVarBaseObj
@@ -112,7 +112,7 @@ type
     len: int
     lock: TLock
     empty: TCond
-    data: array[512, pointer]
+    data: array[2, pointer]
 
   WorkerProc = proc (thread, args: pointer) {.nimcall, gcsafe.}
   Worker = object
@@ -134,6 +134,22 @@ proc await*(fv: FlowVarBase) =
     await(fv.cv)
     destroyCondVar(fv.cv)
 
+proc selectWorker(w: ptr Worker; fn: WorkerProc; data: pointer): bool =
+  if cas(addr w.ready, true, false):
+    w.data = data
+    w.f = fn
+    signal(w.taskArrived)
+    await(w.taskStarted)
+    result = true
+
+proc wakeupWorkerToProcessQueue(w: ptr Worker) =
+  # Note that if this fails somebody else already woke up the thread so it's
+  # perfectly fine to do nothing:
+  if cas(addr w.ready, true, false):
+    w.data = nil
+    w.f = proc (t, a: pointer) {.nimcall.} = discard
+    signal(w.taskArrived)
+
 proc finished(fv: FlowVarBase) =
   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
@@ -144,18 +160,17 @@ proc finished(fv: FlowVarBase) =
   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:
+  while not (q.len < q.data.len):
-      q.data[q.len] = fv.data
+    echo "EXHAUSTED!"
-      inc q.len
-      release(q.lock)
-      break
-    else:
-      # the queue is exhausted! We block until it has been cleaned:
     release(q.lock)
+    wakeupWorkerToProcessQueue(owner)
+    acquire(q.lock)
     wait(q.empty, q.lock)
     waited = true
+  q.data[q.len] = cast[pointer](fv.data)
+  inc q.len
+  release(q.lock)
   fv.data = nil
   # wakeup other potentially waiting threads:
   if waited: signal(q.empty)
@@ -165,6 +180,7 @@ proc cleanFlowVars(w: ptr Worker) =
   acquire(q.lock)
   for i in 0 .. <q.len:
     GC_unref(cast[RootRef](q.data[i]))
+    echo "GC_unref"
   q.len = 0
   release(q.lock)
   signal(q.empty)
@@ -331,14 +347,6 @@ var
 
 initLock stateLock
 
-proc selectWorker(w: ptr Worker; fn: WorkerProc; data: pointer): bool =
-  if cas(addr w.ready, true, false):
-    w.data = data
-    w.f = fn
-    signal(w.taskArrived)
-    await(w.taskStarted)
-    result = true
-
 proc nimSpawn(fn: WorkerProc; data: pointer) {.compilerProc.} =
   # implementation of 'spawn' that is used by the code generator.
   while true:

tests/parallel/tconvexhull.nim

@@ -0,0 +1,62 @@
+discard """
+  output: '''true
+true
+true
+true
+true
+true'''
+"""
+
+# parallel convex hull for Nim bigbreak
+# nim c --threads:on -d:release pconvex_hull.nim
+import algorithm, sequtils, threadpool
+
+type Point = tuple[x, y: float]
+
+proc cmpPoint(a, b: Point): int =
+  result = cmp(a.x, b.x)
+  if result == 0:
+    result = cmp(a.y, b.y)
+
+template cross[T](o, a, b: T): expr =
+  (a.x - o.x) * (b.y - o.y) - (a.y - o.y) * (b.x - o.x)
+
+template pro(): expr =
+  while lr1 > 0 and cross(result[lr1 - 1], result[lr1], p[i]) <= 0:
+    discard result.pop
+    lr1 -= 1
+  result.add(p[i])
+  lr1 += 1
+
+proc half[T](p: seq[T]; upper: bool): seq[T] =
+  var i, lr1: int
+  result = @[]
+  lr1 = -1
+  if upper:
+    i = 0
+    while i <= high(p):
+      pro()
+      i += 1
+  else:
+    i = high(p)
+    while i >= low(p):
+      pro()
+      i -= 1
+  discard result.pop
+
+proc convex_hull[T](points: var seq[T], cmp: proc(x, y: T): int {.closure.}) : seq[T] =
+  if len(points) < 2: return points
+  points.sort(cmp)
+  var ul: array[2, FlowVar[seq[T]]]
+  parallel:
+    for k in 0..ul.high:
+      ul[k] = spawn half[T](points, k == 0)
+  result = concat(^ul[0], ^ul[1])
+
+var s = map(toSeq(0..999999), proc(x: int): Point = (float(x div 1000), float(x mod 1000)))
+setMaxPoolSize 2
+
+#echo convex_hull[Point](s, cmpPoint)
+for i in 0..5:
+  echo convex_hull[Point](s, cmpPoint) ==
+      @[(0.0, 0.0), (999.0, 0.0), (999.0, 999.0), (0.0, 999.0)]