Line | Count | Source |
1 | | /* SPDX-License-Identifier: MPL-2.0 */ |
2 | | |
3 | | #include "precompiled.hpp" |
4 | | #include "fq.hpp" |
5 | | #include "pipe.hpp" |
6 | | #include "err.hpp" |
7 | | #include "msg.hpp" |
8 | | |
9 | 0 | zmq::fq_t::fq_t () : _active (0), _current (0), _more (false) |
10 | 0 | { |
11 | 0 | } |
12 | | |
13 | | zmq::fq_t::~fq_t () |
14 | 0 | { |
15 | 0 | zmq_assert (_pipes.empty ()); |
16 | 0 | } |
17 | | |
18 | | void zmq::fq_t::attach (pipe_t *pipe_) |
19 | 0 | { |
20 | 0 | _pipes.push_back (pipe_); |
21 | 0 | _pipes.swap (_active, _pipes.size () - 1); |
22 | 0 | _active++; |
23 | 0 | } |
24 | | |
25 | | void zmq::fq_t::pipe_terminated (pipe_t *pipe_) |
26 | 0 | { |
27 | 0 | const pipes_t::size_type index = _pipes.index (pipe_); |
28 | | |
29 | | // Remove the pipe from the list; adjust number of active pipes |
30 | | // accordingly. |
31 | 0 | if (index < _active) { |
32 | 0 | _active--; |
33 | 0 | _pipes.swap (index, _active); |
34 | 0 | if (_current == _active) |
35 | 0 | _current = 0; |
36 | 0 | } |
37 | 0 | _pipes.erase (pipe_); |
38 | 0 | } |
39 | | |
40 | | void zmq::fq_t::activated (pipe_t *pipe_) |
41 | 0 | { |
42 | | // Move the pipe to the list of active pipes. |
43 | 0 | _pipes.swap (_pipes.index (pipe_), _active); |
44 | 0 | _active++; |
45 | 0 | } |
46 | | |
47 | | int zmq::fq_t::recv (msg_t *msg_) |
48 | 0 | { |
49 | 0 | return recvpipe (msg_, NULL); |
50 | 0 | } |
51 | | |
52 | | int zmq::fq_t::recvpipe (msg_t *msg_, pipe_t **pipe_) |
53 | 0 | { |
54 | | // Deallocate old content of the message. |
55 | 0 | int rc = msg_->close (); |
56 | 0 | errno_assert (rc == 0); |
57 | | |
58 | | // Round-robin over the pipes to get the next message. |
59 | 0 | while (_active > 0) { |
60 | | // Try to fetch new message. If we've already read part of the message |
61 | | // subsequent part should be immediately available. |
62 | 0 | const bool fetched = _pipes[_current]->read (msg_); |
63 | | |
64 | | // Note that when message is not fetched, current pipe is deactivated |
65 | | // and replaced by another active pipe. Thus we don't have to increase |
66 | | // the 'current' pointer. |
67 | 0 | if (fetched) { |
68 | 0 | if (pipe_) |
69 | 0 | *pipe_ = _pipes[_current]; |
70 | 0 | _more = (msg_->flags () & msg_t::more) != 0; |
71 | 0 | if (!_more) { |
72 | 0 | _current = (_current + 1) % _active; |
73 | 0 | } |
74 | 0 | return 0; |
75 | 0 | } |
76 | | |
77 | | // Check the atomicity of the message. |
78 | | // If we've already received the first part of the message |
79 | | // we should get the remaining parts without blocking. |
80 | 0 | zmq_assert (!_more); |
81 | |
|
82 | 0 | _active--; |
83 | 0 | _pipes.swap (_current, _active); |
84 | 0 | if (_current == _active) |
85 | 0 | _current = 0; |
86 | 0 | } |
87 | | |
88 | | // No message is available. Initialise the output parameter |
89 | | // to be a 0-byte message. |
90 | 0 | rc = msg_->init (); |
91 | 0 | errno_assert (rc == 0); |
92 | 0 | errno = EAGAIN; |
93 | 0 | return -1; |
94 | 0 | } |
95 | | |
96 | | bool zmq::fq_t::has_in () |
97 | 0 | { |
98 | | // There are subsequent parts of the partly-read message available. |
99 | 0 | if (_more) |
100 | 0 | return true; |
101 | | |
102 | | // Note that messing with current doesn't break the fairness of fair |
103 | | // queueing algorithm. If there are no messages available current will |
104 | | // get back to its original value. Otherwise it'll point to the first |
105 | | // pipe holding messages, skipping only pipes with no messages available. |
106 | 0 | while (_active > 0) { |
107 | 0 | if (_pipes[_current]->check_read ()) |
108 | 0 | return true; |
109 | | |
110 | | // Deactivate the pipe. |
111 | 0 | _active--; |
112 | 0 | _pipes.swap (_current, _active); |
113 | 0 | if (_current == _active) |
114 | 0 | _current = 0; |
115 | 0 | } |
116 | | |
117 | 0 | return false; |
118 | 0 | } |