| 1 | /* |
| 2 | Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file |
| 3 | |
| 4 | This file is part of libzmq, the ZeroMQ core engine in C++. |
| 5 | |
| 6 | libzmq is free software; you can redistribute it and/or modify it under |
| 7 | the terms of the GNU Lesser General Public License (LGPL) as published |
| 8 | by the Free Software Foundation; either version 3 of the License, or |
| 9 | (at your option) any later version. |
| 10 | |
| 11 | As a special exception, the Contributors give you permission to link |
| 12 | this library with independent modules to produce an executable, |
| 13 | regardless of the license terms of these independent modules, and to |
| 14 | copy and distribute the resulting executable under terms of your choice, |
| 15 | provided that you also meet, for each linked independent module, the |
| 16 | terms and conditions of the license of that module. An independent |
| 17 | module is a module which is not derived from or based on this library. |
| 18 | If you modify this library, you must extend this exception to your |
| 19 | version of the library. |
| 20 | |
| 21 | libzmq is distributed in the hope that it will be useful, but WITHOUT |
| 22 | ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or |
| 23 | FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public |
| 24 | License for more details. |
| 25 | |
| 26 | You should have received a copy of the GNU Lesser General Public License |
| 27 | along with this program. If not, see <http://www.gnu.org/licenses/>. |
| 28 | */ |
| 29 | |
| 30 | #ifndef __ZMQ_YQUEUE_HPP_INCLUDED__ |
| 31 | #define __ZMQ_YQUEUE_HPP_INCLUDED__ |
| 32 | |
| 33 | #include <stdlib.h> |
| 34 | #include <stddef.h> |
| 35 | |
| 36 | #include "err.hpp" |
| 37 | #include "atomic_ptr.hpp" |
| 38 | |
| 39 | namespace zmq |
| 40 | { |
| 41 | // yqueue is an efficient queue implementation. The main goal is |
| 42 | // to minimise number of allocations/deallocations needed. Thus yqueue |
| 43 | // allocates/deallocates elements in batches of N. |
| 44 | // |
| 45 | // yqueue allows one thread to use push/back function and another one |
| 46 | // to use pop/front functions. However, user must ensure that there's no |
| 47 | // pop on the empty queue and that both threads don't access the same |
| 48 | // element in unsynchronised manner. |
| 49 | // |
| 50 | // T is the type of the object in the queue. |
| 51 | // N is granularity of the queue (how many pushes have to be done till |
| 52 | // actual memory allocation is required). |
| 53 | #ifdef HAVE_POSIX_MEMALIGN |
| 54 | // ALIGN is the memory alignment size to use in the case where we have |
| 55 | // posix_memalign available. Default value is 64, this alignment will |
| 56 | // prevent two queue chunks from occupying the same CPU cache line on |
| 57 | // architectures where cache lines are <= 64 bytes (e.g. most things |
| 58 | // except POWER). It is detected at build time to try to account for other |
| 59 | // platforms like POWER and s390x. |
| 60 | template <typename T, int N, size_t ALIGN = ZMQ_CACHELINE_SIZE> class yqueue_t |
| 61 | #else |
| 62 | template <typename T, int N> class yqueue_t |
| 63 | #endif |
| 64 | { |
| 65 | public: |
| 66 | // Create the queue. |
| 67 | inline yqueue_t () |
| 68 | { |
| 69 | _begin_chunk = allocate_chunk (); |
| 70 | alloc_assert (_begin_chunk); |
| 71 | _begin_pos = 0; |
| 72 | _back_chunk = NULL; |
| 73 | _back_pos = 0; |
| 74 | _end_chunk = _begin_chunk; |
| 75 | _end_pos = 0; |
| 76 | } |
| 77 | |
| 78 | // Destroy the queue. |
| 79 | inline ~yqueue_t () |
| 80 | { |
| 81 | while (true) { |
| 82 | if (_begin_chunk == _end_chunk) { |
| 83 | free (_begin_chunk); |
| 84 | break; |
| 85 | } |
| 86 | chunk_t *o = _begin_chunk; |
| 87 | _begin_chunk = _begin_chunk->next; |
| 88 | free (o); |
| 89 | } |
| 90 | |
| 91 | chunk_t *sc = _spare_chunk.xchg (NULL); |
| 92 | free (sc); |
| 93 | } |
| 94 | |
| 95 | // Returns reference to the front element of the queue. |
| 96 | // If the queue is empty, behaviour is undefined. |
| 97 | inline T &front () { return _begin_chunk->values[_begin_pos]; } |
| 98 | |
| 99 | // Returns reference to the back element of the queue. |
| 100 | // If the queue is empty, behaviour is undefined. |
| 101 | inline T &back () { return _back_chunk->values[_back_pos]; } |
| 102 | |
| 103 | // Adds an element to the back end of the queue. |
| 104 | inline void push () |
| 105 | { |
| 106 | _back_chunk = _end_chunk; |
| 107 | _back_pos = _end_pos; |
| 108 | |
| 109 | if (++_end_pos != N) |
| 110 | return; |
| 111 | |
| 112 | chunk_t *sc = _spare_chunk.xchg (NULL); |
| 113 | if (sc) { |
| 114 | _end_chunk->next = sc; |
| 115 | sc->prev = _end_chunk; |
| 116 | } else { |
| 117 | _end_chunk->next = allocate_chunk (); |
| 118 | alloc_assert (_end_chunk->next); |
| 119 | _end_chunk->next->prev = _end_chunk; |
| 120 | } |
| 121 | _end_chunk = _end_chunk->next; |
| 122 | _end_pos = 0; |
| 123 | } |
| 124 | |
| 125 | // Removes element from the back end of the queue. In other words |
| 126 | // it rollbacks last push to the queue. Take care: Caller is |
| 127 | // responsible for destroying the object being unpushed. |
| 128 | // The caller must also guarantee that the queue isn't empty when |
| 129 | // unpush is called. It cannot be done automatically as the read |
| 130 | // side of the queue can be managed by different, completely |
| 131 | // unsynchronised thread. |
| 132 | inline void unpush () |
| 133 | { |
| 134 | // First, move 'back' one position backwards. |
| 135 | if (_back_pos) |
| 136 | --_back_pos; |
| 137 | else { |
| 138 | _back_pos = N - 1; |
| 139 | _back_chunk = _back_chunk->prev; |
| 140 | } |
| 141 | |
| 142 | // Now, move 'end' position backwards. Note that obsolete end chunk |
| 143 | // is not used as a spare chunk. The analysis shows that doing so |
| 144 | // would require free and atomic operation per chunk deallocated |
| 145 | // instead of a simple free. |
| 146 | if (_end_pos) |
| 147 | --_end_pos; |
| 148 | else { |
| 149 | _end_pos = N - 1; |
| 150 | _end_chunk = _end_chunk->prev; |
| 151 | free (_end_chunk->next); |
| 152 | _end_chunk->next = NULL; |
| 153 | } |
| 154 | } |
| 155 | |
| 156 | // Removes an element from the front end of the queue. |
| 157 | inline void pop () |
| 158 | { |
| 159 | if (++_begin_pos == N) { |
| 160 | chunk_t *o = _begin_chunk; |
| 161 | _begin_chunk = _begin_chunk->next; |
| 162 | _begin_chunk->prev = NULL; |
| 163 | _begin_pos = 0; |
| 164 | |
| 165 | // 'o' has been more recently used than _spare_chunk, |
| 166 | // so for cache reasons we'll get rid of the spare and |
| 167 | // use 'o' as the spare. |
| 168 | chunk_t *cs = _spare_chunk.xchg (o); |
| 169 | free (cs); |
| 170 | } |
| 171 | } |
| 172 | |
| 173 | private: |
| 174 | // Individual memory chunk to hold N elements. |
| 175 | struct chunk_t |
| 176 | { |
| 177 | T values[N]; |
| 178 | chunk_t *prev; |
| 179 | chunk_t *next; |
| 180 | }; |
| 181 | |
| 182 | inline chunk_t *allocate_chunk () |
| 183 | { |
| 184 | #ifdef HAVE_POSIX_MEMALIGN |
| 185 | void *pv; |
| 186 | if (posix_memalign (&pv, ALIGN, sizeof (chunk_t)) == 0) |
| 187 | return (chunk_t *) pv; |
| 188 | return NULL; |
| 189 | #else |
| 190 | return (chunk_t *) malloc (sizeof (chunk_t)); |
| 191 | #endif |
| 192 | } |
| 193 | |
| 194 | // Back position may point to invalid memory if the queue is empty, |
| 195 | // while begin & end positions are always valid. Begin position is |
| 196 | // accessed exclusively be queue reader (front/pop), while back and |
| 197 | // end positions are accessed exclusively by queue writer (back/push). |
| 198 | chunk_t *_begin_chunk; |
| 199 | int _begin_pos; |
| 200 | chunk_t *_back_chunk; |
| 201 | int _back_pos; |
| 202 | chunk_t *_end_chunk; |
| 203 | int _end_pos; |
| 204 | |
| 205 | // People are likely to produce and consume at similar rates. In |
| 206 | // this scenario holding onto the most recently freed chunk saves |
| 207 | // us from having to call malloc/free. |
| 208 | atomic_ptr_t<chunk_t> _spare_chunk; |
| 209 | |
| 210 | ZMQ_NON_COPYABLE_NOR_MOVABLE (yqueue_t) |
| 211 | }; |
| 212 | } |
| 213 | |
| 214 | #endif |
| 215 | |