protected void sendMessages() throws Exception {
        Message[] msgs=generateMessages(num_msgs);
        CountDownLatch latch=new CountDownLatch(1);
        AtomicInteger index=new AtomicInteger(0);
        Sender[] senders=new Sender[num_senders];
        for(int i=0; i < senders.length; i++) {
            senders[i]=new Sender(latch, msgs, index);
            senders[i].start();
        }

        long start=Util.micros();
        latch.countDown(); // starts all sender threads

        for(Sender sender: senders)
            sender.join();

        // wait until the bundler has no pending msgs left
        long park_time=1;
        for(int i=0; i < 1_000_000; i++) {
            int pending_msgs=bundler.size();
            if(pending_msgs == 0)
                break;

            LockSupport.parkNanos(park_time);
            if(i % 10000 == 0) {
                park_time=Math.min(park_time*2, 1_000_000); // 1 ms max park time
            }

        }
        if(bundler.size() > 0)
            throw new Exception(String.format("bundler still has %d pending messages", bundler.size()));

        long time_us=Util.micros()-start;
        AverageMinMax send_avg=null;
        for(Sender sender: senders) {
            if(details)
                System.out.printf("[%d] count=%d, send-time = %s\n", sender.getId(), sender.send.count(), sender.send);
            if(send_avg == null)
                send_avg=sender.send;
            else
                send_avg.merge(sender.send);
        }




        double msgs_sec=num_msgs / (time_us / 1_000.0);
        System.out.printf(Util.bold("\n\nreqs/ms    = %.2f (time: %d us)" +
                                      "\nsend-time  = min/avg/max: %d / %.2f / %d ns\n"),
                          msgs_sec, time_us, send_avg.min(), send_avg.average(), send_avg.max());
    }