mcs-queue: initial checkin

The "new" queue (i.e., "MCS Queue") code directly from:

  ftp://ftp.cs.rochester.edu/pub/packages/sched_conscious_synch/concurrent_queues/concurrent_queues.tar.gz

Except for the MIPS assembly implementation files and Makefile.

From README:

The compressed tar file in this directory conatins the C code for
the algorithms implemented for the paper

%A Maged M. Michael
%A Michael L. Scott
%T Simple, Fast, and Practical Non-Blocking and Blocking
Concurrent Queue Algorithms
%J PROC of the Fifteenth PODC
%C Philadelphia, PA
%D May 1996
%X 23-26 May 1996
%K tr600
%O Earlier version available as
TR 600,
URCSD,
December 1995

diff --git a/mcs-queue/args.c b/mcs-queue/args.c
new file mode 100644 (file)
index 0000000..dba95cb
--- /dev/null
+++ b/mcs-queue/args.c
@@ -0,0 +1,33 @@
+#include "main.h"
+
+extern unsigned backoff_base_bits;
+extern unsigned backoff_cap_bits;
+extern unsigned iterations;
+extern unsigned multi;
+extern unsigned initial_nodes;
+extern unsigned procs;
+extern unsigned repetitions;
+extern unsigned backoff_shift_bits;
+extern unsigned work;
+
+void 
+parse_args(int argc,char **argv)
+{
+extern char * optarg; 
+int c; 
+
+  while((c=getopt(argc,argv,"b:c:i:m:n:p:r:s:w:"))!=EOF)
+    switch(c){
+    case 'b':  backoff_base_bits = atoi(optarg); break;
+    case 'c':  backoff_cap_bits = atoi(optarg); break;
+    case 'i':  iterations = atoi(optarg); break;
+    case 'm':  multi = atoi(optarg);  break;
+    case 'n':  initial_nodes = atoi(optarg);  break;
+    case 'p':   procs = atoi(optarg);   break;
+    case 'r':   repetitions = atoi(optarg);   break;
+    case 's':   backoff_shift_bits = atoi(optarg);   break;
+    case 'w':   work = atoi(optarg);   break;
+    default: 
+      assert(0);
+    }
+}

diff --git a/mcs-queue/backoff.c b/mcs-queue/backoff.c
new file mode 100644 (file)
index 0000000..b3f3602
--- /dev/null
+++ b/mcs-queue/backoff.c
@@ -0,0 +1,27 @@
+extern unsigned backoff;
+extern unsigned backoff_base_bits;
+extern unsigned backoff_cap_bits;
+extern unsigned backoff_shift_bits;
+extern unsigned backoff_base;
+extern unsigned backoff_cap;
+extern unsigned backoff_addend;
+
+void
+init_backoff()
+{
+  backoff_base = (1<<backoff_base_bits)-1;
+  backoff_cap = (1<<backoff_cap_bits)-1;
+  backoff_addend = (1<<backoff_shift_bits)-1;
+}
+
+unsigned
+backoff_delay()
+{
+  unsigned i;
+  
+  for (i=0; i<backoff; i++) ;
+  backoff <<= backoff_shift_bits;
+  backoff += backoff_addend;
+  backoff &= backoff_cap;
+  return i;
+}

