pgm_receiver.cpp 8.29 KB
Newer Older
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 7 8
    libzmq is free software; you can redistribute it and/or modify it under
    the terms of the GNU Lesser General Public License (LGPL) as published
    by the Free Software Foundation; either version 3 of the License, or
9 10
    (at your option) any later version.

11 12 13 14 15 16 17 18 19 20 21 22 23 24
    As a special exception, the Contributors give you permission to link
    this library with independent modules to produce an executable,
    regardless of the license terms of these independent modules, and to
    copy and distribute the resulting executable under terms of your choice,
    provided that you also meet, for each linked independent module, the
    terms and conditions of the license of that module. An independent
    module is a module which is not derived from or based on this library.
    If you modify this library, you must extend this exception to your
    version of the library.

    libzmq is distributed in the hope that it will be useful, but WITHOUT
    ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
    FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
    License for more details.
25

26
    You should have received a copy of the GNU Lesser General Public License
27 28 29
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

30
#include "precompiled.hpp"
31
#include "macros.hpp"
32

malosek's avatar
malosek committed
33
#if defined ZMQ_HAVE_OPENPGM
34

35 36
#include <new>

37
#include "pgm_receiver.hpp"
38
#include "session_base.hpp"
39
#include "v1_decoder.hpp"
40 41
#include "stdint.hpp"
#include "wire.hpp"
42
#include "err.hpp"
43

44
zmq::pgm_receiver_t::pgm_receiver_t (class io_thread_t *parent_,
45
      const options_t &options_) :
46
    io_object_t (parent_),
47
    has_rx_timer (false),
48 49
    pgm_socket (true, options_),
    options (options_),
50
    session (NULL),
51 52
    active_tsi (NULL),
    insize (0)
53 54 55 56 57
{
}

zmq::pgm_receiver_t::~pgm_receiver_t ()
{
malosek's avatar
malosek committed
58 59
    //  Destructor should not be called before unplug.
    zmq_assert (peers.empty ());
60 61
}

62
int zmq::pgm_receiver_t::init (bool udp_encapsulation_, const char *network_)
63
{
64
    return pgm_socket.init (udp_encapsulation_, network_);
65 66
}

67 68
void zmq::pgm_receiver_t::plug (io_thread_t *io_thread_,
    session_base_t *session_)
69
{
Martin Sustrik's avatar
Martin Sustrik committed
70
    //  Retrieve PGM fds and start polling.
71 72
    fd_t socket_fd = retired_fd;
    fd_t waiting_pipe_fd = retired_fd;
73 74 75 76 77 78
    pgm_socket.get_receiver_fds (&socket_fd, &waiting_pipe_fd);
    socket_handle = add_fd (socket_fd);
    pipe_handle = add_fd (waiting_pipe_fd);
    set_pollin (pipe_handle);
    set_pollin (socket_handle);

79
    session = session_;
80 81 82

    //  If there are any subscriptions already queued in the session, drop them.
    drop_subscriptions ();
83 84 85 86
}

void zmq::pgm_receiver_t::unplug ()
{
malosek's avatar
malosek committed
87
    //  Delete decoders.
88
    for (peers_t::iterator it = peers.begin (); it != peers.end (); ++it) {
89 90 91
        if (it->second.decoder != NULL) {
            LIBZMQ_DELETE(it->second.decoder);
        }
malosek's avatar
malosek committed
92 93
    }
    peers.clear ();
94
    active_tsi = NULL;
95

96 97 98 99 100
    if (has_rx_timer) {
        cancel_timer (rx_timer_id);
        has_rx_timer = false;
    }

101 102
    rm_fd (socket_handle);
    rm_fd (pipe_handle);
Martin Sustrik's avatar
Martin Sustrik committed
103

104
    session = NULL;
105 106
}

107 108 109 110 111 112
void zmq::pgm_receiver_t::terminate ()
{
    unplug ();
    delete this;
}

113
void zmq::pgm_receiver_t::restart_output ()
114
{
115
    drop_subscriptions ();
116 117
}

118
void zmq::pgm_receiver_t::restart_input ()
Martin Hurton's avatar
Martin Hurton committed
119
{
120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141
    zmq_assert (session != NULL);
    zmq_assert (active_tsi != NULL);

    const peers_t::iterator it = peers.find (*active_tsi);
    zmq_assert (it != peers.end ());
    zmq_assert (it->second.joined);

    //  Push the pending message into the session.
    int rc = session->push_msg (it->second.decoder->msg ());
    errno_assert (rc == 0);

    if (insize > 0) {
        rc = process_input (it->second.decoder);
        if (rc == -1) {
            //  HWM reached; we will try later.
            if (errno == EAGAIN) {
                session->flush ();
                return;
            }
            //  Data error. Delete message decoder, mark the
            //  peer as not joined and drop remaining data.
            it->second.joined = false;
142
            LIBZMQ_DELETE(it->second.decoder);
143
            insize = 0;
144
        }
145 146 147 148 149 150
    }

    //  Resume polling.
    set_pollin (pipe_handle);
    set_pollin (socket_handle);

151
    active_tsi = NULL;
152
    in_event ();
Martin Hurton's avatar
Martin Hurton committed
153 154
}

