Simple ZMQ Transmitter Illustrating Memory Usage

heavySender.cpp

/**
 * On my CentOS 5 system this application sent the first 3000 messages
 * with a running memory footprint less than 7 MB. In a few minutes, it
 * leveled out at 18MB. Based on the sending rate, the maximum data rate
 * out of this process should be less than 75kbps -- well below the
 * ZMQ_RATE of 10Mbps. While there doesn't seem to be a long-term leak,
 * the fact that this application does the first 3000 sends with a memory
 * footprint far less than it's terminal value is concerning.
 *
 * If I change nothing other than the ZMQ_RATE value and change it to
 * 200000 (200Mbps), the terminal memory usage of the process is now
 * more than 280MB. Clearly, the ZMQ_RATE value has something to do with
 * this memory usage -- even though the actual rate itself is invariant.
 *
 * As a final data point, setting ZMQ_RATE to 50000 (50Mbps), the app
 * again starts at less than 7MB for the first few thousand, but then
 * climbs to a terminal value of 73MB.
 *
 * My concern is the effect that ZMQ_RATE has on the memory usage. In
 * theory, it shouldn't have any effect, but clearly it does, and in a
 * very non-linear way.
 */
// System Headers
#include <iostream>
#include <stdio.h>
#include <string>
#include <stdint.h>
#include <sys/time.h>

// Third-Party Headers
#include <zmq.hpp>

// Other Headers

// Forward Declarations

// Public Constants

// Public DataTypes

// Public Data Constants



/**
 * This is the main test frame -- open up a ZMQ socket to one URL and
 * send a simple message every second. It's meant to be a measurement
 * test frame.
 */
int main(int argc, char *argv[]) {
	bool		error = false;

	// this is the URL to publish on
	std::string	url("epgm://eth0;225.1.1.1:55555");

	// make the ZMQ Context and Socket for what we need to transmit on
	zmq::context_t	*mContext = NULL;
	if (!error) {
		mContext = new zmq::context_t(1);
		if (mContext == NULL) {
			error = true;
			std::cout << "could not create the ZMQ context" << std::endl;
		}
	}
	zmq::socket_t	*mSocket = NULL;
	if (!error) {
		mSocket = new zmq::socket_t(*mContext, ZMQ_PUB);
		if (mSocket == NULL) {
			error = true;
			std::cout << "could not create the ZMQ socket" << std::endl;
		} else {
			// we need to set the maximum rate on this guy to 10Mbps
			int64_t		rate = 10000;
			mSocket->setsockopt(ZMQ_RATE, &rate, sizeof(rate));
			// connect to the right multicast group (with the URL)
			mSocket->connect(url.c_str());
		}
	}
	// verify it's all initialized OK
	if (!error) {
		std::cout << "Initialization complete." << std::endl;
	}

	// now let's listen for a while and just record the stats
	if (!error) {
		// let's make a "payload" that we'll put into all the messages
		std::string		payload("ABCEDGHIJKLMNOPQRSTUVWXYZ0123456789abcedfghijklmnopqrstuvwxyz");
		// now let's get into the sending loop
		uint16_t	cnt = 0;
		while (true) {
			// create a message, fill it, and send it out the socket
			zmq::message_t	msg(payload.size());
			memcpy(msg.data(), payload.data(), payload.size());
			mSocket->send(msg);
			if (++cnt >= 1000) {
				std::cout << "sent another " << cnt << "..." << std::endl;
				cnt = 0;
			}
			// wait a bit and do it again (1ms)
			usleep(1000);
		}
	}

	// clean it all up
	if (mSocket != NULL) {
		delete mSocket;
		mSocket = NULL;
	}
	if (mContext != NULL) {
		delete mContext;
		mContext = NULL;
	}

	return 0;
}