ConcurrentHapMap size not as expected when put done using Multiple Threads

[Vivek Kr Singh]

I wrote a piece of code to use java.util.concurrent.CountDownLatch. In this code 5 threads are being created which insert 5000 values each in a java.util.concurrent.ConcurrentHashMap. Before calculating the final map size latch should be decremented to 0.

ConcurrentHashMap size is not 25000 as expected. The value is lesser than 25000 and it is different on each run, which does suggest that size() is invoked on map when it is under modification. Any help would be appreciated.

package org.vivek.java.util;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

public class ThreadLocalSample {

	// Values will be added to map by all threads
	private ConcurrentHashMap<Integer, String> concurrentHashMap = new ConcurrentHashMap<Integer, String>(
			5000, 0.5f, 5);

	private static final int THREAD_COUNT = 5;

	// Using latch to check ConcurrentHashMap after all threads have exited
	private CountDownLatch latch = new CountDownLatch(5);

	// Key used in ConcurrentHashMap
	private volatile int index = 0;

	// Counter local to each thread
	private ThreadLocal<Integer> counter = new ThreadLocal<Integer>() {
		@Override
		protected Integer initialValue() {
			return 0;
		}
	};

	// Runnable implementation
	private Runnable runnable = new Runnable() {

		/**
		 * Add key value pairs to HashMap and count down on lock
		 * 
		 * @see java.lang.Runnable#run()
		 */
		@Override
		public void run() {

			int i = 0;

			while (i++ < 5000) {
				concurrentHashMap.put(index++, Thread.currentThread().getName()
						+ " " + counter.get());
			}
			// Decrement latch as work completed
			latch.countDown();
		}
	};

	/**
	 * Helper Method
	 */
	private void go() {

		// Start threads in a loop
		for (int i = 0; i < THREAD_COUNT; i++) {
			Thread thread = new Thread(runnable);
			thread.start();
		}

		try {
			// Wait till all threads exit
			latch.await();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}

		// Now check Map Size
		if (concurrentHashMap.size() != 25000) {
			System.out.println("Issue with HashMap Size. Current size is  "
					+ concurrentHashMap.size() + " Index is at " + index);
		} else {
			System.out.println("Latch Working fine");
		}
	}

	public static void main(String... args) {
		new ThreadLocalSample().go();
	}

}

Edit : Changed first parameter in map constructor to 5000

[Ernest Friedman-Hill]

I think the major issue here is the assumption that "i++" is atomic -- it is not. The i++ operation involves a fetch, an increment, and a store, three separate steps. All five threads are trying to fetch and postincrement the same int, and you're going to have some collisions, resulting in some missed increments and map elements being overwritten.

[Henry Wong]

Vivek Kr Singh wrote:ConcurrentHashMap size is not 25000 as expected. The value is lesser than 25000 and it is different on each run, which does suggest that size() is invoked on map when it is under modification.

Not sure what you mean by "size() is invoked on map when it is under modification" -- can you elaborate a bit?

Vivek Kr Singh wrote:
Any help would be appreciated.

I believe the issue (at least the one that I see) is here...

while (i++ < 5000) {
    concurrentHashMap.put(index++, Thread.currentThread().getName()
                       + " " + counter.get());
}

The index variable may be volatile, but that doesn't mean that all operations are thread safe. It just means that atomic operations are thread safe -- as the JVM will not cache the values.

An increment is *not* an atomic operation in Java. It is actually a load, increment in register, and store back to memory. Because of this, you may do a few parallel increments, and since a map don't allow duplicates, lose a few items.

Henry

[Vivek Kr Singh]

Thanks very much. I was using index marked as volatile which only guarantees synchronized read.
I changed index type to java.util.concurrent.atomic.AtomicInteger

// Key used in ConcurrentHashMap
private AtomicInteger atomicIndex = new AtomicInteger(0);

Also i updated put operation on ConcurrentHashMap.

concurrentHashMap.put(atomicIndex.incrementAndGet(), Thread.currentThread().getName()+ " " + counter.get());

This fixed the issue. Now this code is 100% Concurrency safe or is there any pitfall remaining ?

[Henry Wong]

Ernest Friedman-Hill wrote:I think the major issue here is the assumption that "i++" is atomic -- it is not.

Ernest,

I think you mean "index++", as the "i" variable is a local variable.

Henry

[Vivek Kr Singh]

Henry Wong wrote:
Not sure what you mean by "size() is invoked on map when it is under modification" -- can you elaborate a bit?

Wrong Assumption. I was assuming ConcurrentHashMap operations are delayed.

[Henry Wong]

Also i updated put operation on ConcurrentHashMap.

BTW, not that its an issue, since this is only a test program, but...

If you want "index++" functionality, then you should use the atomicIndex.getAndIncrement() method. The incrementAndGet() method implements an atomic pre increment operation.

Henry

[Ernest Friedman-Hill]

Henry Wong wrote:

Ernest Friedman-Hill wrote:I think the major issue here is the assumption that "i++" is atomic -- it is not.

Ernest,

I think you mean "index++", as the "i" variable is a local variable.

Henry

Yeah. Do I still get points for answering first?

[Henry Wong]

Ernest Friedman-Hill wrote:
Yeah. Do I still get points for answering first?

I guess... but... since your lead is insurmountable, I don't keep track...

Henry

[Vivek Kr Singh]

Henry Wong wrote:
If you want "index++" functionality, then you should use the atomicIndex.getAndIncrement() method. The incrementAndGet() method implements an atomic pre increment operation.

I already updated that