nlm: use ConcurrentHashMap to store the locks

Though access to the lock storage serialized for the same file, the
Multimap still throws ConcurrentModificationException.

Use ConcurrentHashMap to store locks as it gives a better throughput:

// Multimaps#synchronizedListMultimap
Result "org.dcache.nfs.v4.nlm.SimpleLmBenchmark.benchmarkConcurrentLocking":
  437057.460 ±(99.9%) 18574.458 ops/s [Average]
  (min, avg, max) = (372606.415, 437057.460, 467748.621), stdev = 24796.370
  CI (99.9%): [418483.001, 455631.918] (assumes normal distribution)

// ConcurrentHashMap
Result "org.dcache.nfs.v4.nlm.SimpleLmBenchmark.benchmarkConcurrentLocking":
  744594.337 ±(99.9%) 45860.571 ops/s [Average]
  (min, avg, max) = (592325.817, 744594.337, 847798.786), stdev = 61222.550
  CI (99.9%): [698733.766, 790454.908] (assumes normal distribution)

Fixes: #70
Acked-by: Paul Millar
Target: master, 0.17

Author: kofemann

core/src/main/java/org/dcache/nfs/v4/nlm/SimpleLm.java

@@ -19,11 +19,13 @@
  */
 package org.dcache.nfs.v4.nlm;
 
-import com.google.common.collect.ArrayListMultimap;
-import com.google.common.collect.Multimap;
 import com.google.common.io.BaseEncoding;
 import com.google.common.util.concurrent.Striped;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.locks.Lock;
 
 /**
@@ -48,7 +50,7 @@ public class SimpleLm extends AbstractLockManager {
     /**
      * Exclusive lock on objects locks.
      */
-    private final Multimap<String, NlmLock> locks = ArrayListMultimap.create();
+    private final ConcurrentHashMap<String, List<NlmLock>> locks = new ConcurrentHashMap<>();
 
     @Override
     protected Lock getObjectLock(byte[] objId) {
@@ -59,31 +61,47 @@ protected Lock getObjectLock(byte[] objId) {
     @Override
     protected Collection<NlmLock> getActiveLocks(byte[] objId) {
         String key = toKey(objId);
-        return locks.get(key);
+        return locks.getOrDefault(key, Collections.emptyList());
     }
 
     @Override
     protected void add(byte[] objId, NlmLock lock) {
         String key = toKey(objId);
-        locks.put(key, lock);
+        Collection<NlmLock> l = locks.computeIfAbsent(key, k -> new ArrayList<>());
+        l.add(lock);
     }
 
     @Override
     protected boolean remove(byte[] objId, NlmLock lock) {
         String key = toKey(objId);
-        return locks.remove(key, lock);
+        Collection<NlmLock> l = locks.get(key);
+        boolean isRemoved = false;
+        if (l != null) {
+            isRemoved = l.remove(lock);
+            if (l.isEmpty()) {
+                locks.remove(key);
+            }
+        }
+        return isRemoved;
     }
 
     @Override
     protected void addAll(byte[] objId, Collection<NlmLock> locks) {
         String key = toKey(objId);
-        locks.forEach(l -> this.locks.put(key, l));
+        Collection<NlmLock> l = this.locks.computeIfAbsent(key, k -> new ArrayList<>());
+        l.addAll(locks);
     }
 
     @Override
     protected void removeAll(byte[] objId, Collection<NlmLock> locks) {
         String key = toKey(objId);
-        locks.forEach(l -> this.locks.remove(key, l));
+        Collection<NlmLock> l = this.locks.get(key);
+        if (l != null) {
+            l.removeAll(locks);
+            if (l.isEmpty()) {
+                this.locks.remove(key);
+            }
+        }
     }
 
     private final String toKey(byte[] objId) {