Re: JMock not thread safe - how does one do multi-threaded testing?

2009/4/1 James Abley <james.abley@...>:
Works perfectly, plus gives me nice failures if I try to use the
non-thread-safe version. Lovely job.

package org.jmock.test.unit.lib.concurrent;

import static org.junit.Assert.*;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.atomic.AtomicInteger;

import org.jmock.Expectations;
import org.jmock.Mockery;
import org.jmock.integration.junit4.JMock;
import org.jmock.integration.junit4.JUnit4Mockery;
import org.jmock.lib.concurrent.SynchronisingImposteriser;
import org.junit.Test;
import org.junit.runner.RunWith;

@RunWith(JMock.class)
public class ConcurrentRequestsTest {

    /**
     * JMock context object.
     */
    private final Mockery context = new JUnit4Mockery() {
        {
            /* Need to use the SynchronisingImposteriser otherwise it
will fail when use in a multi-threaded situation. */
            setImposteriser(new SynchronisingImposteriser());
        }
    };

    @Test
    public void onlyASingleClientPopulatesTheCache() throws Exception {
        final Cache cache = new Cache();

        int clientRequests = 100;

        final Integer key = Integer.valueOf(43);

        ExecutorService exec = Executors.newFixedThreadPool(clientRequests);

        final CountDownLatch startGate = new CountDownLatch(1);
        final CountDownLatch endGate = new CountDownLatch(clientRequests);

        List<Future<Integer>> results = new ArrayList<Future<Integer>>();

        for (int i = 0; i < clientRequests; ++i) {
            final Request request = context.mock(Request.class, "request-" + i);

            context.checking(new Expectations() {
                {
                    allowing(request).getKey();
                    will(returnValue(key));

                    allowing(request).getCallable();
                    will(returnValue(new Callable<Integer>() {

                        public Integer call() throws Exception {

                            /*
                             * Simulate an operation taking a little
while to complete.
                             */
                            Thread.sleep(10);
                            return request.getKey();
                        }
                    }));
                }
            });

            FutureTask<Integer> requestThreadCall = new
FutureTask<Integer>(new Callable<Integer>() {

                public Integer call() throws Exception {
                    startGate.await();
                    try {
                        return cache.retrieveFromCache(request);
                    } finally {
                        endGate.countDown();
                    }
                }
            });

            results.add(requestThreadCall);
            exec.execute(requestThreadCall);
        }

        startGate.countDown();
        endGate.await();

        for (Future<Integer> result : results) {
            assertEquals(key, result.get());
        }

        assertEquals("Only a single cache miss", 1, cache.missCount.intValue());
        assertEquals("everything else was a cache hit or waited for
task completion", clientRequests - 1,
                cache.hitCount.intValue() + cache.waitingCount.intValue());
    }

    static interface Request {

        Integer getKey();

        Callable<Integer> getCallable();
    }

    static class Cache {

        private final ConcurrentMap<Integer, Future<Integer>> cache;
        private final AtomicInteger hitCount;
        private final AtomicInteger missCount;
        private final AtomicInteger waitingCount;

        Cache() {
            this.cache = new ConcurrentHashMap<Integer, Future<Integer>>();
            this.hitCount = new AtomicInteger();
            this.missCount = new AtomicInteger();
            this.waitingCount = new AtomicInteger();
        }

        Integer retrieveFromCache(final Request request) throws
InterruptedException, ExecutionException {
            Future<Integer> value = cache.get(request.getKey());

            if (value == null) {
                FutureTask<Integer> thunk = new
FutureTask<Integer>(request.getCallable());

                value = cache.putIfAbsent(request.getKey(), thunk);

                if (value == null) {
                    cacheMiss();
                    thunk.run();
                    value = thunk;
                } else {
                    cacheHitOrWaitingForCompletion();
                }
            } else {
                cacheHit();
            }

            return value.get();
        }

        private void cacheHit() {
            this.hitCount.incrementAndGet();
        }

        private void cacheHitOrWaitingForCompletion() {
            this.waitingCount.incrementAndGet();
        }

        private void cacheMiss() {
            this.missCount.incrementAndGet();
        }
    }
}

Every time I think I've found a missing feature, I should just check
what's in Subversion since you've always already implemented it!

Thanks,

James

---------------------------------------------------------------------
To unsubscribe from this list, please visit:

    http://xircles.codehaus.org/manage_email