Better implementation of AttributeMap and also add hasAttr(...). See [n…

…etty#2439]

Motivation:
The old DefaultAttributeMap impl did more synchronization then needed and also did not expose a efficient way to check if an attribute exists with a specific key.

Modifications:
* Rewrite DefaultAttributeMap to not use IdentityHashMap and synchronization on the map directly. The new impl uses a combination of AtomicReferenceArray and synchronization per chain (linked-list). Also access the first Attribute per bucket can be done without any synchronization at all and just uses atomic operations. This should fit for most use-cases pretty weel.
* Add hasAttr(...) implementation

Result:
It's now possible to check for the existence of a attribute without create one. Synchronization is per linked-list and the first entry can even be added via atomic operation.

common/src/main/java/io/netty/util/AttributeMap.java

@@ -26,4 +26,9 @@ public interface AttributeMap {
      * an {@link Attribute} which does not have a value set yet.
      */
     <T> Attribute<T> attr(AttributeKey<T> key);
+
+    /**
+     * Returns {@code} true if and only if the given {@link Attribute} exists in this {@link AttributeMap}.
+     */
+    <T> boolean hasAttr(AttributeKey<T> key);
 }

common/src/main/java/io/netty/util/DefaultAttributeMap.java

@@ -17,65 +17,147 @@
 
 import io.netty.util.internal.PlatformDependent;
 
-import java.util.IdentityHashMap;
-import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.atomic.AtomicReferenceArray;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
 /**
- * Default {@link AttributeMap} implementation which use simple synchronization to keep the memory overhead
+ * Default {@link AttributeMap} implementation which use simple synchronization per bucket to keep the memory overhead
  * as low as possible.
  */
 public class DefaultAttributeMap implements AttributeMap {
 
     @SuppressWarnings("rawtypes")
-    private static final AtomicReferenceFieldUpdater<DefaultAttributeMap, Map> updater;
+    private static final AtomicReferenceFieldUpdater<DefaultAttributeMap, AtomicReferenceArray> updater;
 
     static {
         @SuppressWarnings("rawtypes")
-        AtomicReferenceFieldUpdater<DefaultAttributeMap, Map> referenceFieldUpdater =
-                PlatformDependent.newAtomicReferenceFieldUpdater(DefaultAttributeMap.class, "map");
+        AtomicReferenceFieldUpdater<DefaultAttributeMap, AtomicReferenceArray> referenceFieldUpdater =
+                PlatformDependent.newAtomicReferenceFieldUpdater(DefaultAttributeMap.class, "attributes");
         if (referenceFieldUpdater == null) {
-            referenceFieldUpdater = AtomicReferenceFieldUpdater.newUpdater(DefaultAttributeMap.class, Map.class, "map");
+            referenceFieldUpdater = AtomicReferenceFieldUpdater
+                            .newUpdater(DefaultAttributeMap.class, AtomicReferenceArray.class, "attributes");
         }
         updater = referenceFieldUpdater;
     }
 
+    private static final int BUCKET_SIZE = 4;
+    private static final int MASK = BUCKET_SIZE  - 1;
+
     // Initialize lazily to reduce memory consumption; updated by AtomicReferenceFieldUpdater above.
     @SuppressWarnings("UnusedDeclaration")
-    private volatile Map<AttributeKey<?>, Attribute<?>> map;
+    private volatile AtomicReferenceArray<DefaultAttribute<?>> attributes;
 
+    @SuppressWarnings({ "unchecked", "rawtypes" })
     @Override
     public <T> Attribute<T> attr(AttributeKey<T> key) {
-        Map<AttributeKey<?>, Attribute<?>> map = this.map;
-        if (map == null) {
+        if (key == null) {
+            throw new NullPointerException("key");
+        }
+        AtomicReferenceArray<DefaultAttribute<?>> attributes = this.attributes;
+        if (attributes == null) {
             // Not using ConcurrentHashMap due to high memory consumption.
-            map = new IdentityHashMap<AttributeKey<?>, Attribute<?>>(2);
-            if (!updater.compareAndSet(this, null, map)) {
-                map = this.map;
+            attributes = new AtomicReferenceArray<DefaultAttribute<?>>(BUCKET_SIZE);
+
+            if (!updater.compareAndSet(this, null, attributes)) {
+                attributes = this.attributes;
+            }
+        }
+
+        int i = index(key);
+        DefaultAttribute<?> head = attributes.get(i);
+        if (head == null) {
+            // No head exists yet which means we may be able to add the attribute without synchronization and just
+            // use compare and set. At worst we need to fallback to synchronization
+            head = new DefaultAttribute(key);
+            if (attributes.compareAndSet(i, null, head)) {
+                // we were able to add it so return the head right away
+                return (Attribute<T>) head;
+            } else {
+                head = attributes.get(i);
             }
         }
 
-        synchronized (map) {
-            @SuppressWarnings("unchecked")
-            Attribute<T> attr = (Attribute<T>) map.get(key);
-            if (attr == null) {
-                attr = new DefaultAttribute<T>(map, key);
-                map.put(key, attr);
+        synchronized (head) {
+            DefaultAttribute<?> curr = head;
+            for (;;) {
+                if (!curr.removed && curr.key == key) {
+                    return (Attribute<T>) curr;
+                }
+
+                DefaultAttribute<?> next = curr.next;
+                if (next == null) {
+                    DefaultAttribute<T> attr = new DefaultAttribute<T>(head, key);
+                    curr.next =  attr;
+                    attr.prev = curr;
+                }
+            }
+        }
+    }
+
+    @Override
+    public <T> boolean hasAttr(AttributeKey<T> key) {
+        if (key == null) {
+            throw new NullPointerException("key");
+        }
+        AtomicReferenceArray<DefaultAttribute<?>> attributes = this.attributes;
+        if (attributes == null) {
+            // no attribute exists
+            return false;
+        }
+
+        int i = index(key);
+        DefaultAttribute<?> head = attributes.get(i);
+        if (head == null) {
+            // No attribute exists which point to the bucket in which the head should be located
+            return false;
+        }
+
+        // check on the head can be done without synchronization
+        if (head.key == key && !head.removed) {
+            return true;
+        }
+
+        synchronized (head) {
+            // we need to synchronize on the head
+            DefaultAttribute<?> curr = head.next;
+            while (curr != null) {
+                if (!curr.removed && curr.key == key) {
+                    return true;
+                }
+                curr = curr.next;
             }
-            return attr;
+            return false;
         }
     }
 
+    private static int index(AttributeKey<?> key) {
+        return key.id() & MASK;
+    }
+
+    @SuppressWarnings("serial")
     private static final class DefaultAttribute<T> extends AtomicReference<T> implements Attribute<T> {
 
         private static final long serialVersionUID = -2661411462200283011L;
 
-        private final Map<AttributeKey<?>, Attribute<?>> map;
+        // The head of the linked-list this attribute belongs to, which may be itself
+        private final DefaultAttribute<?> head;
         private final AttributeKey<T> key;
 
-        DefaultAttribute(Map<AttributeKey<?>, Attribute<?>> map, AttributeKey<T> key) {
-            this.map = map;
+        // Double-linked list to prev and next node to allow fast removal
+        private DefaultAttribute<?> prev;
+        private DefaultAttribute<?> next;
+
+        // Will be set to true one the attribute is removed via getAndRemove() or remove()
+        private volatile boolean removed;
+
+        DefaultAttribute(DefaultAttribute<?> head, AttributeKey<T> key) {
+            this.head = head;
+            this.key = key;
+        }
+
+        DefaultAttribute(AttributeKey<T> key) {
+            head = this;
             this.key = key;
         }
 
@@ -97,20 +179,33 @@ public T setIfAbsent(T value) {
 
         @Override
         public T getAndRemove() {
+            removed = true;
             T oldValue = getAndSet(null);
             remove0();
             return oldValue;
         }
 
         @Override
         public void remove() {
+            removed = true;
             set(null);
             remove0();
         }
 
         private void remove0() {
-            synchronized (map) {
-                map.remove(key);
+            synchronized (head) {
+                // We only update the linked-list structure if prev != null because if it is null this
+                // DefaultAttribute acts also as head. The head must never be removed completely and just be
+                // marked as removed as all synchronization is done on the head itself for each bucket.
+                // The head itself will be GC'ed once the DefaultAttributeMap is GC'ed. So at most 5 heads will
+                // be removed lazy as the array size is 5.
+                if (prev != null) {
+                    prev.next = next;
+
+                    if (next != null) {
+                        next.prev = prev;
+                    }
+                }
             }
         }
     }

transport/src/main/java/io/netty/channel/DefaultChannelHandlerContext.java

@@ -137,6 +137,11 @@ public <T> Attribute<T> attr(AttributeKey<T> key) {
         return channel.attr(key);
     }
 
+    @Override
+    public <T> boolean hasAttr(AttributeKey<T> key) {
+        return channel.hasAttr(key);
+    }
+
     @Override
     public ChannelHandlerContext fireChannelRegistered() {
         DefaultChannelHandlerContext next = findContextInbound();