[concurrency-interest] Enforcing ordered execution of critical sections?

Martin Buchholz martinrb at google.com
Mon Dec 22 02:52:41 EST 2014


It may be a bug in the signal/propagate part of AQS, one of the trickiest parts.
I can reproduce this and hacky-ported the bug to jsr166 CVS:

Index: SemaphoreTest.java
===================================================================
RCS file: /export/home/jsr166/jsr166/jsr166/src/test/tck/SemaphoreTest.java,v
retrieving revision 1.31
diff -u -r1.31 SemaphoreTest.java
--- SemaphoreTest.java    25 Jul 2011 19:01:08 -0000    1.31
+++ SemaphoreTest.java    22 Dec 2014 07:49:03 -0000
@@ -8,6 +8,7 @@

 import junit.framework.*;
 import java.util.*;
+import java.util.concurrent.*;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Semaphore;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -630,4 +631,95 @@
         assertTrue(s.toString().contains("Permits = -2"));
     }

+    static class OrderedExecutor<T> {
+
+        private final Semaphore semaphore = new Semaphore(2);
+
+        public T execCallableInOrder(int order, Callable<T> callable)
+            throws Exception {
+            assert order >= 1;
+            final int acquires = order + 1;
+            syserr("Acquiring " + acquires + " permits with " +
semaphore.availablePermits() + "  available");
+            semaphore.acquire(acquires);
+            syserr("Acquired " + acquires + " permits with " +
semaphore.availablePermits() + "  available");
+            try {
+                return callable.call();
+            } finally {
+                syserr("Releasing " + (acquires+1) + " permits with
"+ semaphore.availablePermits() + "  available");
+                semaphore.release(acquires + 1);
+                syserr("Released " + (acquires+1) + " permits with "+
semaphore.availablePermits() + "  available");
+            }
+        }
+
+        private synchronized static void syserr(Object o) {
+            System.err.println(String.valueOf(o));
+        }
+    }
+
+    public void testFoo() throws InterruptedException, ExecutionException {
+        Random rand = new Random();
+        final OrderedExecutor<Integer> oe = new OrderedExecutor<Integer>();
+        final int[] counter = new int[1];
+        // Number of critical sections to be executed
+        Integer[] a = new Integer[10];
+        for (int i=0; i< a.length; i++)
+            a[i] = i+1;
+        List<Integer> orders = new ArrayList<Integer>(Arrays.asList(a));
+        // Shuffle them so they will be submitted in random order for execution
+        Collections.shuffle(orders);
+        ExecutorService es = Executors.newCachedThreadPool();
+        List<Future<Integer>> futures = new ArrayList<Future<Integer>>();
+        for (final Integer order: orders) {
+            sysout("Submitting task " + order);
+            // Submit a task that involves both critical and
non-critical sections
+            futures.add(es.submit(new Callable<Integer>() {
+                @Override
+                public Integer call() throws Exception {
+                    Callable<Integer> criticalSection = new
Callable<Integer>() {
+                        @Override
+                        public Integer call() {
+                          syserr("executing critical section " + order);
+                          assertTrue(++counter[0] == order);
+                          try {
+                            Thread.sleep(rand.nextInt(200));
+                          } catch (InterruptedException e) {
+                            throw new Error(e);
+                          }
+                          return order;
+                        }
+                    };
+                    // Executes the critical section in order,
blocking as necessary
+                    Integer ret = oe.execCallableInOrder(order,
criticalSection);
+                    // The rest is non-critical and can be executed
with no order imposed
+                    sysout("start non-critical section " + order);
+                    try {
+                        Thread.sleep(rand.nextInt(1000));
+                    } catch (InterruptedException e) {
+                        throw new Error(e);
+                    }
+                    sysout("end non-critical section " + order);
+                    return ret;
+                }
+            }));
+        }
+        int count = 0;
+        Iterator<Integer> expected = orders.iterator();
+        for (Future<Integer> future: futures) {
+            final int val = future.get();
+            assertTrue("count="+count + ", val=" + val, val ==
expected.next().intValue());
+            count++;
+        }
+        assertTrue(a.length == count);
+        assertTrue(a.length == counter[0]);
+    }
+
+    private synchronized static void sysout(Object o) {
+        System.out.println(String.valueOf(o));
+    }
+
+    private synchronized static void syserr(Object o) {
+        System.err.println(String.valueOf(o));
+    }
+
+
 }


More information about the Concurrency-interest mailing list