diff --git a/mcs-queue/main.c b/mcs-queue/main.c
new file mode 100644 (file)
index 0000000..98ca29a
--- /dev/null
+++ b/mcs-queue/main.c
@@ -0,0 +1,148 @@
+#include "main.h"
+
+#define NUM_PROCESSORS                 12
+
+struct tms tim;
+struct tms tim1;
+
+usptr_t *Handle;
+barrier_t *Barrier;
+usptr_t *lock_handle;
+ulock_t native_lock;
+
+int shmid;
+
+unsigned pid;
+unsigned backoff;
+unsigned backoff_base;
+unsigned backoff_cap;
+unsigned backoff_addend;
+char* name = "";
+unsigned backoff_base_bits = 0;
+unsigned backoff_cap_bits = 0;
+unsigned procs = 1;
+unsigned multi = 1;
+unsigned iterations = 1;
+unsigned initial_nodes = 0;
+unsigned repetitions = 1;
+unsigned backoff_shift_bits = 0;
+unsigned work = 0;
+private_t private;
+shared_mem_t *smp;
+
+void
+native_acquire ()
+{
+  ussetlock(native_lock);
+} 
+
+void
+native_release ()
+{
+  usunsetlock(native_lock);
+}
+
+void
+tts_acq(unsigned* plock)
+{
+  do {
+    if (*plock == 1) {
+      backoff = backoff_base;
+      do {
+       backoff_delay();
+      } while(*plock == 1);
+    }
+  } while (tas(plock) == 1);
+}
+
+void 
+time_test ()
+{
+  unsigned i,j;
+  struct tms time_val;
+  clock_t t1, t2;
+  unsigned val;
+
+  if(pid==0) {
+    init_queue();
+  }
+  init_memory();
+  init_private();
+  init_backoff();
+  barrier(Barrier, procs*multi);
+  if(pid==0)
+  {
+    t1=times(&time_val);
+  }
+  for(i=0;i<iterations;i++) {
+    val = private.value;
+    enqueue(val);
+    for(j=0; j<work;) j++;
+    val = dequeue();
+    for(j=0; j<work;) j++;
+    private.value++;
+  }
+  barrier(Barrier, procs*multi);
+  if(pid==0)
+  {
+    t2=times(&time_val);
+    printf("p%d  m%d  i%d  b%d c%d s%d  w%d time %.0f ms.\n",
+          procs, multi, iterations*procs*multi,
+          backoff_base_bits, backoff_cap_bits,
+          backoff_shift_bits, work, ((t2-t1)*1000)/(double)HZ);
+    fflush(stdout);
+  }
+}
+
+void main_task()
+{
+  unsigned processor;
+  unsigned i;
+
+  processor = (pid/multi)+1;
+  processor %= NUM_PROCESSORS;
+  if(sysmp(MP_MUSTRUN, processor) == -1) { perror("Could not MUSTRUN"); }
+  if (pid==0) {
+    printf("--- %s\n", name);
+    fflush(stdout);
+  }
+  for (i=0; i<repetitions; i++) {
+    time_test();
+  }
+}
+
+void  setup_shmem()
+{
+  shmid = shmget(IPC_PRIVATE, sizeof(shared_mem_t), 511);
+  assert(shmid != -1);
+  smp = (shared_mem_t *)shmat(shmid, 0, 0);
+  assert((int)smp != -1);
+}
+
+void my_m_fork(void (*func)(),int num_procs)
+{
+  for (pid=1;pid<num_procs;pid++) {
+    if(fork()==0) /* Child */ {
+      (*func)(); /* Call the program */
+      return;
+    }
+  }
+  pid=0;
+  (*func)(); /* Call the program */
+}
+
+main(int argc,char **argv)
+{
+  parse_args(argc, argv);
+  name = argv[0];
+  iterations = (iterations + ((procs*multi)>>1))/(procs*multi);
+  setup_shmem();
+  Handle = usinit("/tmp/foo_barrier");
+  Barrier = new_barrier(Handle);
+  init_barrier(Barrier);
+  lock_handle = usinit("/tmp/foo_lock");
+  native_lock = usnewlock(lock_handle);
+  my_m_fork(main_task, procs*multi); 
+  shmctl(shmid, IPC_RMID, smp);
+  exit(0);
+}

diff --git a/mcs-queue/main.h b/mcs-queue/main.h
new file mode 100644 (file)
index 0000000..8ec3cf9
--- /dev/null
+++ b/mcs-queue/main.h
@@ -0,0 +1,13 @@
+#include <stdio.h>
+#include <sys/param.h>
+#include <sys/types.h>
+#include <sys/times.h>
+#include <sys/types.h>
+#include <sys/ipc.h>
+#include <sys/shm.h>
+#include <signal.h>
+#include <sys/cachectl.h>
+#include <sys/sysmp.h>
+#include <ulocks.h>
+#include <assert.h>
+#include "my_queue.h"

