atomic-queue.c

00001 // ----------------------------------------------------------------------------
00002 // Copyright 2015-2017 ARM Ltd.
00003 //
00004 // SPDX-License-Identifier: Apache-2.0
00005 //
00006 // Licensed under the Apache License, Version 2.0 (the "License");
00007 // you may not use this file except in compliance with the License.
00008 // You may obtain a copy of the License at
00009 //
00010 //     http://www.apache.org/licenses/LICENSE-2.0
00011 //
00012 // Unless required by applicable law or agreed to in writing, software
00013 // distributed under the License is distributed on an "AS IS" BASIS,
00014 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
00015 // See the License for the specific language governing permissions and
00016 // limitations under the License.
00017 // ----------------------------------------------------------------------------
00018 
00019 #include "atomic-queue/atomic-queue.h"
00020 #include "atomic.h"
00021 
00022 #include <stddef.h>
00023 #include <stdint.h>
00024 #include <stdio.h>
00025 
00026 #define CORE_UTIL_ASSERT_MSG(test, msg)
00027 
00028 
00029 int aq_element_take(struct atomic_queue_element *e, void *ctx)
00030 {
00031     if (!e) { return ATOMIC_QUEUE_NULL_ELEMENT;}
00032     if (!ctx) { return ATOMIC_QUEUE_INVALID_CONTEXT;}
00033     // Duplicate element check using lock
00034     uintptr_t lock; // This is initialized in the do/while loop.
00035     // Check/obtain a lock on the element.
00036     do {
00037         lock = e->lock;
00038         if (lock) {
00039             return ATOMIC_QUEUE_DUPLICATE_ELEMENT;
00040         }
00041     } while (!aq_atomic_cas_uintptr(&e->lock, lock, (uintptr_t)ctx));
00042     return ATOMIC_QUEUE_SUCCESS;
00043 }
00044 
00045 int aq_element_release(struct atomic_queue_element *e, void **ctx)
00046 {
00047     if (!e) { return ATOMIC_QUEUE_NULL_ELEMENT;}
00048     if (!ctx) { return ATOMIC_QUEUE_INVALID_CONTEXT;}
00049     *ctx = (void *) e->lock;
00050     e->lock = 0;
00051     return ATOMIC_QUEUE_SUCCESS;
00052 }
00053 
00054 int aq_push_tail(struct atomic_queue *q, struct atomic_queue_element *e)
00055 {
00056     CORE_UTIL_ASSERT_MSG(q != NULL, "null queue used");
00057     if (!e) { return ATOMIC_QUEUE_NULL_ELEMENT;}
00058     if (q == NULL) {
00059         return ATOMIC_QUEUE_NULL_QUEUE;
00060     }
00061 
00062     do {
00063         e->next = q->tail;
00064     } while (!aq_atomic_cas_uintptr((uintptr_t *)&q->tail, (uintptr_t)e->next, (uintptr_t)e));
00065 
00066     return ATOMIC_QUEUE_SUCCESS;
00067 }
00068 
00069 struct atomic_queue_element *aq_pop_head(struct atomic_queue *q)
00070 {
00071     CORE_UTIL_ASSERT_MSG(q != NULL, "null queue used");
00072     if (q == NULL) {
00073         return NULL;
00074     }
00075     struct atomic_queue_element *current;
00076     int fail = AQ_ATOMIC_CAS_DEREF_VALUE;
00077     while (fail != AQ_ATOMIC_CAS_DEREF_SUCCESS) {
00078         // Set the element reference pointer to the tail pointer
00079         struct atomic_queue_element *volatile *px = &q->tail;
00080         if (*px == NULL) {
00081             return NULL;
00082         }
00083         fail = AQ_ATOMIC_CAS_DEREF_VALUE;
00084         while (fail == AQ_ATOMIC_CAS_DEREF_VALUE) {
00085             fail = aq_atomic_cas_deref_uintptr((uintptr_t *volatile *)px,
00086                                                (uintptr_t **)&current,
00087                                                (uintptr_t) NULL,
00088                                                NULL,
00089                                                offsetof(struct atomic_queue_element, next));
00090             if (fail == AQ_ATOMIC_CAS_DEREF_VALUE) {
00091                 // Detect a loop to the tail of the queue
00092                 if (current->next == q->tail) {
00093                     return NULL;
00094                 }
00095                 px = &current->next;
00096             }
00097         }
00098     }
00099 
00100     return current;
00101 }
00102 
00103 
00104 int aq_empty(struct atomic_queue *q)
00105 {
00106     return q->tail == NULL;
00107 }
00108 
00109 unsigned aq_count(struct atomic_queue *q)
00110 {
00111     unsigned x;
00112     struct atomic_queue_element *volatile e;
00113     if (aq_empty(q)) {
00114         return 0;
00115     }
00116     e = q->tail;
00117     for (x = 1; e->next != NULL; x++, e = e->next) {
00118         if (e->next == q->tail) {
00119             return (unsigned) - 1;
00120         }
00121     }
00122     return x;
00123 }
00124 
00125 void aq_initialize_element(struct atomic_queue_element *e)
00126 {
00127     if (!e) { return;}
00128     e->lock = 0;
00129     e->next = NULL;
00130 }