pgm_receiver.cpp 8.45 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
    for (peers_t::iterator it = peers.begin (); it != peers.end (); ++it) {
90
        if (it->second.decoder != NULL) {
91
            LIBZMQ_DELETE (it->second.decoder);
92
        }
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
    //  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.
175 176 177
        //  Note the workaround made not to break strict-aliasing rules.
        void *tmp = NULL;
        ssize_t received = pgm_socket.receive (&tmp, &tsi);
178
        inpos = (unsigned char *) tmp;
179 180 181

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

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

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

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

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

219 220 221 222 223 224
        //  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;
225

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

229
            //  We have to move data to the beginning 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
#endif