Java – what’s wrong with this thread safe byte sequence generator?
I need a byte generator that will generate from byte MIN_ Value to byte MAX_ Value When it reaches Max_ Value, it should start from min_ Value restart
I wrote the code using atomicinteger (see below); However, if accessed at the same time and if thread Sleep () artificially slows down the code, so the code doesn't seem to work properly (if there's no sleep, it works normally; however, I suspect it's too fast for concurrency problems)
Code (added some debugging code):
public class ByteGenerator { private static final int INITIAL_VALUE = Byte.MIN_VALUE-1; private AtomicInteger counter = new AtomicInteger(INITIAL_VALUE); private AtomicInteger resetCounter = new AtomicInteger(0); private boolean isSlow = false; private long startTime; public byte nextValue() { int next = counter.incrementAndGet(); //if (isSlow) slowDown(5); if (next > Byte.MAX_VALUE) { synchronized(counter) { int i = counter.get(); //if value is still larger than max byte value,we reset it if (i > Byte.MAX_VALUE) { counter.set(INITIAL_VALUE); resetCounter.incrementAndGet(); if (isSlow) slowDownAndLog(10,"resetting"); } else { if (isSlow) slowDownAndLog(1,"missed"); } next = counter.incrementAndGet(); } } return (byte) next; } private void slowDown(long millis) { try { Thread.sleep(millis); } catch (InterruptedException e) { } } private void slowDownAndLog(long millis,String msg) { slowDown(millis); System.out.println(resetCounter + " " + (System.currentTimeMillis()-startTime) + " " + Thread.currentThread().getName() + ": " + msg); } public void setSlow(boolean isSlow) { this.isSlow = isSlow; } public void setStartTime(long startTime) { this.startTime = startTime; } }
And, test:
public class ByteGeneratorTest { @Test public void testGenerate() throws Exception { ByteGenerator g = new ByteGenerator(); for (int n = 0; n < 10; n++) { for (int i = Byte.MIN_VALUE; i <= Byte.MAX_VALUE; i++) { assertEquals(i,g.nextValue()); } } } @Test public void testGenerateMultiThreaded() throws Exception { final ByteGenerator g = new ByteGenerator(); g.setSlow(true); final AtomicInteger[] counters = new AtomicInteger[Byte.MAX_VALUE-Byte.MIN_VALUE+1]; for (int i = 0; i < counters.length; i++) { counters[i] = new AtomicInteger(0); } Thread[] threads = new Thread[100]; final CountDownLatch latch = new CountDownLatch(threads.length); for (int i = 0; i < threads.length; i++) { threads[i] = new Thread(new Runnable() { public void run() { try { for (int i = Byte.MIN_VALUE; i <= Byte.MAX_VALUE; i++) { byte value = g.nextValue(); counters[value-Byte.MIN_VALUE].incrementAndGet(); } } finally { latch.countDown(); } } },"generator-client-" + i); threads[i].setDaemon(true); } g.setStartTime(System.currentTimeMillis()); for (int i = 0; i < threads.length; i++) { threads[i].start(); } latch.await(); for (int i = 0; i < counters.length; i++) { System.out.println("value #" + (i+Byte.MIN_VALUE) + ": " + counters[i].get()); } //print out the number of hits for each value for (int i = 0; i < counters.length; i++) { assertEquals("value #" + (i+Byte.MIN_VALUE),threads.length,counters[i].get()); } } }
The result on my 2-core machine is that the value #-128 gets 146 hits (all of which should get 100 hits because we have 100 threads)
If anyone has any idea, what's wrong with this code, I'm all ears / eyes
Update: for those who are in a hurry and don't want to scroll down, the right (and shortest and most elegant) way to solve this problem in Java will be as follows:
public byte nextValue() { return (byte) counter.incrementAndGet(); }
Thank you, Heinz!
Solution
According to counter The old value of get() determines incrementandget() Before incrementandget() operation on the counter, the value of the counter can reach Max again_ VALUE.
if (next > Byte.MAX_VALUE) { synchronized(counter) { int i = counter.get(); //here You make sure the the counter is not over the MAX_VALUE if (i > Byte.MAX_VALUE) { counter.set(INITIAL_VALUE); resetCounter.incrementAndGet(); if (isSlow) slowDownAndLog(10,"resetting"); } else { if (isSlow) slowDownAndLog(1,"missed"); //the counter can reach MAX_VALUE again if you wait here long enough } next = counter.incrementAndGet(); //here you increment on return the counter that can reach >MAX_VALUE in the meantime } }
In order for it to work, it must be ensured that decisions are not made on stale information Reset counter or return old value
public byte nextValue() { int next = counter.incrementAndGet(); if (next > Byte.MAX_VALUE) { synchronized(counter) { next = counter.incrementAndGet(); //if value is still larger than max byte value,we reset it if (next > Byte.MAX_VALUE) { counter.set(INITIAL_VALUE + 1); next = INITIAL_VALUE + 1; resetCounter.incrementAndGet(); if (isSlow) slowDownAndLog(10,"resetting"); } else { if (isSlow) slowDownAndLog(1,"missed"); } } } return (byte) next; }