#include "precompiled.hpp"
#include "platform.hpp"
#if defined ZMQ_HAVE_NORM
#include "norm_engine.hpp"
#include "session_base.hpp"
#include "v2_protocol.hpp"
zmq::norm_engine_t::norm_engine_t(io_thread_t* parent_,
const options_t& options_)
: io_object_t(parent_), zmq_session(NULL), options(options_),
norm_instance(NORM_INSTANCE_INVALID), norm_session(NORM_SESSION_INVALID),
is_sender(false), is_receiver(false),
zmq_encoder(0), norm_tx_stream(NORM_OBJECT_INVALID),
tx_first_msg(true), tx_more_bit(false),
zmq_output_ready(false), norm_tx_ready(false),
tx_index(0), tx_len(0),
zmq_input_ready(false)
{
int rc = tx_msg.init();
errno_assert(0 == rc);
}
zmq::norm_engine_t::~norm_engine_t()
{
shutdown(); // in case it was not already called
}
int zmq::norm_engine_t::init(const char* network_, bool send, bool recv)
{
// Parse the "network_" address int "iface", "addr", and "port"
// norm endpoint format: [id,][<iface>;]<addr>:<port>
// First, look for optional local NormNodeId
// (default NORM_NODE_ANY causes NORM to use host IP addr for NormNodeId)
NormNodeId localId = NORM_NODE_ANY;
const char* ifacePtr = strchr(network_, ',');
if (NULL != ifacePtr)
{
size_t idLen = ifacePtr - network_;
if (idLen > 31) idLen = 31;
char idText[32];
strncpy(idText, network_, idLen);
idText[idLen] = '\0';
localId = (NormNodeId)atoi(idText);
ifacePtr++;
}
else
{
ifacePtr = network_;
}
// Second, look for optional multicast ifaceName
char ifaceName[256];
const char* addrPtr = strchr(ifacePtr, ';');
if (NULL != addrPtr)
{
size_t ifaceLen = addrPtr - ifacePtr;
if (ifaceLen > 255) ifaceLen = 255; // return error instead?
strncpy(ifaceName, ifacePtr, ifaceLen);
ifaceName[ifaceLen] = '\0';
ifacePtr = ifaceName;
addrPtr++;
}
else
{
addrPtr = ifacePtr;
ifacePtr = NULL;
}
// Finally, parse IP address and port number
const char* portPtr = strrchr(addrPtr, ':');
if (NULL == portPtr)
{
errno = EINVAL;
return -1;
}
char addr[256];
size_t addrLen = portPtr - addrPtr;
if (addrLen > 255) addrLen = 255;
strncpy(addr, addrPtr, addrLen);
addr[addrLen] = '\0';
portPtr++;
unsigned short portNumber = atoi(portPtr);
if (NORM_INSTANCE_INVALID == norm_instance)
{
if (NORM_INSTANCE_INVALID == (norm_instance = NormCreateInstance()))
{
// errno set by whatever caused NormCreateInstance() to fail
return -1;
}
}
// TBD - What do we use for our local NormNodeId?
// (for now we use automatic, IP addr based assignment or passed in 'id')
// a) Use ZMQ Identity somehow?
// b) Add function to use iface addr
// c) Randomize and implement a NORM session layer
// conflict detection/resolution protocol
norm_session = NormCreateSession(norm_instance, addr, portNumber, localId);
if (NORM_SESSION_INVALID == norm_session)
{
int savedErrno = errno;
NormDestroyInstance(norm_instance);
norm_instance = NORM_INSTANCE_INVALID;
errno = savedErrno;
return -1;
}
// There's many other useful NORM options that could be applied here
if (NormIsUnicastAddress(addr))
{
NormSetDefaultUnicastNack(norm_session, true);
}
else
{
// These only apply for multicast sessions
//NormSetTTL(norm_session, options.multicast_hops); // ZMQ default is 1
NormSetTTL(norm_session, 255); // since the ZMQ_MULTICAST_HOPS socket option isn't well-supported
NormSetRxPortReuse(norm_session, true); // port reuse doesn't work for non-connected unicast
NormSetLoopback(norm_session, true); // needed when multicast users on same machine
if (NULL != ifacePtr)
{
// Note a bad interface may not be caught until sender or receiver start
// (Since sender/receiver is not yet started, this always succeeds here)
NormSetMulticastInterface(norm_session, ifacePtr);
}
}
if (recv)
{
// The alternative NORM_SYNC_CURRENT here would provide "instant"
// receiver sync to the sender's _current_ message transmission.
// NORM_SYNC_STREAM tries to get everything the sender has cached/buffered
NormSetDefaultSyncPolicy(norm_session, NORM_SYNC_STREAM);
if (!NormStartReceiver(norm_session, 2*1024*1024))
{
// errno set by whatever failed
int savedErrno = errno;
NormDestroyInstance(norm_instance); // session gets closed, too
norm_session = NORM_SESSION_INVALID;
norm_instance = NORM_INSTANCE_INVALID;
errno = savedErrno;
return -1;
}
is_receiver = true;
}
if (send)
{
// Pick a random sender instance id (aka norm sender session id)
NormSessionId instanceId = NormGetRandomSessionId();
// TBD - provide "options" for some NORM sender parameters
if (!NormStartSender(norm_session, instanceId, 2*1024*1024, 1400, 16, 4))
{
// errno set by whatever failed
int savedErrno = errno;
NormDestroyInstance(norm_instance); // session gets closed, too
norm_session = NORM_SESSION_INVALID;
norm_instance = NORM_INSTANCE_INVALID;
errno = savedErrno;
return -1;
}
NormSetCongestionControl(norm_session, true);
norm_tx_ready = true;
is_sender = true;
if (NORM_OBJECT_INVALID == (norm_tx_stream = NormStreamOpen(norm_session, 2*1024*1024)))
{
// errno set by whatever failed
int savedErrno = errno;
NormDestroyInstance(norm_instance); // session gets closed, too
norm_session = NORM_SESSION_INVALID;
norm_instance = NORM_INSTANCE_INVALID;
errno = savedErrno;
return -1;
}
}
//NormSetMessageTrace(norm_session, true);
//NormSetDebugLevel(3);
//NormOpenDebugLog(norm_instance, "normLog.txt");
return 0; // no error
} // end zmq::norm_engine_t::init()
void zmq::norm_engine_t::shutdown()
{
// TBD - implement a more graceful shutdown option
if (is_receiver)
{
NormStopReceiver(norm_session);
// delete any active NormRxStreamState
rx_pending_list.Destroy();
rx_ready_list.Destroy();
msg_ready_list.Destroy();
is_receiver = false;
}
if (is_sender)
{
NormStopSender(norm_session);
is_sender = false;
}
if (NORM_SESSION_INVALID != norm_session)
{
NormDestroySession(norm_session);
norm_session = NORM_SESSION_INVALID;
}
if (NORM_INSTANCE_INVALID != norm_instance)
{
NormStopInstance(norm_instance);
NormDestroyInstance(norm_instance);
norm_instance = NORM_INSTANCE_INVALID;
}
} // end zmq::norm_engine_t::shutdown()
void zmq::norm_engine_t::plug (io_thread_t* io_thread_, session_base_t *session_)
{
// TBD - we may assign the NORM engine to an io_thread in the future???
zmq_session = session_;
if (is_sender) zmq_output_ready = true;
if (is_receiver) zmq_input_ready = true;
fd_t normDescriptor = NormGetDescriptor(norm_instance);
norm_descriptor_handle = add_fd(normDescriptor);
// Set POLLIN for notification of pending NormEvents
set_pollin(norm_descriptor_handle);
if (is_sender) send_data();
} // end zmq::norm_engine_t::init()
void zmq::norm_engine_t::unplug()
{
rm_fd(norm_descriptor_handle);
zmq_session = NULL;
} // end zmq::norm_engine_t::unplug()
void zmq::norm_engine_t::terminate()
{
unplug();
shutdown();
delete this;
}
void zmq::norm_engine_t::restart_output()
{
// There's new message data available from the session
zmq_output_ready = true;
if (norm_tx_ready) send_data();
} // end zmq::norm_engine_t::restart_output()
void zmq::norm_engine_t::send_data()
{
// Here we write as much as is available or we can
while (zmq_output_ready && norm_tx_ready)
{
if (0 == tx_len)
{
// Our tx_buffer needs data to send
// Get more data from encoder
size_t space = BUFFER_SIZE;
unsigned char* bufPtr = (unsigned char*)tx_buffer;
tx_len = zmq_encoder.encode(&bufPtr, space);
if (0 == tx_len)
{
if (tx_first_msg)
{
// We don't need to mark eom/flush until a message is sent
tx_first_msg = false;
}
else
{
// A prior message was completely written to stream, so
// mark end-of-message and possibly flush (to force packet transmission,
// even if it's not a full segment so message gets delivered quickly)
// NormStreamMarkEom(norm_tx_stream); // the flush below marks eom
// Note NORM_FLUSH_ACTIVE makes NORM fairly chatty for low duty cycle messaging
// but makes sure content is delivered quickly. Positive acknowledgements
// with flush override would make NORM more succinct here
NormStreamFlush(norm_tx_stream, true, NORM_FLUSH_ACTIVE);
}
// Need to pull and load a new message to send
if (-1 == zmq_session->pull_msg(&tx_msg))
{
// We need to wait for "restart_output()" to be called by ZMQ
zmq_output_ready = false;
break;
}
zmq_encoder.load_msg(&tx_msg);
// Should we write message size header for NORM to use? Or expect NORM
// receiver to decode ZMQ message framing format(s)?
// OK - we need to use a byte to denote when the ZMQ frame is the _first_
// frame of a message so it can be decoded properly when a receiver
// 'syncs' mid-stream. We key off the the state of the 'more_flag'
// I.e.,If more_flag _was_ false previously, this is the first
// frame of a ZMQ message.
if (tx_more_bit)
tx_buffer[0] = (char)0xff; // this is not first frame of message
else
tx_buffer[0] = 0x00; // this is first frame of message
tx_more_bit = (0 != (tx_msg.flags() & msg_t::more));
// Go ahead an get a first chunk of the message
bufPtr++;
space--;
tx_len = 1 + zmq_encoder.encode(&bufPtr, space);
tx_index = 0;
}
}
// Do we have data in our tx_buffer pending
if (tx_index < tx_len)
{
// We have data in our tx_buffer to send, so write it to the stream
tx_index += NormStreamWrite(norm_tx_stream, tx_buffer + tx_index, tx_len - tx_index);
if (tx_index < tx_len)
{
// NORM stream buffer full, wait for NORM_TX_QUEUE_VACANCY
norm_tx_ready = false;
break;
}
tx_len = 0; // all buffered data was written
}
} // end while (zmq_output_ready && norm_tx_ready)
} // end zmq::norm_engine_t::send_data()
void zmq::norm_engine_t::in_event()
{
// This means a NormEvent is pending, so call NormGetNextEvent() and handle
NormEvent event;
if (!NormGetNextEvent(norm_instance, &event))
{
// NORM has died before we unplugged?!
zmq_assert(false);
return;
}
switch(event.type)
{
case NORM_TX_QUEUE_VACANCY:
case NORM_TX_QUEUE_EMPTY:
if (!norm_tx_ready)
{
norm_tx_ready = true;
send_data();
}
break;
case NORM_RX_OBJECT_NEW:
//break;
case NORM_RX_OBJECT_UPDATED:
recv_data(event.object);
break;
case NORM_RX_OBJECT_ABORTED:
{
NormRxStreamState* rxState = (NormRxStreamState*)NormObjectGetUserData(event.object);
if (NULL != rxState)
{
// Remove the state from the list it's in
// This is now unnecessary since deletion takes care of list removal
// but in the interest of being clear ...
NormRxStreamState::List* list = rxState->AccessList();
if (NULL != list) list->Remove(*rxState);
}
delete rxState;
break;
}
case NORM_REMOTE_SENDER_INACTIVE:
// Here we free resources used for this formerly active sender.
// Note w/ NORM_SYNC_STREAM, if sender reactivates, we may
// get some messages delivered twice. NORM_SYNC_CURRENT would
// mitigate that but might miss data at startup. Always tradeoffs.
// Instead of immediately deleting, we could instead initiate a
// user configurable timeout here to wait some amount of time
// after this event to declare the remote sender truly dead
// and delete its state???
NormNodeDelete(event.sender);
break;
default:
// We ignore some NORM events
break;
}
} // zmq::norm_engine_t::in_event()
void zmq::norm_engine_t::restart_input()
{
// TBD - should we check/assert that zmq_input_ready was false???
zmq_input_ready = true;
// Process any pending received messages
if (!msg_ready_list.IsEmpty())
recv_data(NORM_OBJECT_INVALID);
} // end zmq::norm_engine_t::restart_input()
void zmq::norm_engine_t::recv_data(NormObjectHandle object)
{
if (NORM_OBJECT_INVALID != object)
{
// Call result of NORM_RX_OBJECT_UPDATED notification
// This is a rx_ready indication for a new or existing rx stream
// First, determine if this is a stream we already know
zmq_assert(NORM_OBJECT_STREAM == NormObjectGetType(object));
// Since there can be multiple senders (publishers), we keep
// state for each separate rx stream.
NormRxStreamState* rxState = (NormRxStreamState*)NormObjectGetUserData(object);
if (NULL == rxState)
{
// This is a new stream, so create rxState with zmq decoder, etc
rxState = new (std::nothrow) NormRxStreamState(object, options.maxmsgsize);
errno_assert(rxState);
if (!rxState->Init())
{
errno_assert(false);
delete rxState;
return;
}
NormObjectSetUserData(object, rxState);
}
else if (!rxState->IsRxReady())
{
// Existing non-ready stream, so remove from pending
// list to be promoted to rx_ready_list ...
rx_pending_list.Remove(*rxState);
}
if (!rxState->IsRxReady())
{
// TBD - prepend up front for immediate service?
rxState->SetRxReady(true);
rx_ready_list.Append(*rxState);
}
}
// This loop repeats until we've read all data available from "rx ready" inbound streams
// and pushed any accumulated messages we can up to the zmq session.
while (!rx_ready_list.IsEmpty() || (zmq_input_ready && !msg_ready_list.IsEmpty()))
{
// Iterate through our rx_ready streams, reading data into the decoder
// (This services incoming "rx ready" streams in a round-robin fashion)
NormRxStreamState::List::Iterator iterator(rx_ready_list);
NormRxStreamState* rxState;
while (NULL != (rxState = iterator.GetNextItem()))
{
switch(rxState->Decode())
{
case 1: // msg completed
// Complete message decoded, move this stream to msg_ready_list
// to push the message up to the session below. Note the stream
// will be returned to the "rx_ready_list" after that's done
rx_ready_list.Remove(*rxState);
msg_ready_list.Append(*rxState);
continue;
case -1: // decoding error (shouldn't happen w/ NORM, but ...)
// We need to re-sync this stream (decoder buffer was reset)
rxState->SetSync(false);
break;
default: // 0 - need more data
break;
}
// Get more data from this stream
NormObjectHandle stream = rxState->GetStreamHandle();
// First, make sure we're in sync ...
while (!rxState->InSync())
{
// seek NORM message start
if (!NormStreamSeekMsgStart(stream))
{
// Need to wait for more data
break;
}
// read message 'flag' byte to see if this it's a 'final' frame
char syncFlag;
unsigned int numBytes = 1;
if (!NormStreamRead(stream, &syncFlag, &numBytes))
{
// broken stream (shouldn't happen after seek msg start?)
zmq_assert(false);
continue;
}
if (0 == numBytes)
{
// This probably shouldn't happen either since we found msg start
// Need to wait for more data
break;
}
if (0 == syncFlag) rxState->SetSync(true);
// else keep seeking ...
} // end while(!rxState->InSync())
if (!rxState->InSync())
{
// Need more data for this stream, so remove from "rx ready"
// list and iterate to next "rx ready" stream
rxState->SetRxReady(false);
// Move from rx_ready_list to rx_pending_list
rx_ready_list.Remove(*rxState);
rx_pending_list.Append(*rxState);
continue;
}
// Now we're actually ready to read data from the NORM stream to the zmq_decoder
// the underlying zmq_decoder->get_buffer() call sets how much is needed.
unsigned int numBytes = rxState->GetBytesNeeded();
if (!NormStreamRead(stream, rxState->AccessBuffer(), &numBytes))
{
// broken NORM stream, so re-sync
rxState->Init(); // TBD - check result
// This will retry syncing, and getting data from this stream
// since we don't increment the "it" iterator
continue;
}
rxState->IncrementBufferCount(numBytes);
if (0 == numBytes)
{
// All the data available has been read
// Need to wait for NORM_RX_OBJECT_UPDATED for this stream
rxState->SetRxReady(false);
// Move from rx_ready_list to rx_pending_list
rx_ready_list.Remove(*rxState);
rx_pending_list.Append(*rxState);
}
} // end while(NULL != (rxState = iterator.GetNextItem()))
if (zmq_input_ready)
{
// At this point, we've made a pass through the "rx_ready" stream list
// Now make a pass through the "msg_pending" list (if the zmq session
// ready for more input). This may possibly return streams back to
// the "rx ready" stream list after their pending message is handled
NormRxStreamState::List::Iterator iterator(msg_ready_list);
NormRxStreamState* rxState;
while (NULL != (rxState = iterator.GetNextItem()))
{
msg_t* msg = rxState->AccessMsg();
int rc = zmq_session->push_msg(msg);
if (-1 == rc)
{
if (EAGAIN == errno)
{
// need to wait until session calls "restart_input()"
zmq_input_ready = false;
break;
}
else
{
// session rejected message?
// TBD - handle this better
zmq_assert(false);
}
}
// else message was accepted.
msg_ready_list.Remove(*rxState);
if (rxState->IsRxReady()) // Move back to "rx_ready" list to read more data
rx_ready_list.Append(*rxState);
else // Move back to "rx_pending" list until NORM_RX_OBJECT_UPDATED
msg_ready_list.Append(*rxState);
} // end while(NULL != (rxState = iterator.GetNextItem()))
} // end if (zmq_input_ready)
} // end while ((!rx_ready_list.empty() || (zmq_input_ready && !msg_ready_list.empty()))
// Alert zmq of the messages we have pushed up
zmq_session->flush();
} // end zmq::norm_engine_t::recv_data()
zmq::norm_engine_t::NormRxStreamState::NormRxStreamState(NormObjectHandle normStream,
int64_t maxMsgSize)
: norm_stream(normStream), max_msg_size(maxMsgSize),
in_sync(false), rx_ready(false), zmq_decoder(NULL), skip_norm_sync(false),
buffer_ptr(NULL), buffer_size(0), buffer_count(0),
prev(NULL), next(NULL), list(NULL)
{
}
zmq::norm_engine_t::NormRxStreamState::~NormRxStreamState()
{
if (NULL != zmq_decoder)
{
delete zmq_decoder;
zmq_decoder = NULL;
}
if (NULL != list)
{
list->Remove(*this);
list = NULL;
}
}
bool zmq::norm_engine_t::NormRxStreamState::Init()
{
in_sync = false;
skip_norm_sync = false;
if (NULL != zmq_decoder) delete zmq_decoder;
// Note "in_batch_size" comes from config.h
zmq_decoder = new (std::nothrow) v2_decoder_t (in_batch_size, max_msg_size);
alloc_assert (zmq_decoder);
if (NULL != zmq_decoder)
{
buffer_count = 0;
buffer_size = 0;
zmq_decoder->get_buffer(&buffer_ptr, &buffer_size);
return true;
}
else
{
return false;
}
} // end zmq::norm_engine_t::NormRxStreamState::Init()
// This decodes any pending data sitting in our stream decoder buffer
// It returns 1 upon message completion, -1 on error, 1 on msg completion
int zmq::norm_engine_t::NormRxStreamState::Decode()
{
// If we have pending bytes to decode, process those first
while (buffer_count > 0)
{
// There's pending data for the decoder to decode
size_t processed = 0;
// This a bit of a kludgy approach used to weed
// out the NORM ZMQ message transport "syncFlag" byte
// from the ZMQ message stream being decoded (but it works!)
if (skip_norm_sync)
{
buffer_ptr++;
buffer_count--;
skip_norm_sync = false;
}
int rc = zmq_decoder->decode(buffer_ptr, buffer_count, processed);
buffer_ptr += processed;
buffer_count -= processed;
switch (rc)
{
case 1:
// msg completed
if (0 == buffer_count)
{
buffer_size = 0;
zmq_decoder->get_buffer(&buffer_ptr, &buffer_size);
}
skip_norm_sync = true;
return 1;
case -1:
// decoder error (reset decoder and state variables)
in_sync = false;
skip_norm_sync = false; // will get consumed by norm sync check
Init();
break;
case 0:
// need more data, keep decoding until buffer exhausted
break;
}
}
// Reset buffer pointer/count for next read
buffer_count = 0;
buffer_size = 0;
zmq_decoder->get_buffer(&buffer_ptr, &buffer_size);
return 0; // need more data
} // end zmq::norm_engine_t::NormRxStreamState::Decode()
zmq::norm_engine_t::NormRxStreamState::List::List()
: head(NULL), tail(NULL)
{
}
zmq::norm_engine_t::NormRxStreamState::List::~List()
{
Destroy();
}
void zmq::norm_engine_t::NormRxStreamState::List::Destroy()
{
NormRxStreamState* item = head;
while (NULL != item)
{
Remove(*item);
delete item;
item = head;
}
} // end zmq::norm_engine_t::NormRxStreamState::List::Destroy()
void zmq::norm_engine_t::NormRxStreamState::List::Append(NormRxStreamState& item)
{
item.prev = tail;
if (NULL != tail)
tail->next = &item;
else
head = &item;
item.next = NULL;
tail = &item;
item.list = this;
} // end zmq::norm_engine_t::NormRxStreamState::List::Append()
void zmq::norm_engine_t::NormRxStreamState::List::Remove(NormRxStreamState& item)
{
if (NULL != item.prev)
item.prev->next = item.next;
else
head = item.next;
if (NULL != item.next)
item.next ->prev = item.prev;
else
tail = item.prev;
item.prev = item.next = NULL;
item.list = NULL;
} // end zmq::norm_engine_t::NormRxStreamState::List::Remove()
zmq::norm_engine_t::NormRxStreamState::List::Iterator::Iterator(const List& list)
: next_item(list.head)
{
}
zmq::norm_engine_t::NormRxStreamState* zmq::norm_engine_t::NormRxStreamState::List::Iterator::GetNextItem()
{
NormRxStreamState* nextItem = next_item;
if (NULL != nextItem) next_item = nextItem->next;
return nextItem;
} // end zmq::norm_engine_t::NormRxStreamState::List::Iterator::GetNextItem()
const char *zmq::norm_engine_t::get_endpoint () const
{
return "";
}
#endif // ZMQ_HAVE_NORM