Java Concurrency

Question

I'm trying to implement some kind of accumulation logic in a multi threaded environment; I’m wondering is there any better/faster way to do it without the lock and synchronized keyword? The following is my current code:

public class ConcurrentHashMapTest {

    private static final int NB_THREADS = 1_000;

    private final Map<String, Integer> cnts = new HashMap<>();

    private static final Lock RWLOCK = new ReentrantLock(true);

    private static final String[] NAMES = {
        "A", "B"
    };

    public void testIt() {
        ExecutorService executor =
            Executors.newFixedThreadPool(NB_THREADS);
        for (int i = 0; i < NB_THREADS; i++) {
            Runnable task = new WorkerThread();
            executor.submit(task);
        }
        executor.shutdown();
        try {
            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println(cnts);
    }

    private void accumulate(String name) {
        RWLOCK.lock();
        try {
            Integer cnt = cnts.get(name);
            if (cnt == null) {
                cnts.put(name, 1);
            } else {
                cnts.put(name, cnt + 1);
            }
        } finally {
            RWLOCK.unlock();
        }
    }

    private class WorkerThread implements Runnable {
        @Override
        public void run() {
            accumulate(NAMES[ThreadLocalRandom.current().nextInt(0, NAMES.length)]);
        }
    }
}

You might want to have a look at guava multiset : code.google.com/p/guava-libraries/wiki/… — benzonico
– benzonico, Commented May 6, 2014 at 14:20
Define "better". About "faster", do not optimize prematurely. Only if you see it is the real bottleneck of your program performance, then think how to do it faster. — m0skit0
– m0skit0, Commented May 6, 2014 at 14:21
I understand that, each thread will only call accumulate method, on which you have taken lock. It kills multi threading. — Saurabh Batra
– Saurabh Batra, Commented May 6, 2014 at 14:22
Your testIt method is not thread-safe. It's not enough that cnts is protected by the RWLOCK during modifications; it must also be protected during all reads, including the System.out.println(cnts) in testIt. That is, you need to lock, print cnts, then unlock. — yshavit
– yshavit, Commented May 6, 2014 at 14:48
Doesn't it only print cnts after all worker threads have terminated? Actually, it looks like that's the intent, but it could do it then, or after a timeout (if they haven't finished); the latter could be a problem. — David Conrad
– David Conrad, Commented May 6, 2014 at 15:02

David Conrad · Accepted Answer · 2014-05-06 14:52:05Z

3

Java 8:

private final Map<String, AtomicInteger> cnts =
        new ConcurrentHashMap<>();

private void accumulate(String name) {
    cnts.computeIfAbsent(name, k -> new AtomicInteger()).incrementAndGet();
}

The ConcurrentHashMap can be freely accessed from multiple threads. The computeIfAbsent method takes a lambda to evaluate to get a value for the key if the key is not present in the map, and adds it if and only if there is no such mapping, and then returns that value. It's effectively putIfAbsent followed by get. The value is a new AtomicInteger with the value 0. Whether there was an existing value, or whether a new one with value 0 was just added, in either case increment it.

Java 7:

private final ConcurrentMap<String, AtomicInteger> cnts =
        new ConcurrentHashMap<>();

private void accumulate(String name) {
    cnts.putIfAbsent(name, new AtomicInteger());
    cnts.get(name).incrementAndGet();
}

For Java 7, there is no computeIfAbsent method, but that effectively just does a putIfAbsent followed by a get, so the same effect is achieved by calling those methods. There is no concern that the value already existed in the map; a new, zero AtomicInteger is added if and only if the map had no value for that key. Even if another thread got in there before us and added a zero, both threads would then see and increment that same AtomicInteger instance.

edited May 6, 2014 at 14:52

answered May 6, 2014 at 14:44

David Conrad

16.5k3 gold badges49 silver badges60 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

user3502676 Over a year ago

Your suggestion really neat and now I can remove the try and finally block.

akhil_mittal Over a year ago

Another improvement that can be considered is: Java 8: private final Map<String, LongAdder> countsMap = new HashMap<>(); And in accumulate method we have: countsMap.computeIfAbsent(name, k -> new LongAdder()).increment(); The expected throughput of this class is significantly higher under high contention, at the expense of higher space consumption.

Nazgul · Accepted Answer · 2014-05-06 14:25:30Z

1

use a concurrent hash map with String and AtomicInteger. Both are thread safe and thus can be used freely.

answered May 6, 2014 at 14:25

Nazgul

1,9021 gold badge11 silver badges15 bronze badges

Comments

user1717259 · Accepted Answer · 2014-05-06 14:41:30Z

I'd be wary of using fairness on your ReentrantLock in this case, as there's no benefit to your accumulator if longer waiting threads get access first. Take a look at Brian Goetz's 'Java Concurrency in Practice'

Why wouldn't we want to make all locks fair? After all, fairness is good and unfairness is bad, right? (It's not accidental that whenever kids want to appeal a decision, "that's not fair" almost certainly comes up. We think fairness is pretty important, and they know it.) In reality, the fairness guarantee for locks is a very strong one, and comes at a significant performance cost. The bookkeeping and synchronization required to ensure fairness mean that contended fair locks will have much lower throughput than unfair locks. As a default, you should set fair to false unless it is critical to the correctness of your algorithm that threads be serviced in exactly the order they queued up.

Tim B · Accepted Answer · 2014-05-06 14:22:03Z

0

You could use a Map of name to AtomicInteger and use double-check locking when there is no counter in the map at all. Be aware that you need to use the volatile keyword for effective double-check locking.

This way you will only lock the whole map for actually adding brand new entries, the rest of the processing can happen in parallel.

You risk massively over-complicating your program here though and possibly even reducing performance in real-world cases. Is contention on this map really a performance bottle-neck?

answered May 6, 2014 at 14:22

Tim B

41.3k16 gold badges88 silver badges132 bronze badges

1 Comment

Bill Michell Over a year ago

Or even a ConcurrentHashMap of AtomicIntegers, and use the putIfAbsent() method to create new zeroed entries in the map

Cédric K. · Accepted Answer · 2014-05-06 14:49:51Z

0

According to Oracle Java 7 API : implementation of HashMap is not synchronized.

You can use Hashtable implementation or declare : private final Map<String, Integer> cnts = Collections.synchronizedMap(new HashMap<String, Integer>());

answered May 6, 2014 at 14:49

Cédric K.

1341 gold badge1 silver badge4 bronze badges

1 Comment

David Conrad Over a year ago

Synchronization doesn't help here, since you need to A) get the value for a key, if any, and B) put a 1 if it was missing, or C) add 1 to the value and put it back if it was present. Hashtable and synchronizedMap only synchronize each individual step, but don't synchronize between them. (Another thread could add a value in between the steps.) This is exactly why Hashtable was deprecated, and why the OP is using a ReentrantLock to begin with.

OldCurmudgeon · Accepted Answer · 2014-05-06 15:11:53Z

I think what you are looking for is a Multiton:

/**
 * Holds a thread-safe map of unique create-once items.
 *
 * Contract:
 *
 * Only one object will be made for each key presented.
 *
 * Thread safe.
 *
 * @author OldCurmudgeon
 * @param <K>
 * @param <V>
 */
public class Multiton<K, V> {

