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_YPIPE_HPP_INCLUDED__
31#define __ZMQ_YPIPE_HPP_INCLUDED__
32
33#include "atomic_ptr.hpp"
34#include "yqueue.hpp"
35#include "ypipe_base.hpp"
36
37namespace zmq
38{
39// Lock-free queue implementation.
40// Only a single thread can read from the pipe at any specific moment.
41// Only a single thread can write to the pipe at any specific moment.
42// T is the type of the object in the queue.
43// N is granularity of the pipe, i.e. how many items are needed to
44// perform next memory allocation.
45
46template <typename T, int N> class ypipe_t : public ypipe_base_t<T>
47{
48 public:
49 // Initialises the pipe.
50 inline ypipe_t ()
51 {
52 // Insert terminator element into the queue.
53 _queue.push ();
54
55 // Let all the pointers to point to the terminator.
56 // (unless pipe is dead, in which case c is set to NULL).
57 _r = _w = _f = &_queue.back ();
58 _c.set (&_queue.back ());
59 }
60
61 // Following function (write) deliberately copies uninitialised data
62 // when used with zmq_msg. Initialising the VSM body for
63 // non-VSM messages won't be good for performance.
64
65#ifdef ZMQ_HAVE_OPENVMS
66#pragma message save
67#pragma message disable(UNINIT)
68#endif
69
70 // Write an item to the pipe. Don't flush it yet. If incomplete is
71 // set to true the item is assumed to be continued by items
72 // subsequently written to the pipe. Incomplete items are never
73 // flushed down the stream.
74 inline void write (const T &value_, bool incomplete_)
75 {
76 // Place the value to the queue, add new terminator element.
77 _queue.back () = value_;
78 _queue.push ();
79
80 // Move the "flush up to here" poiter.
81 if (!incomplete_)
82 _f = &_queue.back ();
83 }
84
85#ifdef ZMQ_HAVE_OPENVMS
86#pragma message restore
87#endif
88
89 // Pop an incomplete item from the pipe. Returns true if such
90 // item exists, false otherwise.
91 inline bool unwrite (T *value_)
92 {
93 if (_f == &_queue.back ())
94 return false;
95 _queue.unpush ();
96 *value_ = _queue.back ();
97 return true;
98 }
99
100 // Flush all the completed items into the pipe. Returns false if
101 // the reader thread is sleeping. In that case, caller is obliged to
102 // wake the reader up before using the pipe again.
103 inline bool flush ()
104 {
105 // If there are no un-flushed items, do nothing.
106 if (_w == _f)
107 return true;
108
109 // Try to set 'c' to 'f'.
110 if (_c.cas (_w, _f) != _w) {
111 // Compare-and-swap was unseccessful because 'c' is NULL.
112 // This means that the reader is asleep. Therefore we don't
113 // care about thread-safeness and update c in non-atomic
114 // manner. We'll return false to let the caller know
115 // that reader is sleeping.
116 _c.set (_f);
117 _w = _f;
118 return false;
119 }
120
121 // Reader is alive. Nothing special to do now. Just move
122 // the 'first un-flushed item' pointer to 'f'.
123 _w = _f;
124 return true;
125 }
126
127 // Check whether item is available for reading.
128 inline bool check_read ()
129 {
130 // Was the value prefetched already? If so, return.
131 if (&_queue.front () != _r && _r)
132 return true;
133
134 // There's no prefetched value, so let us prefetch more values.
135 // Prefetching is to simply retrieve the
136 // pointer from c in atomic fashion. If there are no
137 // items to prefetch, set c to NULL (using compare-and-swap).
138 _r = _c.cas (&_queue.front (), NULL);
139
140 // If there are no elements prefetched, exit.
141 // During pipe's lifetime r should never be NULL, however,
142 // it can happen during pipe shutdown when items
143 // are being deallocated.
144 if (&_queue.front () == _r || !_r)
145 return false;
146
147 // There was at least one value prefetched.
148 return true;
149 }
150
151 // Reads an item from the pipe. Returns false if there is no value.
152 // available.
153 inline bool read (T *value_)
154 {
155 // Try to prefetch a value.
156 if (!check_read ())
157 return false;
158
159 // There was at least one value prefetched.
160 // Return it to the caller.
161 *value_ = _queue.front ();
162 _queue.pop ();
163 return true;
164 }
165
166 // Applies the function fn to the first elemenent in the pipe
167 // and returns the value returned by the fn.
168 // The pipe mustn't be empty or the function crashes.
169 inline bool probe (bool (*fn_) (const T &))
170 {
171 bool rc = check_read ();
172 zmq_assert (rc);
173
174 return (*fn_) (_queue.front ());
175 }
176
177 protected:
178 // Allocation-efficient queue to store pipe items.
179 // Front of the queue points to the first prefetched item, back of
180 // the pipe points to last un-flushed item. Front is used only by
181 // reader thread, while back is used only by writer thread.
182 yqueue_t<T, N> _queue;
183
184 // Points to the first un-flushed item. This variable is used
185 // exclusively by writer thread.
186 T *_w;
187
188 // Points to the first un-prefetched item. This variable is used
189 // exclusively by reader thread.
190 T *_r;
191
192 // Points to the first item to be flushed in the future.
193 T *_f;
194
195 // The single point of contention between writer and reader thread.
196 // Points past the last flushed item. If it is NULL,
197 // reader is asleep. This pointer should be always accessed using
198 // atomic operations.
199 atomic_ptr_t<T> _c;
200
201 ZMQ_NON_COPYABLE_NOR_MOVABLE (ypipe_t)
202};
203}
204
205#endif
206