sink.pl

#!/usr/bin/perl

use warnings;

use ZMQ::LibZMQ3;
use ZMQ::Constants qw(ZMQ_PULL);

my $context = zmq_init;

# Receiving work effect from workers on :5559
my $receiver = zmq_socket($context, ZMQ_PULL);
zmq_bind($receiver, 'tcp://127.0.0.1:5559');

while(1) {
    my $data = zmq_recvmsg($receiver);
    $data = zmq_msg_data($data);
    print("Received: $data");
}

ventilator.pl

#!/usr/bin/perl

use warnings;

use ZMQ::LibZMQ3;
use ZMQ::Constants qw(ZMQ_PUSH);

my $context = zmq_init;

# Sending works to workers on :5558
my $sender = zmq_socket($context, ZMQ_PUSH);
zmq_bind($sender, 'tcp://127.0.0.1:5558');

# List directory from 1st argument
my $dirs = `ls $ARGV[0]`;

print("Press any key when all workers are connected\n");
<STDIN>;

foreach (split(/\n/, $dirs)) {
    my $toSend = "$ARGV[0]/$_";
    print("sending $toSend\n");
    zmq_sendmsg($sender, $toSend);
}

# Wait until keypress
<STDIN>;

worker.pl

#!/usr/bin/perl

use warnings;

use ZMQ::LibZMQ3;
use ZMQ::Constants qw(ZMQ_PULL);
use ZMQ::Constants qw(ZMQ_PUSH);

my $context = zmq_init;

# Receiving works from ventilator on :5558
my $receiver = zmq_socket($context, ZMQ_PULL);
zmq_connect($receiver, 'tcp://127.0.0.1:5558');

# Sending work effect to sink on :5559
my $sender = zmq_socket($context, ZMQ_PUSH);
zmq_connect($sender, 'tcp://127.0.0.1:5559');

while(1)
{
    my $data = zmq_recvmsg($receiver);
    $data = zmq_msg_data($data);
    
    print("Received: $data\n");
    zmq_sendmsg($sender, `ls $data`);
}