[zeromq-dev] ZMQ Message Size Limitations

[Pieter Hintjens]

Hi Mark,

There's no clear reason the ventilator would kill itself when sending
a larger message. Can you strip down the case to a minimal one, that
we can reproduce?

Thanks,
-Pieter

On Wed, Apr 9, 2014 at 10:26 PM, Mark Hotchkiss
<mark.hotchkiss at yahoo.com> wrote:
> 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
>
>
>
>
> _______________________________________________
> zeromq-dev mailing list
> zeromq-dev at lists.zeromq.org
> http://lists.zeromq.org/mailman/listinfo/zeromq-dev
>