public void send(Message msg) throws Exception {
        if(msg == null)
            throw new IllegalArgumentException("message must not be null");
        num_threads.incrementAndGet();

        int tmp_index=getWriteIndex(); // decrements write_permits
        // System.out.printf("[%d] tmp_index=%d\n", Thread.currentThread().getId(), tmp_index);
        if(tmp_index == -1) {
            log.warn("buf is full (num_permits: %d, bundler: %s)\n", write_permits.get(), toString());
            num_threads.decrementAndGet();
            return;
        }

        buf[tmp_index]=msg;
        long acc_bytes=accumulated_bytes.addAndGet(msg.size());
        int current_threads=num_threads.decrementAndGet();
        boolean no_other_threads=current_threads == 0;

        boolean unpark=(acc_bytes >= transport.getMaxBundleSize() && accumulated_bytes.compareAndSet(acc_bytes, 0))
          ||  no_other_threads;

        // only 2 threads at a time should do this (1st cond and 2nd cond), so we have to reduce this to
        // 1 thread as advanceWriteIndex() is not thread safe
        if(unpark && unparking.compareAndSet(false, true)) {
            int num_advanced=advanceWriteIndex();
            size.addAndGet(num_advanced);
            if(num_advanced > 0) {
                Thread thread=bundler_thread.getThread();
                if(thread != null)
                    LockSupport.unpark(thread);
            }
            unparking.set(false);
        }
    }