155 156
void zmq::pgm_receiver_t::in_event ()
{
Martin Sustrik's avatar
Martin Sustrik committed
157
    // Read data from the underlying pgm_socket.
malosek's avatar
malosek committed
158
    const pgm_tsi_t *tsi = NULL;
159

160 161 162 163 164
    if (has_rx_timer) {
        cancel_timer (rx_timer_id);
        has_rx_timer = false;
    }

165 166 167 168 169
    //  TODO: This loop can effectively block other engines in the same I/O
    //  thread in the case of high load.
    while (true) {

        //  Get new batch of data.
170 171 172
        //  Note the workaround made not to break strict-aliasing rules.
        void *tmp = NULL;
        ssize_t received = pgm_socket.receive (&tmp, &tsi);
173
        inpos = (unsigned char*) tmp;
174 175 176

        //  No data to process. This may happen if the packet received is
        //  neither ODATA nor ODATA.
177
        if (received == 0) {
178
            if (errno == ENOMEM || errno == EBUSY) {
179 180 181 182
                const long timeout = pgm_socket.get_rx_timeout ();
                add_timer (timeout, rx_timer_id);
                has_rx_timer = true;
            }
183
            break;
184
        }
185 186 187 188 189 190

        //  Find the peer based on its TSI.
        peers_t::iterator it = peers.find (*tsi);

        //  Data loss. Delete decoder and mark the peer as disjoint.
        if (received == -1) {
191 192 193
            if (it != peers.end ()) {
                it->second.joined = false;
                if (it->second.decoder != NULL) {
194
                    LIBZMQ_DELETE(it->second.decoder);
195
                }
196 197
            }
            break;
malosek's avatar
malosek committed
198
        }
199

200 201 202
        //  New peer. Add it to the list of know but unjoint peers.
        if (it == peers.end ()) {
            peer_info_t peer_info = {false, NULL};
203
            it = peers.insert (peers_t::value_type (*tsi, peer_info)).first;
204
        }
205

206 207
        insize = static_cast <size_t> (received);

208
        //  Read the offset of the fist message in the current packet.
209 210 211 212
        zmq_assert (insize >= sizeof (uint16_t));
        uint16_t offset = get_uint16 (inpos);
        inpos += sizeof (uint16_t);
        insize -= sizeof (uint16_t);
213

214 215
        //  Join the stream if needed.
        if (!it->second.joined) {
216

217 218 219 220
            //  There is no beginning of the message in current packet.
            //  Ignore the data.
            if (offset == 0xffff)
                continue;
221

222
            zmq_assert (offset <= insize);
223
            zmq_assert (it->second.decoder == NULL);
224

225
            //  We have to move data to the beginning of the first message.
226 227
            inpos += offset;
            insize -= offset;
malosek's avatar
malosek committed
228

229 230
            //  Mark the stream as joined.
            it->second.joined = true;
malosek's avatar
malosek committed
231

232
            //  Create and connect decoder for the peer.
233 234
            it->second.decoder = new (std::nothrow)
                v1_decoder_t (0, options.maxmsgsize);
235
            alloc_assert (it->second.decoder);
236
        }
malosek's avatar
malosek committed
237

238 239 240 241 242 243 244 245 246 247
        int rc = process_input (it->second.decoder);
        if (rc == -1) {
            if (errno == EAGAIN) {
                active_tsi = tsi;

                //  Stop polling.
                reset_pollin (pipe_handle);
                reset_pollin (socket_handle);

                break;
248 249
            }

250
            it->second.joined = false;
251
            LIBZMQ_DELETE(it->second.decoder);
252
            insize = 0;
253
        }
Martin Sustrik's avatar
Martin Sustrik committed
254
    }
255 256

    //  Flush any messages decoder may have produced.
257
    session->flush ();
258
}
malosek's avatar
malosek committed
259

260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282
int zmq::pgm_receiver_t::process_input (v1_decoder_t *decoder)
{
    zmq_assert (session != NULL);

    while (insize > 0) {
        size_t n = 0;
        int rc = decoder->decode (inpos, insize, n);
        if (rc == -1)
            return -1;
        inpos += n;
        insize -= n;
        if (rc == 0)
            break;
        rc = session->push_msg (decoder->msg ());
        if (rc == -1) {
            errno_assert (errno == EAGAIN);
            return -1;
        }
    }
    return 0;
}


283 284 285 286
void zmq::pgm_receiver_t::timer_event (int token)
{
    zmq_assert (token == rx_timer_id);

287 288
    //  Timer cancels on return by poller_base.
    has_rx_timer = false;
289 290 291
    in_event ();
}

292 293 294
void zmq::pgm_receiver_t::drop_subscriptions ()
{
    msg_t msg;
295
    msg.init ();
296
    while (session->pull_msg (&msg) == 0)
297 298 299
        msg.close ();
}

300 301
#endif