pgm_receiver.cpp 8.39 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
{
70
    LIBZMQ_UNUSED (io_thread_);
Martin Sustrik's avatar
Martin Sustrik committed
71
    //  Retrieve PGM fds and start polling.
72 73
    fd_t socket_fd = retired_fd;
    fd_t waiting_pipe_fd = retired_fd;
74 75 76 77 78 79
    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);

80
    session = session_;
81 82 83

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

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

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

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

105
    session = NULL;
106 107
}

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

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

119
void zmq::pgm_receiver_t::restart_input ()
Martin Hurton's avatar
Martin Hurton committed
120
{
121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142
    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;
143
            LIBZMQ_DELETE(it->second.decoder);
144
            insize = 0;
145
        }
146 147 148 149 150 151
    }

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

152
    active_tsi = NULL;
153
    in_event ();
Martin Hurton's avatar
Martin Hurton committed
154
}
155 156 157 158 159

const char *zmq::pgm_receiver_t::get_endpoint () const
{
    return "";
}
Martin Hurton's avatar
Martin Hurton committed
160

161 162
void zmq::pgm_receiver_t::in_event ()
{
Martin Sustrik's avatar
Martin Sustrik committed
163
    // Read data from the underlying pgm_socket.
malosek's avatar
malosek committed
164
    const pgm_tsi_t *tsi = NULL;
165

166 167 168 169 170
    if (has_rx_timer) {
        cancel_timer (rx_timer_id);
        has_rx_timer = false;
    }

171 172 173 174 175
    //  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.
176 177 178
        //  Note the workaround made not to break strict-aliasing rules.
        void *tmp = NULL;
        ssize_t received = pgm_socket.receive (&tmp, &tsi);
179
        inpos = (unsigned char*) tmp;
180 181 182

        //  No data to process. This may happen if the packet received is
        //  neither ODATA nor ODATA.
183
        if (received == 0) {
184
            if (errno == ENOMEM || errno == EBUSY) {
185 186 187 188
                const long timeout = pgm_socket.get_rx_timeout ();
                add_timer (timeout, rx_timer_id);
                has_rx_timer = true;
            }
189
            break;
190
        }
191 192 193 194 195 196

        //  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) {
197 198 199
            if (it != peers.end ()) {
                it->second.joined = false;
                if (it->second.decoder != NULL) {
200
                    LIBZMQ_DELETE(it->second.decoder);
201
                }
202 203
            }
            break;
malosek's avatar
malosek committed
204
        }
205

206 207 208
        //  New peer. Add it to the list of know but unjoint peers.
        if (it == peers.end ()) {
            peer_info_t peer_info = {false, NULL};
209
            it = peers.insert (peers_t::value_type (*tsi, peer_info)).first;
210
        }
211

212 213
        insize = static_cast <size_t> (received);

214
        //  Read the offset of the fist message in the current packet.
215 216 217 218
        zmq_assert (insize >= sizeof (uint16_t));
        uint16_t offset = get_uint16 (inpos);
        inpos += sizeof (uint16_t);
        insize -= sizeof (uint16_t);
219

220 221
        //  Join the stream if needed.
        if (!it->second.joined) {
222

223 224 225 226
            //  There is no beginning of the message in current packet.
            //  Ignore the data.
            if (offset == 0xffff)
                continue;
227

228
            zmq_assert (offset <= insize);
229
            zmq_assert (it->second.decoder == NULL);
230

231
            //  We have to move data to the beginning of the first message.
232 233
            inpos += offset;
            insize -= offset;
malosek's avatar
malosek committed
234

235 236
            //  Mark the stream as joined.
            it->second.joined = true;
malosek's avatar
malosek committed
237

238
            //  Create and connect decoder for the peer.
239 240
            it->second.decoder = new (std::nothrow)
                v1_decoder_t (0, options.maxmsgsize);
241
            alloc_assert (it->second.decoder);
242
        }
malosek's avatar
malosek committed
243

244 245 246 247 248 249 250 251 252 253
        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;
254 255
            }

256
            it->second.joined = false;
257
            LIBZMQ_DELETE(it->second.decoder);
258
            insize = 0;
259
        }
Martin Sustrik's avatar
Martin Sustrik committed
260
    }
261 262

    //  Flush any messages decoder may have produced.
263
    session->flush ();
264
}
malosek's avatar
malosek committed
265

266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288
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;
}


289 290 291 292
void zmq::pgm_receiver_t::timer_event (int token)
{
    zmq_assert (token == rx_timer_id);

293 294
    //  Timer cancels on return by poller_base.
    has_rx_timer = false;
295 296 297
    in_event ();
}

298 299 300
void zmq::pgm_receiver_t::drop_subscriptions ()
{
    msg_t msg;
301
    msg.init ();
302
    while (session->pull_msg (&msg) == 0)
303 304 305
        msg.close ();
}

306 307
#endif