pgm_receiver.cpp 6.63 KB
Newer Older
1
/*
2
    Copyright (c) 2007-2010 iMatix Corporation
3 4 5 6

    This file is part of 0MQ.

    0MQ is free software; you can redistribute it and/or modify it under
7
    the terms of the GNU Lesser General Public License as published by
8 9 10 11 12 13
    the Free Software Foundation; either version 3 of the License, or
    (at your option) any later version.

    0MQ 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
14
    GNU Lesser General Public License for more details.
15

16
    You should have received a copy of the GNU Lesser General Public License
17 18 19 20 21
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "platform.hpp"

malosek's avatar
malosek committed
22
#if defined ZMQ_HAVE_OPENPGM
23

24 25
#include <new>

26 27 28 29
#ifdef ZMQ_HAVE_WINDOWS
#include "windows.hpp"
#endif

30 31 32 33 34 35 36
#include "pgm_receiver.hpp"
#include "err.hpp"
#include "stdint.hpp"
#include "wire.hpp"
#include "i_inout.hpp"

zmq::pgm_receiver_t::pgm_receiver_t (class io_thread_t *parent_, 
37
      const options_t &options_) :
38
    io_object_t (parent_),
39
    has_rx_timer (false),
40 41
    pgm_socket (true, options_),
    options (options_),
42 43 44
    inout (NULL),
    mru_decoder (NULL),
    pending_bytes (0)
45 46 47 48 49
{
}

zmq::pgm_receiver_t::~pgm_receiver_t ()
{
malosek's avatar
malosek committed
50 51
    //  Destructor should not be called before unplug.
    zmq_assert (peers.empty ());
52 53
}

54
int zmq::pgm_receiver_t::init (bool udp_encapsulation_, const char *network_)
55
{
56
    return pgm_socket.init (udp_encapsulation_, network_);
57 58
}

59
void zmq::pgm_receiver_t::plug (io_thread_t *io_thread_, i_inout *inout_)
60
{
Martin Sustrik's avatar
Martin Sustrik committed
61
    //  Retrieve PGM fds and start polling.
62 63 64 65 66 67 68 69 70 71 72 73 74
    int socket_fd;
    int waiting_pipe_fd;
    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);

    inout = inout_;
}

void zmq::pgm_receiver_t::unplug ()
{
malosek's avatar
malosek committed
75
    //  Delete decoders.
Martin Sustrik's avatar
Martin Sustrik committed
76
    for (peers_t::iterator it = peers.begin (); it != peers.end (); it++) {
malosek's avatar
malosek committed
77 78 79 80 81
        if (it->second.decoder != NULL)
            delete it->second.decoder;
    }
    peers.clear ();

82 83 84
    mru_decoder = NULL;
    pending_bytes = 0;

85 86 87 88 89
    if (has_rx_timer) {
        cancel_timer (rx_timer_id);
        has_rx_timer = false;
    }

90 91
    rm_fd (socket_handle);
    rm_fd (pipe_handle);
Martin Sustrik's avatar
Martin Sustrik committed
92

93 94 95
    inout = NULL;
}

96 97 98 99 100 101 102
void zmq::pgm_receiver_t::terminate ()
{
    unplug ();
    delete this;
}

void zmq::pgm_receiver_t::activate_out ()
103 104 105 106
{
    zmq_assert (false);
}

107
void zmq::pgm_receiver_t::activate_in ()
Martin Hurton's avatar
Martin Hurton committed
108
{
109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132
    //  It is possible that the most recently used decoder
    //  processed the whole buffer but failed to write
    //  the last message into the pipe.
    if (pending_bytes == 0) {
        if (mru_decoder != NULL)
            mru_decoder->process_buffer (NULL, 0);
        return;
    }

    zmq_assert (mru_decoder != NULL);
    zmq_assert (pending_ptr != NULL);

    //  Ask the decoder to process remaining data.
    size_t n = mru_decoder->process_buffer (pending_ptr, pending_bytes);
    pending_bytes -= n;

    if (pending_bytes > 0)
        return;

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

    in_event ();
Martin Hurton's avatar
Martin Hurton committed
133 134
}

135 136
void zmq::pgm_receiver_t::in_event ()
{
Martin Sustrik's avatar
Martin Sustrik committed
137 138
    // Read data from the underlying pgm_socket.
    unsigned char *data = NULL;
malosek's avatar
malosek committed
139
    const pgm_tsi_t *tsi = NULL;
140

141 142
    zmq_assert (pending_bytes == 0);

143 144 145 146 147
    if (has_rx_timer) {
        cancel_timer (rx_timer_id);
        has_rx_timer = false;
    }

148 149 150 151 152 153 154 155 156
    //  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.
        ssize_t received = pgm_socket.receive ((void**) &data, &tsi);

        //  No data to process. This may happen if the packet received is
        //  neither ODATA nor ODATA.
157
        if (received == 0) {
158
            if (errno == ENOMEM || errno == EBUSY) {
159 160 161 162
                const long timeout = pgm_socket.get_rx_timeout ();
                add_timer (timeout, rx_timer_id);
                has_rx_timer = true;
            }
163
            break;
164
        }
165 166 167 168 169 170

        //  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) {
171 172 173 174 175 176 177 178
            if (it != peers.end ()) {
                it->second.joined = false;
                if (it->second.decoder == mru_decoder)
                    mru_decoder = NULL;
                if (it->second.decoder != NULL) {
                    delete it->second.decoder;
                    it->second.decoder = NULL;
                }
179 180
            }
            break;
malosek's avatar
malosek committed
181
        }
182

183 184 185
        //  New peer. Add it to the list of know but unjoint peers.
        if (it == peers.end ()) {
            peer_info_t peer_info = {false, NULL};
186
            it = peers.insert (peers_t::value_type (*tsi, peer_info)).first;
187
        }
188

189 190 191 192 193
        //  Read the offset of the fist message in the current packet.
        zmq_assert ((size_t) received >= sizeof (uint16_t));
        uint16_t offset = get_uint16 (data);
        data += sizeof (uint16_t);
        received -= sizeof (uint16_t);
194

195 196
        //  Join the stream if needed.
        if (!it->second.joined) {
197

198 199 200 201
            //  There is no beginning of the message in current packet.
            //  Ignore the data.
            if (offset == 0xffff)
                continue;
202

203 204
            zmq_assert (offset <= received);
            zmq_assert (it->second.decoder == NULL);
205

206 207 208
            //  We have to move data to the begining of the first message.
            data += offset;
            received -= offset;
malosek's avatar
malosek committed
209

210 211
            //  Mark the stream as joined.
            it->second.joined = true;
malosek's avatar
malosek committed
212

213
            //  Create and connect decoder for the peer.
214
            it->second.decoder = new (std::nothrow) decoder_t (0);
215 216
            it->second.decoder->set_inout (inout);
        }
malosek's avatar
malosek committed
217

218 219
        mru_decoder = it->second.decoder;

Martin Sustrik's avatar
Martin Sustrik committed
220
        //  Push all the data to the decoder.
221
        ssize_t processed = it->second.decoder->process_buffer (data, received);
222 223 224 225 226 227 228 229 230 231
        if (processed < received) {
            //  Save some state so we can resume the decoding process later.
            pending_bytes = received - processed;
            pending_ptr = data + processed;
            //  Stop polling.
            reset_pollin (pipe_handle);
            reset_pollin (socket_handle);

            break;
        }
Martin Sustrik's avatar
Martin Sustrik committed
232
    }
233 234 235

    //  Flush any messages decoder may have produced.
    inout->flush ();
236
}
malosek's avatar
malosek committed
237

238 239 240 241
void zmq::pgm_receiver_t::timer_event (int token)
{
    zmq_assert (token == rx_timer_id);

242 243
    //  Timer cancels on return by poller_base.
    has_rx_timer = false;
244 245 246
    in_event ();
}

247 248
#endif