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 | |