pgm_receiver.cpp 8.74 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
void zmq::pgm_receiver_t::plug (io_thread_t *io_thread_,
68
                                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 90
    for (peers_t::iterator it = peers.begin (), end = peers.end (); it != end;
         ++it) {
91
        if (it->second.decoder != NULL) {
92
            LIBZMQ_DELETE (it->second.decoder);
93
        }
malosek's avatar
malosek committed
94 95
    }
    peers.clear ();
96
    active_tsi = NULL;
97

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

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

106
    session = NULL;
107 108
}

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

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

120
bool zmq::pgm_receiver_t::restart_input ()
Martin Hurton's avatar
Martin Hurton committed
121
{
122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138
    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 ();
139
                return true;
140 141 142 143
            }
            //  Data error. Delete message decoder, mark the
            //  peer as not joined and drop remaining data.
            it->second.joined = false;
144
            LIBZMQ_DELETE (it->second.decoder);
145
            insize = 0;
146
        }
147 148 149 150 151 152
    }

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

153
    active_tsi = NULL;
154
    in_event ();
155 156

    return true;
Martin Hurton's avatar
Martin Hurton committed
157
}
158

159
const zmq::endpoint_uri_pair_t &zmq::pgm_receiver_t::get_endpoint () const
160
{
161
    return _empty_endpoint;
162
}
Martin Hurton's avatar
Martin Hurton committed
163

164 165
void zmq::pgm_receiver_t::in_event ()
{
166 167 168
    // If active_tsi is not null, there is a pending restart_input.
    // Keep the internal state as is so that restart_input would process the right data
    if (active_tsi) {
169
        return;
170
    }
171

Martin Sustrik's avatar
Martin Sustrik committed
172
    // Read data from the underlying pgm_socket.
malosek's avatar
malosek committed
173
    const pgm_tsi_t *tsi = NULL;
174

175 176 177 178 179
    if (has_rx_timer) {
        cancel_timer (rx_timer_id);
        has_rx_timer = false;
    }

180 181 182 183
    //  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.
184
        //  Note the workaround made not to break strict-aliasing rules.
185
        insize = 0;
186 187
        void *tmp = NULL;
        ssize_t received = pgm_socket.receive (&tmp, &tsi);
188 189 190

        //  No data to process. This may happen if the packet received is
        //  neither ODATA nor ODATA.
191
        if (received == 0) {
192
            if (errno == ENOMEM || errno == EBUSY) {
193 194 195 196
                const long timeout = pgm_socket.get_rx_timeout ();
                add_timer (timeout, rx_timer_id);
                has_rx_timer = true;
            }
197
            break;
198
        }
199 200 201 202 203 204

        //  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) {
205 206 207
            if (it != peers.end ()) {
                it->second.joined = false;
                if (it->second.decoder != NULL) {
208
                    LIBZMQ_DELETE (it->second.decoder);
209
                }
210 211
            }
            break;
malosek's avatar
malosek committed
212
        }
213

214 215 216
        //  New peer. Add it to the list of know but unjoint peers.
        if (it == peers.end ()) {
            peer_info_t peer_info = {false, NULL};
217
            it = peers.ZMQ_MAP_INSERT_OR_EMPLACE (*tsi, peer_info).first;
218
        }
219

220
        insize = static_cast<size_t> (received);
221
        inpos = (unsigned char *) tmp;
222

223
        //  Read the offset of the fist message in the current packet.
224 225 226 227
        zmq_assert (insize >= sizeof (uint16_t));
        uint16_t offset = get_uint16 (inpos);
        inpos += sizeof (uint16_t);
        insize -= sizeof (uint16_t);
228

229 230 231 232 233 234
        //  Join the stream if needed.
        if (!it->second.joined) {
            //  There is no beginning of the message in current packet.
            //  Ignore the data.
            if (offset == 0xffff)
                continue;
235

236
            zmq_assert (offset <= insize);
237
            zmq_assert (it->second.decoder == NULL);
238

239
            //  We have to move data to the beginning of the first message.
240 241
            inpos += offset;
            insize -= offset;
malosek's avatar
malosek committed
242

243 244
            //  Mark the stream as joined.
            it->second.joined = true;
malosek's avatar
malosek committed
245

246
            //  Create and connect decoder for the peer.
247 248
            it->second.decoder =
              new (std::nothrow) v1_decoder_t (0, options.maxmsgsize);
249
            alloc_assert (it->second.decoder);
250
        }
malosek's avatar
malosek committed
251

252 253 254 255 256 257 258 259 260 261
        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;
262 263
            }

264
            it->second.joined = false;
265
            LIBZMQ_DELETE (it->second.decoder);
266
            insize = 0;
267
        }
Martin Sustrik's avatar
Martin Sustrik committed
268
    }
269 270

    //  Flush any messages decoder may have produced.
271
    session->flush ();
272
}
malosek's avatar
malosek committed
273

274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296
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;
}


297 298 299 300
void zmq::pgm_receiver_t::timer_event (int token)
{
    zmq_assert (token == rx_timer_id);

301 302
    //  Timer cancels on return by poller_base.
    has_rx_timer = false;
303 304 305
    in_event ();
}

306 307 308
void zmq::pgm_receiver_t::drop_subscriptions ()
{
    msg_t msg;
309
    msg.init ();
310
    while (session->pull_msg (&msg) == 0)
311 312 313
        msg.close ();
}

314
#endif