Revision implements CacheValue { static Revision newRevision(int clusterId) { long timestamp = getCurrentTimestamp(); int c; synchronized (Revision.class) { if (timestamp < lastRevisionTimestamp) { timestamp = lastRevisionTimestamp; } if (timestamp == lastRevisionTimestamp) { c = ++lastRevisionCount; } else { lastRevisionTimestamp = timestamp; lastRevisionCount = c = 0; } } return new Revision(timestamp, c, clusterId); } Revision(long timestamp, int counter, int clusterId);  Revision(long timestamp, int counter, int clusterId, boolean branch); static long getCurrentTimestamp(); static long getTimestampDifference(Revision r1, Revision r2); static Revision fromString(String rev); @Override String toString(); StringBuilder toStringBuilder(StringBuilder sb); String toReadableString(); long getTimestamp(); int getCounter(); boolean isBranch(); Revision asBranchRevision(); Revision asTrunkRevision(); @Override int hashCode(); @Override boolean equals(Object other); int getClusterId(); @Override int getMemory();  }
@Test public void uniqueRevision2() throws Exception { List<Thread> threads = new ArrayList<Thread>(); final AtomicBoolean stop = new AtomicBoolean(); final Set<Revision> set = Collections .synchronizedSet(new HashSet<Revision>()); final Revision[] duplicate = new Revision[1]; for (int i = 0; i < 20; i++) { Thread thread = new Thread(new Runnable() { @Override public void run() { Revision[] last = new Revision[1024]; while (!stop.get()) { for (Revision r : last) { set.remove(r); } for (int i = 0; i < last.length; i++) { last[i] = Revision.newRevision(1); } for (Revision r : last) { if (!set.add(r)) { duplicate[0] = r; } } } } }); thread.start(); threads.add(thread); } Thread.sleep(200); stop.set(true); for (Thread t : threads) { t.join(); } assertNull("Duplicate revision", duplicate[0]); }
@Test public void uniqueRevision() throws Exception { final BlockingQueue<Revision> revisionQueue = Queues.newLinkedBlockingQueue(); int noOfThreads = 60; final int noOfLoops = 1000; List<Thread> workers = new ArrayList<Thread>(); final AtomicBoolean stop = new AtomicBoolean(); final CountDownLatch startLatch = new CountDownLatch(1); final CountDownLatch stopLatch = new CountDownLatch(noOfThreads); for (int i = 0; i < noOfThreads; i++) { workers.add(new Thread(new Runnable() { @Override public void run() { Uninterruptibles.awaitUninterruptibly(startLatch); for (int j = 0; j < noOfLoops && !stop.get(); j++) { revisionQueue.add(Revision.newRevision(1)); } stopLatch.countDown(); } })); } final List<Revision> duplicates = Lists.newArrayList(); final Set<Revision> seenRevs = Sets.newHashSet(); workers.add(new Thread(new Runnable() { @Override public void run() { startLatch.countDown(); while (!stop.get()) { List<Revision> revs = Lists.newArrayList(); Queues.drainUninterruptibly(revisionQueue, revs, 5, 100, TimeUnit.MILLISECONDS); record(revs); } List<Revision> revs = Lists.newArrayList(); revisionQueue.drainTo(revs); record(revs); } private void record(List<Revision> revs) { for (Revision rev : revs) { if (!seenRevs.add(rev)) { duplicates.add(rev); } } if (!duplicates.isEmpty()) { stop.set(true); } } })); for (Thread t : workers) { t.start(); } stopLatch.await(); stop.set(true); for (Thread t : workers) { t.join(); } assertTrue(String.format("Duplicate rev seen %s %n Seen %s", duplicates, seenRevs), duplicates.isEmpty()); }