queuetest.java

来自「JGRoups源码」· Java 代码 · 共 1,041 行 · 第 1/2 页

//                    }
//                }
//            }.start();
//
//            queue.waitUntilEmpty(0);
//            assertEquals(queue.size(), 0);
//        }
//        catch(Exception e) {
//            e.printStackTrace();
//            fail(e.toString());
//        }
//    }
//
//    public void testWaitUntilEmpty2() {
//        try {
//            queue.add("one");
//            queue.add("two");
//            queue.add("three");
//
//            new Thread() {
//                public void run() {
//                    try {
//                        sleep(1000);
//                        queue.remove();
//                        queue.remove();
//                    }
//                    catch(Exception e) {
//                    }
//                }
//            }.start();
//
//            queue.waitUntilEmpty(3000);
//            fail("shouldn't get here; we should have caught a TimeoutException");
//        }
//        catch(TimeoutException timeout) {
//            assertTrue(true);
//        }
//        catch(Exception e) {
//            e.printStackTrace();
//            fail(e.toString());
//        }
//    }
//
//
//    public void testWaitUntilQueueClosed() {
//         try {
//            queue.add("one");
//            queue.add("two");
//            queue.add("three");
//
//            new Thread() {
//                public void run() {
//                    try {
//                        sleep(1000);
//                        queue.close(false);
//                    }
//                    catch(Exception e) {
//                    }
//                }
//            }.start();
//
//            queue.waitUntilEmpty(0);
//            fail("shouldn't get here; we should have caught a QueueClosedException");
//        }
//        catch(TimeoutException timeout) {
//            fail("we should not have gottem here");
//        }
//         catch(QueueClosedException ex2) {
//             assertTrue(true);
//         }
//        catch(Exception e) {
//             e.printStackTrace();
//             fail();
//        }
//    }


    /** Multiple threads call remove(), one threads then adds an element. Only 1 thread should actually terminate
     * (the one that has the element) */
    public void testBarrier() {
        RemoveOneItem[] removers=new RemoveOneItem[10];
        int num_dead=0;

        for(int i=0; i < removers.length; i++) {
            removers[i]=new RemoveOneItem(i);
            removers[i].start();
        }

        Util.sleep(1000);

        System.out.println("-- adding element 99");
        try {
            queue.add(new Long(99));
        }
        catch(Exception ex) {
            System.err.println(ex);
        }

        Util.sleep(5000);
        System.out.println("-- adding element 100");
        try {
            queue.add(new Long(100));
        }
        catch(Exception ex) {
            System.err.println(ex);
        }

        Util.sleep(1000);

        for(int i=0; i < removers.length; i++) {
            System.out.println("remover #" + i + " is " + (removers[i].isAlive() ? "alive" : "terminated"));
            if(!removers[i].isAlive()) {
                num_dead++;
            }
        }

        assertEquals(2, num_dead);
    }

    /** Multiple threads call remove(), one threads then adds an element. Only 1 thread should actually terminate
     * (the one that has the element) */
    public void testBarrierWithTimeOut()
    {
        RemoveOneItemWithTimeout[] removers = new RemoveOneItemWithTimeout[10];
        int num_dead = 0;

        for (int i = 0; i < removers.length; i++)
        {
            removers[i] = new RemoveOneItemWithTimeout(i, 1000);
            removers[i].start();
        }

        Util.sleep(5000);

        System.out.println("-- adding element 99");
        try
        {
            queue.add(new Long(99));
        }
        catch (Exception ex)
        {
            System.err.println(ex);
        }

        Util.sleep(5000);
        System.out.println("-- adding element 100");
        try
        {
            queue.add(new Long(100));
        }
        catch (Exception ex)
        {
            System.err.println(ex);
        }

        Util.sleep(1000);

        for (int i = 0; i < removers.length; i++)
        {
            System.out.println("remover #" + i + " is " + (removers[i].isAlive() ? "alive" : "terminated"));
            if (!removers[i].isAlive())
            {
                num_dead++;
            }
        }

        assertEquals(2, num_dead);

        queue.close(false); // will cause all threads still blocking on peek() to return

        Util.sleep(2000);

        num_dead = 0;
        for (int i = 0; i < removers.length; i++)
        {
            System.out.println("remover #" + i + " is " + (removers[i].isAlive() ? "alive" : "terminated"));
            if (!removers[i].isAlive())
            {
                num_dead++;
            }
        }
        assertEquals(10, num_dead);

    }


    /** Multiple threads add one element, one thread read them all.
     * (the one that has the element) */
    public void testMultipleWriterOneReader()
    {
        AddOneItem[] adders = new AddOneItem[10];
        int num_dead = 0;
        int num_items = 0;
        int items = 1000;

        for (int i = 0; i < adders.length; i++)
        {
            adders[i] = new AddOneItem(i, items);
            adders[i].start();
        }

        while (num_items < (adders.length*items))
        {
            try
            {
                queue.remove();
                num_items++;
            }
            catch (Exception ex)
            {
                System.err.println(ex);
            }
        }

        Util.sleep(1000);

        for (int i = 0; i < adders.length; i++)
        {
            System.out.println("adder #" + i + " is " + (adders[i].isAlive() ? "alive" : "terminated"));
            if (!adders[i].isAlive())
            {
                num_dead++;
            }
        }

        assertEquals(10, num_dead);

        queue.close(false); // will cause all threads still blocking on peek() to return
    }


    /**
     * Times how long it takes to add and remove 1000000 elements concurrently (1 reader, 1 writer)
     */
    public void testConcurrentAddRemove() {
        final long   NUM=1000000;
        long         num_received=0;
        Object       ret;
        long         start, stop;

        start=System.currentTimeMillis();

        new Thread() {
            public void run() {
                for(int i=0; i < NUM; i++) {
                    try {
                        queue.add(new Object());
                    }
                    catch(QueueClosedException e) {
                    }
                }
            }
        }.start();

        while(num_received < NUM) {
            try {
                ret=queue.remove();
                if(ret != null)
                    num_received++;
            }
            catch(QueueClosedException e) {
                e.printStackTrace();
                fail();
            }
        }
        assertEquals(NUM, num_received);
        stop=System.currentTimeMillis();
        System.out.println("time to add/remove " + NUM + " elements: " + (stop-start));
    }



    /** Has multiple threads add(), remove() and peek() elements to/from the queue */
    public void testConcurrentAccess() {
        final int NUM_THREADS=10;
        final int INTERVAL=20000;

        Writer[] writers=new Writer[NUM_THREADS];
        Reader[] readers=new Reader[NUM_THREADS];
        int[] writes=new int[NUM_THREADS];
        int[] reads=new int[NUM_THREADS];
        long total_reads=0, total_writes=0;


        for(int i=0; i < writers.length; i++) {
            readers[i]=new Reader(i, reads);
            readers[i].start();
            writers[i]=new Writer(i, writes);
            writers[i].start();
        }

        Util.sleep(INTERVAL);

        System.out.println("current queue size=" + queue.size());

        for(int i=0; i < writers.length; i++) {
            writers[i].stopThread();
        }

        for(int i=0; i < readers.length; i++) {
            readers[i].stopThread();
        }

        queue.close(false); // will cause all threads still blocking on peek() to return

        System.out.println("current queue size=" + queue.size());

        for(int i=0; i < writers.length; i++) {
            try {
                writers[i].join(300);
                readers[i].join(300);
            }
            catch(Exception ex) {
                System.err.println(ex);
            }
        }


        for(int i=0; i < writes.length; i++) {
            System.out.println("Thread #" + i + ": " + writes[i] + " writes, " + reads[i] + " reads");
            total_writes+=writes[i];
            total_reads+=reads[i];
        }
        System.out.println("total writes=" + total_writes + ", total_reads=" + total_reads +
                ", diff=" + Math.abs(total_writes - total_reads));
    }

    class AddOneItem extends Thread
    {
        Long retval = null;
        int rank = 0;
        int iteration = 0;

        AddOneItem(int rank, int iteration)
        {
            super("AddOneItem thread #" + rank);
            this.rank = rank;
            this.iteration = iteration;
            setDaemon(true);
        }

        public void run()
        {
            try
            {
                for (int i = 0; i < iteration; i++)
                {
                    queue.add(new Long(rank));
                    // Util.sleepRandom(1);
                    // System.out.println("Thread #" + rank + " added element (" + rank + ")");
                }
            }
            catch (QueueClosedException closed)
            {
                System.err.println("Thread #" + rank + ": queue was closed");
            }
        }

    }

    class RemoveOneItem extends Thread {
        Long retval=null;
        int rank=0;


        RemoveOneItem(int rank) {
            super("RemoveOneItem thread #" + rank);
            this.rank=rank;
            setDaemon(true);
        }

        public void run() {
            try {
                retval=(Long)queue.remove();
                // System.out.println("Thread #" + rank + " removed element (" + retval + ")");
            }
            catch(QueueClosedException closed) {
                System.err.println("Thread #" + rank + ": queue was closed");
            }
        }

        Long getRetval() {
            return retval;
        }
    }

    class RemoveOneItemWithTimeout extends Thread
    {
        Long retval = null;
        int rank = 0;
        long timeout = 0;

        RemoveOneItemWithTimeout(int rank, long timeout)
        {
            super("RemoveOneItem thread #" + rank);
            this.rank = rank;
            this.timeout=timeout;
            setDaemon(true);
        }

        public void run()
        {
            boolean finished = false;
            while (!finished)
            {
                try
                {
                    retval = (Long) queue.remove(timeout);
                    // System.out.println("Thread #" + rank + " removed element (" + retval + ")");
                    finished = true;
                }
                catch (QueueClosedException closed)
                {
                    System.err.println("Thread #" + rank + ": queue was closed");
                    finished = true;
                }
                catch (TimeoutException e)
                {
                }
            }
        }

        Long getRetval()
        {
            return retval;
        }
    }




    class Writer extends Thread {
        int rank=0;
        int num_writes=0;
        boolean running=true;
        int[] writes=null;

        Writer(int i, int[] writes) {
            super("WriterThread");
            rank=i;
            this.writes=writes;
            setDaemon(true);
        }


        public void run() {
            while(running) {
                try {
                    queue.add(new Long(System.currentTimeMillis()));
                    num_writes++;
                }
                catch(QueueClosedException closed) {
                    running=false;
                }
                catch(Throwable t) {
                    System.err.println("QueueTest.Writer.run(): exception=" + t);
                }
            }
            writes[rank]=num_writes;
        }

        void stopThread() {
            running=false;
        }
    }


    class Reader extends Thread {
        int rank;
        int num_reads=0;
        int[] reads=null;
        boolean running=true;


        Reader(int i, int[] reads) {
            super("ReaderThread");
            rank=i;
            this.reads=reads;
            setDaemon(true);
        }


        public void run() {
            Long el;

            while(running) {
                try {
                    el=(Long)queue.remove();
                    if(el == null) { // @remove
                        System.out.println("QueueTest.Reader.run(): peek() returned null element. " +
                                "queue.size()=" + queue.size() + ", queue.closed()=" + queue.closed());
                    }
                    assertNotNull(el);
                    num_reads++;
                }
                catch(QueueClosedException closed) {
                    running=false;
                }
                catch(Throwable t) {
                    System.err.println("QueueTest.Reader.run(): exception=" + t);
                }
            }
            reads[rank]=num_reads;
        }


        void stopThread() {
            running=false;
        }

    }


    public static void main(String[] args) {
        String[] testCaseName={QueueTest.class.getName()};
        junit.textui.TestRunner.main(testCaseName);
    }

}