3party/libzmq
0
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-01-01 10:50:28 +08:00
Code Issues Projects Releases Wiki Activity

problem with PGM messages larger than 1 MTU fixed

Browse Source
This commit is contained in:
Martin Sustrik 2010-02-02 15:11:25 +01:00
parent 27e47bdcc0
commit 9cee8f9c3e
1 changed files with 52 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

100
src/pgm_receiver.cpp
View File

@ -93,72 +93,76 @@ void zmq::pgm_receiver_t::in_event ()
// Read data from the underlying pgm_socket. // Read data from the underlying pgm_socket.
unsigned char *data = NULL; unsigned char *data = NULL;
const pgm_tsi_t *tsi = NULL; const pgm_tsi_t *tsi = NULL;
ssize_t received = pgm_socket.receive ((void**) &data, &tsi);
// No data to process. This may happen if the packet received is // TODO: This loop can effectively block other engines in the same I/O
// neither ODATA nor ODATA. // thread in the case of high load.
if (received == 0) while (true) {
return;
// Find the peer based on its TSI. // Get new batch of data.
peers_t::iterator it = peers.find (*tsi); ssize_t received = pgm_socket.receive ((void**) &data, &tsi);
// Data loss. Delete decoder and mark the peer as disjoint. // No data to process. This may happen if the packet received is
if (received == -1) { // neither ODATA nor ODATA.
zmq_assert (it != peers.end ()); if (received == 0)
it->second.joined = false; break;
if (it->second.decoder != NULL) {
delete it->second.decoder; // Find the peer based on its TSI.
it->second.decoder = NULL; peers_t::iterator it = peers.find (*tsi);
// Data loss. Delete decoder and mark the peer as disjoint.
if (received == -1) {
zmq_assert (it != peers.end ());
it->second.joined = false;
if (it->second.decoder != NULL) {
delete it->second.decoder;
it->second.decoder = NULL;
}
break;
} }
return;
}
// New peer. Add it to the list of know but unjoint peers. // New peer. Add it to the list of know but unjoint peers.
if (it == peers.end ()) { if (it == peers.end ()) {
peer_info_t peer_info = {false, NULL}; peer_info_t peer_info = {false, NULL};
it = peers.insert (std::make_pair (*tsi, peer_info)).first; it = peers.insert (std::make_pair (*tsi, peer_info)).first;
} }
// Read the offset of the fist message in the current packet. // Read the offset of the fist message in the current packet.
zmq_assert ((size_t) received >= sizeof (uint16_t)); zmq_assert ((size_t) received >= sizeof (uint16_t));
uint16_t offset = get_uint16 (data); uint16_t offset = get_uint16 (data);
data += sizeof (uint16_t); data += sizeof (uint16_t);
received -= sizeof (uint16_t); received -= sizeof (uint16_t);
// Join the stream if needed. // Join the stream if needed.
if (!it->second.joined) { if (!it->second.joined) {
// There is no beginning of the message in current packet. // There is no beginning of the message in current packet.
// Ignore the data. // Ignore the data.
if (offset == 0xffff) if (offset == 0xffff)
return; continue;
zmq_assert (offset <= received); zmq_assert (offset <= received);
zmq_assert (it->second.decoder == NULL); zmq_assert (it->second.decoder == NULL);
// We have to move data to the begining of the first message. // We have to move data to the begining of the first message.
data += offset; data += offset;
received -= offset; received -= offset;
// Mark the stream as joined. // Mark the stream as joined.
it->second.joined = true; it->second.joined = true;
// Create and connect decoder for the peer. // Create and connect decoder for the peer.
it->second.decoder = new (std::nothrow) zmq_decoder_t (0, NULL, 0); it->second.decoder = new (std::nothrow) zmq_decoder_t (0, NULL, 0);
it->second.decoder->set_inout (inout); it->second.decoder->set_inout (inout);
} }
if (received) {
// Push all the data to the decoder. // Push all the data to the decoder.
// TODO: process_buffer may not process entire buffer! // TODO: process_buffer may not process entire buffer!
ssize_t processed = it->second.decoder->process_buffer (data, received); ssize_t processed = it->second.decoder->process_buffer (data, received);
zmq_assert (processed == received); zmq_assert (processed == received);
// Flush any messages decoder may have produced.
inout->flush ();
} }
// Flush any messages decoder may have produced.
inout->flush ();
} }
#endif #endif