zeromq_rcvbuf.cc

//
// 25-aug-2012 -- Rob Iverson
// Attempt to reproduce my "ZMQ_RCVBUF" issue for the purposes
// of reporting it to zeromq mailing list or bug tracker.
// To run this:
// zeromq_rcvbuf -send
// zeromq_rcvbuf -recv  (run this before the sender is done)
//
#include <zmq.hpp>
#include <string>
#include <iostream>

// 350,000 shows the problem for me; 250,000 does not
#define MESSAGE_SIZE 350000

// a sleep is required for me to see the problem
#define SEND_SLEEP_TIME 1

static int zmq_sender();
static int zmq_receiver();

int main(int argc, char **argv)
{
  if ( argc < 2 ) {
    std::cerr << "Usage: " << argv[0] << " [ -send | -recv ]" << std::endl;
    exit(1);
  }

  if ( !strcmp(argv[1],"-send") ) {
    return zmq_sender();
  }
  else if ( !strcmp(argv[1],"-recv") ) {
    return zmq_receiver();
  }
  exit(1);
}

static int zmq_sender()
{
  zmq::context_t context(1);
  zmq::socket_t socket(context, ZMQ_PUB);

  socket.connect("tcp://localhost:5555");

  for (int i = 0 ; i < 10 ; i++)
  {
    zmq::message_t message(MESSAGE_SIZE);

    *(static_cast<int*>(message.data())) = i;

    sleep(SEND_SLEEP_TIME);

    std::cout << "Sending message #" << i << std::endl;
    socket.send(message);
  }

  // Sleeping for 5s here makes the problem go away
  //sleep(5);

  socket.close();

  return 0;
}

static int zmq_receiver()
{
  zmq::context_t context(1);
  zmq::socket_t socket(context, ZMQ_SUB);

  socket.bind("tcp://*:5555");

  // subscribe for all messages
  socket.setsockopt(ZMQ_SUBSCRIBE,NULL,0);

  // Adjusting the RCVBUF size makes more messages go through
  //   ::uint64_t new_rcvbuf = 200000;
  //   socket.setsockopt(ZMQ_RCVBUF, &new_rcvbuf, sizeof(new_rcvbuf));

  for (int i = 0 ; i < 10 ; i++)
  {
    zmq::message_t request;

    std::cout << "+ Blocking waiting for message" << std::endl;
    socket.recv(&request);

    int sent_index = *(static_cast<int*>(request.data()));

    std::cout << "Received Message #" << sent_index\
              << " of size " << request.size() << std::endl;
  }

  return 0;
}