    // Map from the key to the futures of the items.
    private final ConcurrentMap<K, Future<V>> multitons = new ConcurrentHashMap<>();
    // The creator can create an item of type V.
    private final Creator<K, V> creator;

    public Multiton(Creator<K, V> creator) {
        this.creator = creator;
    }

    /**
     * There can be only one.
     *
     * Use a FutureTask to do the creation to ensure only one construction.
     *
     * @param key
     * @return
     * @throws InterruptedException
     * @throws ExecutionException
     */
    public V get(final K key) throws InterruptedException, ExecutionException {
        // Already made?
        Future<V> f = multitons.get(key);
        if (f == null) {
            // Plan the future but do not create as yet.
            FutureTask<V> ft = new FutureTask<>(() -> creator.create(key));
            // Store it.
            f = multitons.putIfAbsent(key, ft);
            if (f == null) {
                // It was successfully stored - it is the first (and only)
                f = ft;
                // Make it happen.
                ft.run();
            }
        }
        // Wait for it to finish construction and return the constructed.
        return f.get();
    }

    /**
     * Returns a Map indicating the current state.
     *
     * @return a Map which should reflect the current state.
     *
     * @throws java.lang.InterruptedException
     * @throws java.util.concurrent.ExecutionException
     */
    public Map<K, V> getMap() throws InterruptedException, ExecutionException {
        Map<K, V> map = new HashMap<>();
        for (Map.Entry<K, Future<V>> e : multitons.entrySet()) {
            map.put(e.getKey(), e.getValue().get());
        }
        return map;
    }

