3
\$\begingroup\$

For a personal project, I'm working on a Java profiler that specifically targets ThreadPoolExecutor and its subclasses and provides statistics about throughput of executed tasks. To support this I've written the following utility classes, which are intended to help compute a smoothed average rate of events over time. They are intended to be thread-safe and lock-free; I want to interfere as little as possible with the threads that are counting (and executing) tasks.

First is the EventMeter, which is used to count one type of event. This supports computing an average rate of events by keeping a running sum of counted events and being able to discard old data. (Pardon my TODO.)

public final class EventMeter {
    private final long bucketLengthNanos;
    private final double[] buckets;
    private volatile double sum;
    private int bucketNum;
    private AtomicLong accumulator;

    public EventMeter(int numBuckets, long bucketLengthNanos) {
        // TODO assert that numBuckets >= 1, bucketLengthNanos >= some small number

        this.bucketLengthNanos = bucketLengthNanos;
        this.buckets = new double[numBuckets];
        this.sum = 0.0;
        this.bucketNum = 0;
        this.accumulator = new AtomicLong(Double.doubleToLongBits(0.0));
    }

    public long getBucketLengthNanos() {
        return bucketLengthNanos;
    }

    public void countEvent() {
        long oldAccumulator;
        long newAccumulator;
        do {
            oldAccumulator = accumulator.get();
            newAccumulator = Double.doubleToLongBits(Double.longBitsToDouble(oldAccumulator) + 1.0);
        } while (!accumulator.compareAndSet(oldAccumulator, newAccumulator));
    }

    public double getRate() {
        return sum / buckets.length;
    }

    public void fillBucketFromAccumulator(double accumulatorFraction) {
        long oldAccumulator;
        long newAccumulator;
        do {
            oldAccumulator = accumulator.get();
            newAccumulator = Double.doubleToLongBits(Double.longBitsToDouble(oldAccumulator) * (1.0 - accumulatorFraction));
        } while (!accumulator.compareAndSet(oldAccumulator, newAccumulator));

        double oldAccumulatorValue = Double.longBitsToDouble(oldAccumulator);

        double oldBucketContents = buckets[bucketNum];
        double newBucketContents = oldAccumulatorValue * accumulatorFraction;

        buckets[bucketNum] = newBucketContents;
        bucketNum = (bucketNum + 1) % buckets.length;

        sum = sum - oldBucketContents + newBucketContents;
    }
}

Next is the EventMeterManager, which uses a worker thread to advance the "buckets" in a collection of EventMeters at a constant rate.

public final class EventMeterManager {
    private final DelayQueue<AdvanceMeterTask> meterQueue;
    private final Map<EventMeter, AdvanceMeterTask> tasks;
    private final AtomicInteger activeMeters;
    private AtomicReference<Thread> worker;

    public EventMeterManager() {
        this.meterQueue = new DelayQueue<AdvanceMeterTask>();
        this.worker = new AtomicReference<Thread>();
        this.tasks = new ConcurrentHashMap<EventMeter, AdvanceMeterTask>();
        this.activeMeters = new AtomicInteger(0);
    }

    public EventMeter startEventMeter(int numBuckets, long bucketLengthNanos) {
        final EventMeter meter = new EventMeter(numBuckets, bucketLengthNanos);

        if (activeMeters.incrementAndGet() == 1) {
            final Thread newWorker = new Thread(new DriverWorker(), "EventMeterManager worker");

            // It will either already be null or it will be shortly.
            while (!worker.compareAndSet(null, newWorker)) {}

            newWorker.start();
        }

        final AdvanceMeterTask advanceTask = new AdvanceMeterTask(meter, System.nanoTime());
        tasks.put(meter, advanceTask);
        meterQueue.put(advanceTask);

        return meter;
    }

    public void stopEventMeter(EventMeter meter) {
        final AdvanceMeterTask cancelledTask = tasks.remove(meter);
        if (cancelledTask != null) {
            cancelledTask.cancel();

            if (activeMeters.decrementAndGet() == 0) {
                worker.get().interrupt();
                worker.set(null);
            }
        }
    }

    private class DriverWorker implements Runnable {
        @Override
        public void run() {
            try {
                while (!Thread.interrupted()) {
                    final AdvanceMeterTask task = meterQueue.take();
                    task.run();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    private class AdvanceMeterTask implements Delayed, Runnable {
        private final EventMeter meter;
        private long bucketStartTime;
        private long bucketEndTime;
        private volatile boolean cancelled;

        public AdvanceMeterTask(final EventMeter meter, final long firstBucketStartTime) {
            this.meter = meter;
            this.bucketStartTime = firstBucketStartTime;
            this.bucketEndTime = this.bucketStartTime + meter.getBucketLengthNanos();
        }

        @Override
        public int compareTo(Delayed o) {
            final long difference = getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
            if (difference < 0) {
                return -1;
            } else if (difference > 0) {
                return 1;
            } else {
                return 0;
            }
        }

        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(bucketEndTime - System.nanoTime(), TimeUnit.NANOSECONDS);
        }

        @Override
        public void run() {
            if (!cancelled) {
                final long now = System.nanoTime();
                final long bucketLength = meter.getBucketLengthNanos();

                while (bucketEndTime < now) {
                    double bucketFraction = (double) bucketLength / (now - bucketStartTime);
                    meter.fillBucketFromAccumulator(bucketFraction);
                    bucketStartTime = bucketEndTime;
                    bucketEndTime += bucketLength;
                }

                meterQueue.put(this);
            }
        }

        public void cancel() {
            this.cancelled = true;
        }
    }
}

In addition to general commentary (which is always appreciated) I'd like specific feedback, if possible on these things:

  • Are these classes as thread-safe as I hope they are?
  • These classes seem pretty coupled but in the interest of not creating a bunch of unnecessary threads I didn't want EventMeter to create its own worker thread. Is there an approach I can take allows EventMeters to share a thread without requiring a client to use EventMeterManager?
\$\endgroup\$
  • 1
    \$\begingroup\$ What's the point of the buckets? \$\endgroup\$ – Craig Aug 9 '11 at 13:50
2
\$\begingroup\$

fillBucketFromAccumulator() is not thread-safe. Two threads can write to the same slot in the array and then both advance by one.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.