[zeromq-dev] Resource puzzle with the Divide and Conquer (Ventilator->Workers->Sink) Example

[Mark Hotchkiss]

I have written before about taking the example starting on page 16 of Pieter’s book, adding arguments to each of the three components so that by adjusting the number of Jobs, Worker workload (sleep seconds), MessageSize, and number of processes (either unique host, or tmutliple processes on a single host). 

Our application requires performance tuning to take the best advantage to the hardware and software as we can. It is a very expensive process and we are attempting to use ZMQ as a primary processing technology. But understanding its performance has been a puzzle and I would appreciate any help on metrics to adjust to get the best results.

I have run many tests and have previously reported that I have found an upper limit on the MessageSize that I can send from the Ventilator to the Workers in an environment with CentOS running the Ventilator and the Sink, and hosting the Workers on a 33 node Beowulf Cluster of Raspberry Pi’s running ARCH Linux.

I have since run many test on a single host with multi-core high end resources loaded up with memory. I have run the example with the following parameters.

Jobs pushed from the Ventilator and received by the Sink: 20000     
Worker Workload: 2 sec (simulated by sleep)    
MessageSize from Ventilator to each Worker: 50K bytes (relatively small from my experience)  
Number of Worker processes: 1000
Completion Time: 40 seconds            

If I push the number of Worker processes to just 1025, the Sink process freezes at 19770 jobs and eats up all of the memory available.

I present the code below that is changed and abbreviated to a minimum from the original in the book for the purposes of adjusting resources and expense. I use an unmodified zhelpers.h as well. 

I hope that understanding performance in test environments would be important to the forum. I have done my best to simplfy the example and steps to run it.

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();
void *sender = zmq_socket(context, ZMQ_PUSH);
zmq_bind(sender, "tcp://*:5557");
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");
s_send(sink, "0");
srandom((unsigned) time(NULL));
int task_nbr;
int total_msec = 0;
for (task_nbr = 0; task_nbr < numberOfJobs; task_nbr++) {
unsigned char *p = (unsigned char *) malloc(
messageSize * sizeof(unsigned char));
memset((void *) p, 'A', messageSize);
p[messageSize - 1] = '\0';
s_send(sender, p);
free(p);
}
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]);
void *context = zmq_ctx_new();
void *receiver = zmq_socket(context, ZMQ_PULL);
zmq_connect(receiver, "tcp://localhost:5557");
void *sender = zmq_socket(context, ZMQ_PUSH);
zmq_connect(sender, "tcp://localhost:5558");
while (1) {
char *string = s_recv(receiver);
fflush(stdout);
s_sleep(workLoad);
free(string);
s_send(sender, "Done");
}
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);
void *context = zmq_ctx_new();
void *receiver = zmq_socket(context, ZMQ_PULL);
zmq_bind(receiver, "tcp://*:5558");
char *string = s_recv(receiver);
free(string);
int64_t start_time = s_clock();
int task_nbr;
for (task_nbr = 0; task_nbr < numberOfConfirmations; task_nbr++) {
char *string = s_recv(receiver);
if ((task_nbr / 10) * 10 == task_nbr) {
printf("%s: %d\n", string, task_nbr);
} else {
printf(".");
}
free(string);
fflush(stdout);
}
printf("Total elapsed time: %d seconds\n",
(int) (s_clock() - start_time) / 1000);
zmq_close(receiver);
zmq_ctx_destroy(context);
return 0;
}
*****************************    


To Run:

1. Open three terminal windows and move to the directory where the tasksink, taskwork, and taskvent executables exist.

2. Start tasksink with a numberOfJobs argument

 ./tasksink 20000

3. Start the Workers with a 50K Workload. You can run the single command:

 ./taskwork 50000

Or run a UNIX script “startWorkers.sh” that looks like this with a number of arguments:

./startWorkers 50000 1000

c=1
while [ $c -le $2 ]
do
       ./taskwork $1 &
       echo "./taskwork $1 &"
       ((c++))
done. 

If you repeat the test with 1025 processes instead of 1000, the Jobs will not complete. 

Any help would be appreciated.

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