Mirror of forum.nim-lang.org

11580 :: Run a proc stored in an object in another thread with a threadpool

• Isofruit (orginal) [2024-05-10T15:02:33+02:00] view original

  Heyho, I am trying to run a {.nimcall.} proc that is stored inside the field of an object on another thread using taskpool/weave_io or whatever other functional threadpool we have.

  I'm absolutely failing on getting it to work however.

  Code-wise, this roughly represents what I want:

  import weave_io
  import std/isolation
  
  type Actor = ref object
    field1: int
    field2: int
    command: proc(x,y: int)
  
  proc runner(x, y: int) {.raises: [], gcsafe.} =
    echo x, y
  
  let actor = Actor(field1: 3, field2: 4, command: runner)
  proc run(actor: Actor) =
    actor.command(actor.field1, actor.field2)
  
  var tp: Threadpool = Threadpool.new()
  tp.spawn run(actor)
  tp.shutdown()
  Run

  field1 and 2 down the line will turn Table[int, pointer] but that is for another time. For now I simply want this to execute.

  The problem is that regardless of how I slice it, I can't find any way to execute a proc this way:

  • With tp.spawn actor.command(actor.field1, actor.field2) the spawn macro of both taskpool and weave_io (both having the exact same issue) explodes, unable to understand this syntax
  • With the above example it complains that run can throw an exception, but even if you add a try-catch you can't make it safe enough because passing in actor isn't safe. Even if run instead takes in Isolated[Actor] (because at this point I do not care, I'll deep copy everything if I have to) it is impossible, because to extract from an Isolated[T] it must be var and it is impossible to capture a var isolatedActor in weaves spawn macro because that isn't safe

  The only alternative that I'm aware of is malebolgia but that shares the same problems at first glance - It requires the parameters to be isolatable for safety reasons, the problem here is that I know its safe but I can't circumvent these precautions.

---

• mratsim (orginal) [2024-05-10T15:37:43+02:00] view original

  I have no doc in weave-io because I didn't change the internals to optimize for IO 😅

  Do you have code? You say nimcall but it;s

  type Actor = ref object
    field1: int
    field2: int
    command: proc(x,y: int)
  Run

  instead of

  
  type Actor = ref object
    field1: int
    field2: int
    command: proc(x,y: int) {.nimcall.}
  Run

---

• Isofruit (orginal) [2024-05-10T16:34:10+02:00] view original

  I just messed up the example here. I tried a lot of permutations over the past few hours, including some where the spawn call was abstracted in a runIn proc, and more.

  The vast majority of the time I was testing with command being {.nimcall.} before I experimented with "command" being a generic and forcing people to inherit from Actor (which it turns out was very meh for different reasons an why I didn't follow that approach further).

  If you want a full blown view of what I'm working on (It's not like I've written that much code yet). The actual meat is at the bottom of the code, that's a functional example, 2 examples that don't work and one that works if I keep Actor as value type as in this slice of code.

  # typeTable.nim - Base-type for storing pointers of various different types in a table, some form of "controlled" type erasure in a sense
  import std/[tables, hashes]
  
  type TypeId = distinct pointer
  proc `==`*(x, y: TypeId): bool {.borrow.}
  proc hash*(x: TypeId): Hash {.borrow.}
  proc getTypeId[T](typ: typedesc[T]): TypeId =
    var info {.global.}: int
    return TypeId(info.addr)
  
  type TypeTable* = distinct Table[TypeId, pointer]
  proc hasKey*[T](table: TypeTable, key: typedesc[T]): bool =
    let innerKey = key.getTypeId()
    return Table[TypeId, pointer](table).hasKey(innerKey)
  
  proc `[]`*[T](table: TypeTable; typ: typedesc[T]): T =
    let key = typ.getTypeId()
    let valuePtr = Table[TypeId, pointer](table)[key]
    return cast[T](valuePtr)
  
  proc `[]=`*[T: ptr](table: var TypeTable, key: typedesc[T], value: T) =
    let key = T.getTypeId()
    let valuePtr: pointer = value
    Table[TypeId, pointer](table)[key] = valuePtr
  
  # mailbox.nim - abstracts over Channels to make them easily storable and retrievable from typeTable
  import threading/channels
  import std/[options, isolation]
  
  type Mailbox*[T] = ptr Chan[T]
  
  proc newMailbox*[T](capacity: int): Mailbox[T] =
    let mailbox = createShared(Chan[T])
    mailbox[] = newChan[T](capacity)
    return mailbox
  
  proc destroyMailbox*[T](mailbox: Mailbox[T]) =
    freeShared(mailbox)
  
  proc send*[T](mailbox: Mailbox[T], value: T) =
    mailbox[].send(
      unsafeIsolate(deepCopy(value))
    )
  
  proc trySend*[T](mailbox: Mailbox[T], value: T): bool =
    return mailbox[].trySend(
      unsafeIsolate(deepCopy(value))
    )
  
  proc recv*[T](mailbox: Mailbox[T]): T = mailbox[].recv()
  
  proc tryRecv*[T](mailbox: Mailbox[T]): Option[T] =
    var msg: T
    let hasMsg = mailbox[].tryRecv(msg)
    if hasMsg:
      return some(msg)
    else:
      return none(T)
  
  proc peek*[T](mailbox: Mailbox[T]): int = mailbox[].peek()
  
  # mailboxTable.nim - abstracts over typeTable yet again to be more specifically about Mailboxes.
  type MailboxTable* = distinct TypeTable
  
  proc hasMailbox*[T](table: MailboxTable, typ: typedesc[T]): bool =
    return TypeTable(table).hasKey(Mailbox[T])
  
  proc `[]`*[T](table: MailboxTable, typ: typedesc[T]): Mailbox[T] =
    return TypeTable(table)[Mailbox[T]]
  
  proc `[]=`*[T](table: var MailboxTable, typ: typedesc[T], mailbox: Mailbox[T]) =
    TypeTable(table)[Mailbox[T]] = mailbox
  
  # actor.nim
  import std/options
  
  type Actor* = object
    targetMailboxes*: MailboxTable
    sourceMailboxes*: MailboxTable
    command*: proc(sources, targets: MailboxTable) {.nimcall.}
  
  proc newActor*(command: proc(sources: MailboxTable, targets: MailboxTable) {.nimcall.}): Actor =
    return Actor(command: command)
  
  proc hasTarget*[T](actor: Actor, targetTyp: typedesc[T]): bool =
    return actor.targetMailboxes.hasMailbox(T)
  
  proc addTarget*[T](actor: var Actor, mailbox: Mailbox[T]) =
    actor.targetMailboxes[T] = mailbox
  
  proc getTarget*[T](actor: Actor, typ: typedesc[T]): Option[Mailbox[T]] =
    if actor.hasTarget(T):
      let value = actor.targetMailboxes[T]
      return some(value)
    else:
      return none(Mailbox[T])
  
  proc hasSource[T](actor: Actor, sourceTyp: typedesc[T]): bool =
    return actor.sourceMailboxes.hasMailbox(T)
  
  proc addSource*[T](actor: var Actor, mailbox: Mailbox[T]) =
    actor.sourceMailboxes[T] = mailbox
  
  proc getSource*[T](actor: Actor, typ: typedesc[T]): Option[Mailbox[T]] =
    if actor.hasSource(T):
      let value = actor.sourceMailboxes[T]
      return some(value)
    else:
      return none(Mailbox[T])
  
  
  
  
  
  ## EXAMPLE ##
  import std/os
  import taskpools
  
  let threads = 4
  var tp = Taskpool.new(num_threads = threads)
  
  type A = ref object
    name: string
  
  proc run(sources: MailboxTable, targets: MailboxTable) {.raises:[], nimcall, gcsafe.} =
      sleep(1000)
      try:
        echo sources[A].recv().repr
      except KeyError:
        echo "Busted Keyerror"
  
  var act = newActor(run)
  
  
  let source = newMailbox[A](1)
  act.addSource(source)
  let msg = A(name: "test")
  
  
  # Example 1 - Works, but not the syntax I want
  tp.spawn run(act.sourceMailboxes, act.targetMailboxes)
  source.send(msg)
  
  # Example 2 - Does not work because spawn macro ("Invalid node kind nnkDotExpr for macros.`$`")
  # 2.1
  # proc runActor(actor: Actor, pool: Taskpool) =
  #   pool.spawn act.command(act.sourceMailboxes, act.targetMailboxes)
  # runActor(act, tp)
  # source.send(msg)
  
  # 2.2
  # proc runActor(actor: Actor, pool: Taskpool) =
  #   let runner = actor.command
  #   pool.spawn runner(act.sourceMailboxes, act.targetMailboxes)
  # runActor(act, tp)
  # source.send(msg)
  
  # Example 3 - Works 50% - Does not work with Actor being ref type, does work with Actor being value type
  # proc run(actor: Actor) {.gcsafe, raises: [].} =
  #   try:
  #     echo "Running"
  #     {.gcsafe.}: actor.command(actor.sourceMailboxes, actor.targetMailboxes)
  #   except:
  #     echo "Error"
  # tp.spawn act.run()
  # source.send(msg)
  
  
  # Cleanup
  tp.syncAll()
  tp.shutDown()
  Run

---