Re: Fast Semaphore

From:

Joe Seigh <jseigh_01@xemaps.com>

Newsgroups:

comp.lang.java.programmer

Date:

Sat, 07 Apr 2007 10:22:20 -0400

Message-ID:

<R-2dnafiyNMRNYrbnZ2dnUVZ_vumnZ2d@comcast.com>

Robert Klemme wrote:

On 07.04.2007 00:04, Daniel Pitts wrote:

On Apr 6, 3:44 am, Joe Seigh <jseigh...@xemaps.com> wrote:

Here's a port of a fast pathed semaphore I did elsewhere. It
only does single permit acquire and release. If you use it
in conjunction with ConcurrentLinkedQueue you can get a blocking
queuue that's up to 3X faster than LinkedBlockingQueue under
contention.

What benchmarks did you use? Code?

I'm curious, too, how it compares to j.u.c.Semaphore.

Ok. I have a new design pattern that subsets interfaces so
I can compare apple to orange implementations without having
to implement the fruit, plant, multi-celled organism, cell,
dna, organic molecule, molecule, and atom interfaces for
orange. Java's architects really have nothing better to do
with other people's time.

import java.util.LinkedList;
import java.util.concurrent.*;
import java.util.*;

interface Sem {
    public void acquire() throws InterruptedException;
    public void release();
}

class NSem extends Semaphore implements Sem {
    public NSem(int n, boolean fair) {super(n, fair); }
}

class FSem extends FastSemaphore implements Sem {
    public FSem(int n, boolean fair) { super(n, fair); }
}

interface fifo<T> {
    public void queue(T o);
    public T dequeue() throws InterruptedException;
}

class ConcurrentFifoQueue<T>
    implements fifo<T>
{
    ConcurrentLinkedQueue<T> queue;
    Sem sem;

    ConcurrentFifoQueue(Sem s) {
        queue = new ConcurrentLinkedQueue<T>();
        sem = s;
    }
    public void queue(T o) {
        queue.offer(o);
        sem.release();
    }
    public T dequeue() throws InterruptedException {
        sem.acquire();
        return queue.poll();
    }
}

class BlockingFifoQueue<T>
extends java.util.concurrent.LinkedBlockingQueue<T>
implements fifo<T>
{
    public void queue(T o) { try {put(o); } catch (InterruptedException e) {} }
    public T dequeue() throws InterruptedException { return take(); }
}

public class qtest {

    /**
     * multi-threaded queueing test
     *
     */
    final static int loopcount = 20000;
    final static int nodecount = 200;
    final static int threadcount = 20;
    final static Formatter fmt = new java.util.Formatter(System.out);

    public void test(final fifo<Object> fullq, final fifo<Object> emptyq, String desc) {
        int j;
        long t0, t1; // start, stop times
        Thread producer[] = new Thread[threadcount];
        Thread consumer[] = new Thread[threadcount];
        final CyclicBarrier barrier = new CyclicBarrier(producer.length + consumer.length + 1);

        for (j = 0; j < nodecount; j++) {
            emptyq.queue(new Object());
        }

        for (j = 0; j < producer.length; j++) {
            producer[j] = new Thread(new Runnable() {
                public void run() {
                    try {barrier.await(); } catch (Exception e) {}
                    for (int j = 0; j < loopcount; j++) {
                        try {fullq.queue(emptyq.dequeue()); } catch (InterruptedException e) {}
                    }
                    try {barrier.await(); } catch (Exception e) {}
                }
            });
            producer[j].setDaemon(true);
            producer[j].start();
        }

        for (j = 0; j < consumer.length; j++) {
            consumer[j] = new Thread(new Runnable() {
                public void run() {
                    try {barrier.await(); } catch (Exception e) {}
                    for (int j = 0; j < loopcount; j++) {
                        try {emptyq.queue(fullq.dequeue()); } catch (InterruptedException e) {}
                    }
                    try {barrier.await(); } catch (Exception e) {}
                }
            });
            consumer[j].setDaemon(true);
            consumer[j].start();
        }

        try {barrier.await(); } catch (Exception e) {}
        t0 = System.nanoTime();

        try {barrier.await(); } catch (Exception e) {}
        t1 = System.nanoTime();
        double x = ((t1 - t0)/1e9);
        System.out.println(desc + ":");
        fmt.format("runtime = %12.9f secs\n\n", x);
    }

    public static void main(String[] args) {

        qtest q = new qtest();

        fmt.format("loop count = %6d\n", loopcount);
        fmt.format("queue size = %6d\n", nodecount);
        fmt.format("producer count = %6d\n", threadcount);
        fmt.format("consumer count = %6d\n\n", threadcount);

        q.test(
                new BlockingFifoQueue<Object>(),
                new BlockingFifoQueue<Object>(),
                "LinkedBlockingQueue");

        q.test(
                new ConcurrentFifoQueue<Object>(new NSem(0, false)),
                new ConcurrentFifoQueue<Object>(new NSem(0, false)),
                "ConcurrentLinkedQueue w/ unfair semaphore");

        q.test(
                new ConcurrentFifoQueue<Object>(new NSem(0, true)),
                new ConcurrentFifoQueue<Object>(new NSem(0, true)),
                "ConcurrentLinkedQueue w/ fair semaphore");

        q.test(
                new ConcurrentFifoQueue<Object>(new FSem(0, false)),
                new ConcurrentFifoQueue<Object>(new FSem(0, false)),
                "ConcurrentLinkedQueue w/ unfair fast semaphore");

        q.test(
                new ConcurrentFifoQueue<Object>(new FSem(0, true)),
                new ConcurrentFifoQueue<Object>(new FSem(0, true)),
                "ConcurrentLinkedQueue w/ fair fast semaphore");

        System.out.println("qtest exiting...");

    }

}

--
Joe Seigh

When you get lemons, you make lemonade.
When you get hardware, you make software.