    /**
     * User provides one of these to do the construction.
     *
     * @param <K>
     * @param <V>
     */
    public abstract static class Creator<K, V> {

        // Return a new item under the key.
        abstract V create(K key) throws ExecutionException;

    }

}

Usage - for demonstration - adds up all integers up to 999, keying on their first digit:

Multiton<String, AtomicInteger> counts = new Multiton<>(
        new Creator<String, AtomicInteger>() {

            @Override
            AtomicInteger create(String key) throws ExecutionException {
                return new AtomicInteger();
            }
        }
);

public void test() throws InterruptedException, ExecutionException {
    for (int i = 0; i < 1000; i++) {
        counts.get(Integer.toString(i).substring(0, 1)).addAndGet(i);
    }
    System.out.println(counts.getMap());
}

Prints:

{0=0, 1=15096, 2=25197, 3=35298, 4=45399, 5=55500, 6=65601, 7=75702, 8=85803, 9=95904}

Java < 8 version:

/**
 * Holds a thread-safe map of unique create-once items.
 *
 * Contract:
 *
 * Only one object will be made for each key presented.
 *
 * Thread safe.
 *
 * @author OldCurmudgeon
 * @param <K>
 * @param <V>
 */
public class Multiton<K, V> {

  // Map from the key to the futures of the items.
  private final ConcurrentMap<K, Future<V>> multitons = new ConcurrentHashMap<>();
  // The creator can create an item of type V.
  private final Creator<K, V> creator;

  public Multiton(Creator<K, V> creator) {
    this.creator = creator;
  }

  /**
   * There can be only one.
   *
   * Use a FutureTask to do the creation to ensure only one construction.
   *
   * @param key
   * @return
   * @throws InterruptedException
   * @throws ExecutionException
   */
  public V get(final K key) throws InterruptedException, ExecutionException {
    // Already made?
    Future<V> f = multitons.get(key);
    if (f == null) {
      // Plan the future but do not create as yet.
      FutureTask<V> ft = new FutureTask<>(new Callable<V>() {

        @Override
        public V call() throws Exception {
          // Doing this inline may be a little contrived but it maintains the linkage with the Java-8 version.
          return creator.create(key);
        }

      }
      );
      // Store it.
      f = multitons.putIfAbsent(key, ft);
      if (f == null) {
        // It was successfully stored - it is the first (and only)
        f = ft;
        // Make it happen.
        ft.run();
      }
    }
    // Wait for it to finish construction and return the constructed.
    return f.get();
  }

  /**
   * Returns a Map indicating the current state.
   *
   * @return a Map which should reflect the current state.
   *
   * @throws java.lang.InterruptedException
   * @throws java.util.concurrent.ExecutionException
   */
  public Map<K, V> getMap() throws InterruptedException, ExecutionException {
      Map<K, V> map = new HashMap<>();
      for (Map.Entry<K, Future<V>> e : multitons.entrySet()) {
          map.put(e.getKey(), e.getValue().get());
      }
      return map;
  }

  /**
   * User provides one of these to do the construction.
   *
   * @param <K>
   * @param <V>
   */
  public abstract static class Creator<K, V> {

    // Return a new item under the key.
    abstract V create(K key) throws ExecutionException;

  }

}

Collectives™ on Stack Overflow

Java Concurrency

6 Answers 6

2 Comments

Comments

Comments

1 Comment

1 Comment

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

6 Answers 6

2 Comments

Comments

Comments

1 Comment

1 Comment

Comments

Your Answer

Sign up or log in

Post as a guest

Related