pgm_receiver.cpp 8.34 KB
Newer Older
1
/*
2
    Copyright (c) 2007-2015 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 "macros.hpp"
31 32
#include "platform.hpp"

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

35 36
#include <new>

37 38 39 40
#ifdef ZMQ_HAVE_WINDOWS
#include "windows.hpp"
#endif

41
#include "pgm_receiver.hpp"
42
#include "session_base.hpp"
43
#include "v1_decoder.hpp"
44 45
#include "stdint.hpp"
#include "wire.hpp"
46
#include "err.hpp"
47 48

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

zmq::pgm_receiver_t::~pgm_receiver_t ()
{
malosek's avatar
malosek committed
62 63
    //  Destructor should not be called before unplug.
    zmq_assert (peers.empty ());
64 65
}

66
int zmq::pgm_receiver_t::init (bool udp_encapsulation_, const char *network_)
67
{
68
    return pgm_socket.init (udp_encapsulation_, network_);
69 70
}

71 72
void zmq::pgm_receiver_t::plug (io_thread_t *io_thread_,
    session_base_t *session_)
73
{
Martin Sustrik's avatar
Martin Sustrik committed
74
    //  Retrieve PGM fds and start polling.
75 76
    fd_t socket_fd = retired_fd;
    fd_t waiting_pipe_fd = retired_fd;
77 78 79 80 81 82
    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);

83
    session = session_;
84 85 86

    //  If there are any subscriptions already queued in the session, drop them.
    drop_subscriptions ();
87 88 89 90
}

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

100 101 102 103 104
    if (has_rx_timer) {
        cancel_timer (rx_timer_id);
        has_rx_timer = false;
    }

105 106
    rm_fd (socket_handle);
    rm_fd (pipe_handle);
Martin Sustrik's avatar
Martin Sustrik committed
107

108
    session = NULL;
109 110
}

111 112 113 114 115 116
void zmq::pgm_receiver_t::terminate ()
{
    unplug ();
    delete this;
}

117
void zmq::pgm_receiver_t::restart_output ()
118
{
119
    drop_subscriptions ();
120 121
}

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

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

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

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

164 165 166 167 168
    if (has_rx_timer) {
        cancel_timer (rx_timer_id);
        has_rx_timer = false;
    }

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

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

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

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

210 211
        insize = static_cast <size_t> (received);

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

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

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

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

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

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

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

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

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

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

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


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

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

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

304 305
#endif