[zeromq-dev] ZMQ Message Size Limitations

[Mark Hotchkiss]

Message Size Limitations in Divide And Conquer Architecture

OS: CentOS and Arch Linux, ZMQ Version: 4.0.3

We are relatively new users of ZMQ and have been performing some load testing with a modified version of the “Divide and Conquer” example. Using standard C, I have made modifications to the taskvent, taskwork, and tasksink classes so that I can adjust the number of jobs sent and confirmed by the Ventilator and Sink respectively, the workload of the Worker (through sleep), and most importantly for now, the Message size sent from the Ventilator.

I have experimented with these classes in two environments: 

1. On a single CentOS Linux host (128GB memory) where all three components run, but starting multiple Workers (8-10) processes to simulate a Worker cluster.

2. Using one CentOS Linux host (6GB memory) to run the Ventilator and Sink, and a real cluster of 33 Raspberry Pi Arch Linux hosted Workers in a local isolated TCP/IP network.

From preliminary testing, I am finding that regardless of the number of jobs and workload on the Workers, there seems to be an upper limit on the size of the Message that I can push from the Ventilator that ranges from between 1.63MB on the single host to about 2.3MB on the real cluster.

The real cluster of Raspberry Pis (Workers) with the CentOS host (Ventilator and Sink) with just 6GB of memory seems to use ~93% of available memory when pushing a 2.3MB message. Anything significantly over that and the Ventilator kills itself. If I keep the Message size at about 2.3MB it can complete 5000 jobs with a 60 second workload on each Worker in about 2.5 hours.

On the single host with 128GB of memory running just 5 jobs and 1 second of workload, if I adjust the Message size to much over 1.67MB, the jobs won’t complete near the end (just hangs on the Sink end) but not for memory limitations as on the real cluster. The Ventilator completes instantly. In fact I see virtually no memory usage by any of the processes when executing before hanging.

We are not seeing anything in the documentation that mentions any specified limitations on Message size and would like to know if there is some further configuration that we need to be paying attention to so that we can send messages somewhere on the order of 10-15MB.

I am including all of the modified code based on the ZeroMQ here. The “zhelpers.h” has not been modified from the ZeroMQ example from the book/website:

taskvent.c
***************************************
#include "zhelpers.h"

int main(int argc, char const *argv[]) {

if (argc < 3) {
printf( "You must provide the number of jobs to be generated 
and message size.\n");
exit(0);
}

int numberOfJobs = atoi(argv[1]);
printf("Generate %d Number of Jobs\n", numberOfJobs);

int messageSize = atoi(argv[2]);
printf("Set message size to %d bytes\n", messageSize);

void *context = zmq_ctx_new();

//  Socket to send messages on
void *sender = zmq_socket(context, ZMQ_PUSH);
zmq_bind(sender, "tcp://*:5557");

//  Socket to send start of batch message on
void *sink = zmq_socket(context, ZMQ_PUSH);
zmq_connect(sink, "tcp://localhost:5558");

printf("Press Enter when the workers are ready: ");
getchar();
printf("Sending tasks to workers...\n");

//  The first message is "0" and signals start of batch
s_send(sink, "0");

//  Initialize random number generator
srandom((unsigned) time(NULL));

//  Send numberOfJobs/tasks
int task_nbr;
int total_msec = 0;     //  Total expected cost in msecs
for (task_nbr = 0; task_nbr < numberOfJobs; task_nbr++) {

// Stuff a character array with a bunch of AAAAA
// characters of quantity based on messageSize
unsigned char *p = (unsigned char *) malloc(
messageSize * sizeof(unsigned char));
memset((void *) p, 'A', messageSize);
p[messageSize - 1] = '\0';

// Send the message and reclaim memory
s_send(sender, p);
free(p);
}

// This was for estimation but is useless for our purposes
//printf("Total expected cost: %d seconds\n", total_msec / 1000);

zmq_close(sink);
zmq_close(sender);
zmq_ctx_destroy(context);
return 0;
}
*************************

taskwork.c
*************************
#include "zhelpers.h"

int main(int argc, char const *argv[]) {

if (argc < 2) {
printf(
"You must provide the workload.\n");
exit(0);
}

int workLoad = atoi(argv[1]);

//  Socket to receive messages on
void *context = zmq_ctx_new();
void *receiver = zmq_socket(context, ZMQ_PULL);
//zmq_connect(receiver, "tcp://localhost:5557");
zmq_connect(receiver, "tcp://192.168.1.48:5557");

//  Socket to send messages to
void *sender = zmq_socket(context, ZMQ_PUSH);
//zmq_connect(sender, "tcp://localhost:5558");
zmq_connect(sender, "tcp://192.168.1.48:5558");

char hostname[1024];
hostname[1023] = '\0';
gethostname(hostname, 1023);

//  Process tasks forever
while (1) {
char *string = s_recv(receiver);

// Print the full string for debugging
//printf("%s.", string);     //  Show progress

fflush(stdout);

// Sleep for the workLoad argument time
s_sleep(workLoad);

// Clean up memory and send the reply to tasksink
free(string);
s_send(sender, hostname); //  Send results to sink
}
zmq_close(receiver);
zmq_close(sender);
zmq_ctx_destroy(context);
return 0;
}
**************************

tasksink.c
*************************
#include "zhelpers.h"

int main(int argc, char const *argv[]) {

if (argc < 2) {
printf("You must provide the number of confirmations.\n");
exit(0);
}

int numberOfConfirmations = atoi(argv[1]);
printf("Generate %d Number of Confirmation\n", numberOfConfirmations);

//  Prepare our context and socket
void *context = zmq_ctx_new();
void *receiver = zmq_socket(context, ZMQ_PULL);
zmq_bind(receiver, "tcp://*:5558");

//  Wait for start of batch
char *string = s_recv(receiver);
free(string);

//  Start our clock now
int64_t start_time = s_clock();

//  Process numberOfConfirmations confirmations
int task_nbr;
for (task_nbr = 0; task_nbr < numberOfConfirmations; task_nbr++) {
char *string = s_recv(receiver);
//free(string);

// Report the received message and task number
if ((task_nbr / 10) * 10 == task_nbr) {
printf("%s: %d\n", string, task_nbr);
} else {
printf(".");
}
free(string);
fflush(stdout);
}
//  Calculate and report duration of batch
//printf("Total elapsed time: %d msec\n", (int) (s_clock() - start_time));
printf("Total elapsed time: %d seconds\n", (int) (s_clock() - start_time)/1000);

zmq_close(receiver);
zmq_ctx_destroy(context);
return 0;
}
***********************************

Any help would be appreciated. Thanks.

Mark
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <https://lists.zeromq.org/pipermail/zeromq-dev/attachments/20140409/ab74b3a4/attachment.htm>