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#include "precompiled.hpp"
31#include "fq.hpp"
32#include "pipe.hpp"
33#include "err.hpp"
34#include "msg.hpp"
35
36zmq::fq_t::fq_t () : _active (0), _last_in (NULL), _current (0), _more (false)
37{
38}
39
40zmq::fq_t::~fq_t ()
41{
42 zmq_assert (_pipes.empty ());
43}
44
45void zmq::fq_t::attach (pipe_t *pipe_)
46{
47 _pipes.push_back (pipe_);
48 _pipes.swap (_active, _pipes.size () - 1);
49 _active++;
50}
51
52void zmq::fq_t::pipe_terminated (pipe_t *pipe_)
53{
54 const pipes_t::size_type index = _pipes.index (pipe_);
55
56 // Remove the pipe from the list; adjust number of active pipes
57 // accordingly.
58 if (index < _active) {
59 _active--;
60 _pipes.swap (index, _active);
61 if (_current == _active)
62 _current = 0;
63 }
64 _pipes.erase (pipe_);
65
66 if (_last_in == pipe_) {
67 _last_in = NULL;
68 }
69}
70
71void zmq::fq_t::activated (pipe_t *pipe_)
72{
73 // Move the pipe to the list of active pipes.
74 _pipes.swap (_pipes.index (pipe_), _active);
75 _active++;
76}
77
78int zmq::fq_t::recv (msg_t *msg_)
79{
80 return recvpipe (msg_, NULL);
81}
82
83int zmq::fq_t::recvpipe (msg_t *msg_, pipe_t **pipe_)
84{
85 // Deallocate old content of the message.
86 int rc = msg_->close ();
87 errno_assert (rc == 0);
88
89 // Round-robin over the pipes to get the next message.
90 while (_active > 0) {
91 // Try to fetch new message. If we've already read part of the message
92 // subsequent part should be immediately available.
93 bool fetched = _pipes[_current]->read (msg_);
94
95 // Note that when message is not fetched, current pipe is deactivated
96 // and replaced by another active pipe. Thus we don't have to increase
97 // the 'current' pointer.
98 if (fetched) {
99 if (pipe_)
100 *pipe_ = _pipes[_current];
101 _more = (msg_->flags () & msg_t::more) != 0;
102 if (!_more) {
103 _last_in = _pipes[_current];
104 _current = (_current + 1) % _active;
105 }
106 return 0;
107 }
108
109 // Check the atomicity of the message.
110 // If we've already received the first part of the message
111 // we should get the remaining parts without blocking.
112 zmq_assert (!_more);
113
114 _active--;
115 _pipes.swap (_current, _active);
116 if (_current == _active)
117 _current = 0;
118 }
119
120 // No message is available. Initialise the output parameter
121 // to be a 0-byte message.
122 rc = msg_->init ();
123 errno_assert (rc == 0);
124 errno = EAGAIN;
125 return -1;
126}
127
128bool zmq::fq_t::has_in ()
129{
130 // There are subsequent parts of the partly-read message available.
131 if (_more)
132 return true;
133
134 // Note that messing with current doesn't break the fairness of fair
135 // queueing algorithm. If there are no messages available current will
136 // get back to its original value. Otherwise it'll point to the first
137 // pipe holding messages, skipping only pipes with no messages available.
138 while (_active > 0) {
139 if (_pipes[_current]->check_read ())
140 return true;
141
142 // Deactivate the pipe.
143 _active--;
144 _pipes.swap (_current, _active);
145 if (_current == _active)
146 _current = 0;
147 }
148
149 return false;
150}
151