diff --git a/mcs-queue/my_queue.c b/mcs-queue/my_queue.c
new file mode 100644 (file)
index 0000000..a7a3ee2
--- /dev/null
+++ b/mcs-queue/my_queue.c
@@ -0,0 +1,143 @@
+#include "main.h"
+
+extern unsigned pid;
+extern unsigned iterations;
+extern unsigned initial_nodes;
+extern unsigned backoff;
+extern unsigned backoff_base;
+extern private_t private;
+extern shared_mem_t* smp;
+
+void
+init_private()
+{
+  private.node = 2 + initial_nodes + pid;
+  private.value = 1 + initial_nodes + (pid * iterations);
+
+}
+
+void
+init_memory()
+{
+}
+
+unsigned
+new_node()
+{
+  return private.node;
+}
+
+void
+reclaim(unsigned node)
+{
+  private.node = node;
+}
+
+void
+init_queue()
+{
+  unsigned i;
+
+  /* initialize queue */
+  smp->head.sep.ptr = 1;
+  smp->head.sep.count = 0;
+  smp->tail.sep.ptr = 1;
+  smp->tail.sep.count = 0;
+  smp->nodes[1].next.sep.ptr = NULL;
+  smp->nodes[1].next.sep.count = 0;
+  
+  /* initialize avail list */
+  for (i=2; i<MAX_NODES; i++) {
+    smp->nodes[i].next.sep.ptr = i+1;
+    smp->nodes[i].next.sep.count = 0;
+  }
+  smp->nodes[MAX_NODES].next.sep.ptr = NULL;
+  smp->nodes[MAX_NODES].next.sep.count = 0;
+  
+  /* initialize queue contents */
+  if (initial_nodes > 0) {
+    for (i=2; i<initial_nodes+2; i++) {
+      smp->nodes[i].value = i;
+      smp->nodes[i-1].next.sep.ptr = i;
+      smp->nodes[i].next.sep.ptr = NULL;
+    }
+    smp->head.sep.ptr = 1;
+    smp->tail.sep.ptr = 1 + initial_nodes;    
+  }
+}
+
+void
+enqueue(unsigned val)
+{
+  unsigned success;
+  unsigned node;
+  pointer_t tail;
+  pointer_t next;
+
+  node = new_node();
+  smp->nodes[node].value = val;
+  smp->nodes[node].next.sep.ptr = NULL;
+
+  backoff = backoff_base;
+  for (success = FALSE; success == FALSE; ) {
+    tail.con = smp->tail.con;
+    next.con = smp->nodes[tail.sep.ptr].next.con;
+    if (tail.con == smp->tail.con) {
+      if (next.sep.ptr == NULL) {
+       backoff = backoff_base;
+       success = cas(&smp->nodes[tail.sep.ptr].next, 
+                     next.con,
+                     MAKE_LONG(node, next.sep.count+1));
+      }
+      if (success == FALSE) {
+       cas(&smp->tail,
+           tail.con,
+           MAKE_LONG(smp->nodes[tail.sep.ptr].next.sep.ptr,
+                     tail.sep.count+1));
+       backoff_delay();
+      }
+    }
+  }
+  cas(&smp->tail, 
+      tail.con,
+      MAKE_LONG(node, tail.sep.count+1));
+}
+
+unsigned
+dequeue()
+{
+  unsigned value;
+  unsigned success;
+  pointer_t head;
+  pointer_t tail;
+  pointer_t next;
+
+  backoff = backoff_base;
+  for (success = FALSE; success == FALSE; ) {
+    head.con = smp->head.con;
+    tail.con = smp->tail.con;
+    next.con = smp->nodes[head.sep.ptr].next.con;
+    if (smp->head.con == head.con) {
+      if (head.sep.ptr == tail.sep.ptr) {
+       if (next.sep.ptr == NULL) {
+         return NULL;
+       }
+       cas(&smp->tail,
+           tail.con,
+           MAKE_LONG(next.sep.ptr, tail.sep.count+1));
+       backoff_delay();
+      } else {
+       value = smp->nodes[next.sep.ptr].value;
+       success = cas(&smp->head,
+                     head.con,
+                     MAKE_LONG(next.sep.ptr, head.sep.count+1));
+       if (success == FALSE) {
+         backoff_delay();
+       }
+      }
+    }
+  }
+  reclaim(head.sep.ptr);
+  return value;
+}
+

diff --git a/mcs-queue/my_queue.h b/mcs-queue/my_queue.h
new file mode 100644 (file)
index 0000000..34c9cbd
--- /dev/null
+++ b/mcs-queue/my_queue.h
@@ -0,0 +1,38 @@
+#define TRUE                           1
+#define FALSE                          0
+#define NULL                           0
+
+#define MAX_NODES                      0xff
+#define MAX_SERIAL                     10000
+
+#define MAKE_LONG(lo, hi)              ((hi)<<16)+(lo)
+
+typedef union pointer {
+  struct {
+    volatile unsigned short count;
+    volatile unsigned short ptr;
+  } sep;
+  volatile unsigned long con;
+}pointer_t;
+
+typedef struct node {
+  unsigned value;
+  pointer_t next;
+  unsigned foo[30];
+} node_t;
+
+typedef struct private {
+  unsigned node;
+  unsigned value;
+  unsigned serial[MAX_SERIAL];
+} private_t;
+
+typedef struct shared_mem {
+  pointer_t head;
+  unsigned foo1[31];
+  pointer_t tail;
+  unsigned foo2[31];
+  node_t nodes[MAX_NODES+1];
+  unsigned serial;
+} shared_mem_t;
+