Mirror of forum.nim-lang.org

5195 :: Trying to combine threads with async by channels

• jackhftang (orginal) [2019-09-16T07:44:04+02:00] view original

  When I start small, passing pointer of channel seems working fine. But when I grouping logics into proc, I got segmentation fault, illegal storage access or channel only work on one direction, not the other. I cannot reason about the behavior, even after consulting the manual... so I search for help here.

  In short, I want to ask

  1. Is it a correct If I pass the WorkChannel object (like below) to thread,
  2. Is it safe to declare channels in local varible in proc, and take the addr and pass to a thread (like newWorkerChannel)?
  3. Is threads, channels library stable?

  nim version is 0.20.2 on ubunutu

  import os, asyncdispatch
  
  type
    WorkerChannel*[R,T] = object
      rx: ptr Channel[R]
      tx: ptr Channel[T]
    Worker*[R,T] = proc(args: WorkerChannel[R,T]) {.thread,nimcall.}
  
  proc newWorkerChannel[R,T]( maxTxItems: int = 0, maxRxItems: int = 0): WorkerChannel[R,T] =
    var tx: Channel[T]
    var rx: Channel[R]
    tx.open(maxTxItems)
    rx.open(maxRxItems)
    result.tx = tx.addr
    result.rx = rx.addr
  
  proc twist*[R,T](ch: WorkerChannel[R,T]) : WorkerChannel[T,R] =
    WorkerChannel[T,R](tx: ch.rx, rx: ch.tx)
  
  proc send*[R,T](ch: WorkerChannel[R,T], x: T, retyInterval: int = 100): Future[void] {.async.} =
    while not ch.tx[].trySend x:
      await sleepAsync(retyInterval)
  
  proc sendSync*[R,T](ch: WorkerChannel[R,T], x: T) = ch.tx[].send x
  
  proc recv*[R,T](ch: WorkerChannel[R,T], maxDelay: int = 100): Future[R] {.async.} =
    while true:
      let (avail, msg)= ch.rx[].tryRecv
      if avail: return msg
      await sleepAsync(maxDelay)
  
  proc recvSync*[R,T](ch: WorkerChannel[R,T]): R = ch.rx[].recv
  
  proc newWorker*[R,T](taskFunc: Worker[R,T], maxTxItems: int = 0, maxRxItems: int = 0): WorkerChannel[T,R] =
    ## Worker recieve txType and send rxType
    var th : Thread[WorkerChannel[R, T]]
    var ch = newWorkerChannel[R, T](maxTxItems=maxTxItems, maxRxItems=maxRxItems)
    createThread(th, taskFunc, ch)
    ch.twist()
  
  # ---------------------------------------------------------
  # sync
  
  proc taskSync(ch: WorkerChannel[int,string]) {.thread.} =
    # echo ch.recvSync
    for i in 1 .. 10:
      ch.sendSync $i
  
  proc worker() =
    # var ch = newWorker(taskSync)
    var th : Thread[WorkerChannel[int, string]]
    var chan = newWorkerChannel[int, string]()
    createThread(th, taskSync, chan)
    var ch = chan.twist()
    
    ## this line lead to Illegal storage access
    # ch.sendSync 10
    while true:
      echo ch.recvSync
  
  worker()
  
  # ---------------------------------------------------------
  # async
  
  proc taskAsync(ch: WorkerChannel[int,string]) {.async.} =
    var i = 0
    while true:
      echo "sending ", i, " on thread=", getThreadId()
      await ch.send($i)
      inc i
      await sleepAsync(500)
  
  proc task(ch: WorkerChannel[int,string]) {.thread.} =
    waitFor taskAsync(ch)
  
  proc main() {.async.} =
    # var chan = newAsyncWorker[int, string](task)
    var chan = newWorker(task)
    while true:
      var i = await chan.recv
      echo "recevied ", i
      await sleepAsync(100)
  
  ## If I try to run task on main thread, this works.
  # waitFor taskAsync(newWorkerChannel[int, string]()) # work
  # task(newWorkerChannel[int, string]()) # work
  
  ## If I run on threads, some time I got Illegal storage access, some time I got exit after two recv
  # asyncCheck main()
  # try: runForever()
  # except: discard
  
  
  Run

---