#include #include #include #include /* global variable for compare and exchange */ pthread_mutex_t CE_MUTEX; // compare and exchange /* This c code is porting from C# version The Empty Exception class: 1 class EmptyException : Exception {} The Node Class: 1 public class Node where T : class { 2 public T val; 3 public Node next; 4 public Node (T val) { 5 this.val = val; 6 this.next = null; 7 } 8 } */ /* Node and Queue structure */ typedef struct NODE { void *val; struct NODE *next; } Node; typedef struct LOCKFREEQUEUE { Node *head, *tail; } LockFreeQueue; /* 1 [TestClass] 2 public class QueueTest { 3 public Struct EnqContainer { 4 String first; 5 String second; 6 LockFreeQueue q; 7 } */ typedef struct ENQCONTAINER { char *first; char *second; LockFreeQueue *queue; } EnqContainer; /* function prototype */ Node *NewNode(void *init_val); LockFreeQueue *NewLockFreeQueue(); void InitCE (pthread_mutex_t *ce_lock); void DestroyCE (pthread_mutex_t *ce_lock); int CompareAndExchange (void **destination, void *value, void *comparand); void LockFreeQueue_enq(LockFreeQueue *this_queue, void *x); void *LockFreeQueue_deq(LockFreeQueue *this_queue); void LockFreeQueue_enq_WithError1 (LockFreeQueue *this_queue, void *x); void LockFreeQueue_enq_WithError2 (LockFreeQueue *this_queue, void *x); void *EnqTwoStrings (void *contain); void *EnqTwoStrings_WithError1 (void *contain); void *EnqTwoStrings_WithError2 (void *contain); void *DeqThreeStrings (void *lfq); Node *NewNode(void *init_val) { Node *new_node = (Node*) malloc(sizeof(Node)); new_node->val = init_val; new_node->next = NULL; return new_node; } /* The LockFreeQueue Class: 1 public class LockFreeQueue where T : class { 2 public Node head, tail; 3 public LockFreeQueue () { 4 Node node = new Node(); 5 head = tail = node; 6 } 7 */ LockFreeQueue *NewLockFreeQueue() { Node *init_head_tail = NewNode(NULL); LockFreeQueue *new_queue = (LockFreeQueue*) malloc(sizeof(LockFreeQueue)); new_queue->head = new_queue->tail = init_head_tail; return new_queue; } /* Interlocked.CompareExchange( 17 ref last.next,node,next)) { 18 Interlocked.CompareExchange( 19 ref tail, node, last); */ /* compare and exchange functions */ void InitCE (pthread_mutex_t *ce_lock) { pthread_mutex_init(ce_lock, NULL); return; } void DestroyCE (pthread_mutex_t *ce_lock) { pthread_mutex_destroy(ce_lock); return; } // return 1 for true and 0 for fasle int CompareAndExchange (void **destination, void *value, void *comparand) { int success = 0; pthread_mutex_lock(&CE_MUTEX); if ((*destination) == comparand) { (*destination) = value; success = 1; } pthread_mutex_unlock(&CE_MUTEX); return success; } /* 29 30 public T deq() { 31 while (true) { 32 Node first = head; 33 Node last = tail; 34 Node next = first.next; 35 if (first == head) { 36 if (first == last) { 37 if (next == null) { 38 throw new EmptyException(); 39 } 40 Interlocked.CompareExchange( 41 ref tail,next,last); 42 } else { 43 T val = next.val; 44 if (first == 45 Interlocked.CompareExchange( 46 ref head,next,first)) 47 return val; 48 } 49 } 50 } 51 } 52 } */ /* lock free queue dequeue */ // return NULL for nothing void *LockFreeQueue_deq(LockFreeQueue *this_queue) { Node *first = NULL; Node *last = NULL; Node *next = NULL; void *value = NULL; while (1) { first = this_queue->head; last = this_queue->tail; next = first->next; if (first == this_queue->head) { if (first == last) { if (next == NULL) { fprintf(stderr, "Nothing to deq\n"); return NULL; //exit(1); } CompareAndExchange(&(this_queue->tail), next, last); } else { value = next->val; if ( CompareAndExchange(&(this_queue->head), next, first) ) return value; } } } return NULL; } /* 8 public void enq(T x) { 9 Node node = new Node(x); 10 while (true) { 11 Node last = tail; 12 Node next = last.next; 13 if (last == tail) { 14 if (next == null) { 15 if (next == 16 Interlocked.CompareExchange( 17 ref last.next,node,next)) { 18 Interlocked.CompareExchange( 19 ref tail, node, last); 20 return; 21 } 22 } else { 23 Interlocked.CompareExchange( 24 ref tail, next, last); 25 } 26 } ================================================ original java code 15 if (last == tail.get()) { 16 if (next == null) { 17 if (last.next.compareAndSet( 18 next, node)) { 19 tail.compareAndSet(last, node); 20 return; 21 } 22 } else { 23 tail.compareAndSet(last, next); 24 } 25 } ================================================ 27 } 28 } */ /* lock free queue enqueue */ void LockFreeQueue_enq(LockFreeQueue *this_queue, void *x) { Node *last = NULL; Node *next = NULL; Node *node = NewNode(x); while (1) { last = this_queue->tail; next = last->next; if (last == this_queue->tail) { if (next == NULL) { if ( CompareAndExchange(&(this_queue->tail->next), node, next) ) { CompareAndExchange(&(this_queue->tail), node, last); return; } } else { CompareAndExchange(&(this_queue->tail), next, last); } } } return; } /* 8 9 public void EnqTwoStrings (Object o) { 10 EnqContainer e = (EnqContainer)o; 11 e.q.enq(e.first); 12 e.q.enq(e.second); 13 } */ void *EnqTwoStrings (void *contain) { LockFreeQueue_enq (((EnqContainer*)contain)->queue, ((EnqContainer*)contain)->first); LockFreeQueue_enq (((EnqContainer*)contain)->queue, ((EnqContainer*)contain)->second); return NULL; } /* 15 public void DequeueThreeStrings (Object o) { 16 LockFreeQueue lfq = (LockFreeQueue)o; 17 try { lfq.q.deq() } catch (EmptyException e) {} 18 try { lfq.q.deq() } catch (EmptyException e) {} 19 try { lfq.q.deq() } catch (EmptyException e) {} 20 } */ void *DeqThreeStrings (void *lfq) { LockFreeQueue_deq((LockFreeQueue*)lfq); LockFreeQueue_deq((LockFreeQueue*)lfq); LockFreeQueue_deq((LockFreeQueue*)lfq); return NULL; } /* 22 [TestMethod] 23 [HostType("Chess")] 24 [TestProperty("ChessMonitorVolatiles","True")] 25 [TestProperty("ChessBound","4")] 26 public void TestLockFreeQueueWithError2(){ 27 QueueTest qt = new QueueTest(); 28 LockFreeQueue lfq = 29 new LockFreeQueueWithError2(); 30 EnqContainer firstParams, secondParams; 31 firstParams.q = secondParams.q = lfq; 32 firstParams.first = "a"; 33 firstParams.second = "b"; 34 secondParams.first = "c"; 35 secondParams.second = "d"; 36 Thread t2 = new Thread( 37 new ParameterizedThreadStart( 38 qt.EnqTwoStrings)); 39 Thread t3 = new Thread( 40 new ParameterizedThreadStart( 41 qt.DeqThreeStrings)); 42 t2.start(secondParams); 43 t3.start(lfq); 44 QueueTwoItems(firstParams); 45 t2.join(); 46 t3.join(); 47 } 48 } */ void TestLockFreeQueueWithError2 () { pthread_t t1, t2, t3; LockFreeQueue *lfq = NewLockFreeQueue(); EnqContainer firstParams, secondParams; firstParams.queue = secondParams.queue = lfq; firstParams.first = "a"; firstParams.second = "b"; secondParams.first = "c"; secondParams.second = "d"; pthread_create(&t1, NULL, EnqTwoStrings, &firstParams); pthread_create(&t2, NULL, EnqTwoStrings, &secondParams); pthread_create(&t3, NULL, DeqThreeStrings, lfq); pthread_join(t1, NULL); pthread_join(t2, NULL); pthread_join(t3, NULL); Node *current = lfq->head->next; while (current != NULL) { printf("node: %s\n", (char*)current->val); current = current->next; } } int main (void) { InitCE(&CE_MUTEX); TestLockFreeQueueWithError2(); DestroyCE(&CE_MUTEX); return 0; }