Shared Queue

sharedqueue.nim

import locks, options

# 
# Simple shared queue implementation
# Translated into Nim from https://idea.popcount.org/2012-09-11-concurrent-queue-in-c/
#


type
  SharedQueueNode[A] = ptr object
    item: A
    next: SharedQueueNode[A]

  SharedQueue*[A] = object
    inq: SharedQueueNode[A]
    outq: SharedQueueNode[A]
    lock: Lock


proc allocSharedQueueNode[A](): SharedQueueNode[A] =
  result = cast[type result](allocShared0(sizeof(result[])))
  
proc deallocSharedQueueNode[A](item: SharedQueueNode[A]) =
  deallocShared(item)

proc enqueueNode[A](q: var SharedQueue[A], item: SharedQueueNode[A]) =
  while true:
    let inq = q.inq
    item.next = inq
    if cas(addr q.inq, inq, item):
      break

proc enqueue*[A](q: var SharedQueue[A], item: A) =
  let qitem = allocSharedQueueNode[A]()
  qitem.item = item
  q.enqueueNode(qitem)


proc initSharedQueue*[A](q: var SharedQueue[A]) =
  q.inq = nil
  q.outq = nil
  initLock q.lock
  
proc deinitSharedQueue*[A](q: var SharedQueue[A]) =
  var item = q.dequeue()
  while item.isSome:
    item = q.dequeue()

proc dequeueNode[A](q: var SharedQueue[A]): SharedQueueNode[A] =
  withLock q.lock:
    if q.outq == nil:
      while true:
        var head = q.inq
        if head == nil:
          break
        if cas(addr q.inq, head, nil):
          while head != nil:
            let next = head.next
            head.next = q.outq
            q.outq = head
            head = next
          break

    result = q.outq
    if result != nil:
      q.outq = result.next

proc dequeue*[A](q: var SharedQueue[A]): Option[A] =
  let item =  q.dequeueNode()
  if item == nil:
    result = none(A)
  else:
    result = some[A](item.item)
    deallocSharedQueueNode(item)


when isMainModule:

  proc producer(q: ptr SharedQueue[int]) =
    q[].enqueue(10)
    q[].enqueue(20)
    q[].enqueue(30)
    q[].enqueue(20)

 
  proc consumer(q: ptr SharedQueue[int]) =
    echo q[].dequeue()
    echo q[].dequeue()
    echo q[].dequeue()
    echo q[].dequeue()
    echo q[].dequeue()
    echo q[].dequeue()

  let q = cast[ptr SharedQueue[int]](allocShared0(sizeof(SharedQueue[int])))

  var t1,t2: Thread[ptr SharedQueue[int]]

  initSharedQueue(q[])

  createThread(t1, producer, q)
  createThread(t2, consumer, q)

  joinThreads(t1,t2)
  

I've written another approach for shared structure. That shared structure I used for solving sleeping barbershop problem