Better implementation of AttributeMap and also add hasAttr(...). See [#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.
This commit is contained in:
parent
7bf7c7b16a
commit
c3bd7a8ff1
@ -26,4 +26,9 @@ public interface AttributeMap {
|
|||||||
* an {@link Attribute} which does not have a value set yet.
|
* an {@link Attribute} which does not have a value set yet.
|
||||||
*/
|
*/
|
||||||
<T> Attribute<T> attr(AttributeKey<T> key);
|
<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);
|
||||||
}
|
}
|
||||||
|
@ -17,65 +17,147 @@ package io.netty.util;
|
|||||||
|
|
||||||
import io.netty.util.internal.PlatformDependent;
|
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.AtomicReference;
|
||||||
|
import java.util.concurrent.atomic.AtomicReferenceArray;
|
||||||
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
|
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.
|
* as low as possible.
|
||||||
*/
|
*/
|
||||||
public class DefaultAttributeMap implements AttributeMap {
|
public class DefaultAttributeMap implements AttributeMap {
|
||||||
|
|
||||||
@SuppressWarnings("rawtypes")
|
@SuppressWarnings("rawtypes")
|
||||||
private static final AtomicReferenceFieldUpdater<DefaultAttributeMap, Map> updater;
|
private static final AtomicReferenceFieldUpdater<DefaultAttributeMap, AtomicReferenceArray> updater;
|
||||||
|
|
||||||
static {
|
static {
|
||||||
@SuppressWarnings("rawtypes")
|
@SuppressWarnings("rawtypes")
|
||||||
AtomicReferenceFieldUpdater<DefaultAttributeMap, Map> referenceFieldUpdater =
|
AtomicReferenceFieldUpdater<DefaultAttributeMap, AtomicReferenceArray> referenceFieldUpdater =
|
||||||
PlatformDependent.newAtomicReferenceFieldUpdater(DefaultAttributeMap.class, "map");
|
PlatformDependent.newAtomicReferenceFieldUpdater(DefaultAttributeMap.class, "attributes");
|
||||||
if (referenceFieldUpdater == null) {
|
if (referenceFieldUpdater == null) {
|
||||||
referenceFieldUpdater = AtomicReferenceFieldUpdater.newUpdater(DefaultAttributeMap.class, Map.class, "map");
|
referenceFieldUpdater = AtomicReferenceFieldUpdater
|
||||||
|
.newUpdater(DefaultAttributeMap.class, AtomicReferenceArray.class, "attributes");
|
||||||
}
|
}
|
||||||
updater = referenceFieldUpdater;
|
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.
|
// Initialize lazily to reduce memory consumption; updated by AtomicReferenceFieldUpdater above.
|
||||||
@SuppressWarnings("UnusedDeclaration")
|
@SuppressWarnings("UnusedDeclaration")
|
||||||
private volatile Map<AttributeKey<?>, Attribute<?>> map;
|
private volatile AtomicReferenceArray<DefaultAttribute<?>> attributes;
|
||||||
|
|
||||||
|
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||||
@Override
|
@Override
|
||||||
public <T> Attribute<T> attr(AttributeKey<T> key) {
|
public <T> Attribute<T> attr(AttributeKey<T> key) {
|
||||||
Map<AttributeKey<?>, Attribute<?>> map = this.map;
|
if (key == null) {
|
||||||
if (map == null) {
|
throw new NullPointerException("key");
|
||||||
|
}
|
||||||
|
AtomicReferenceArray<DefaultAttribute<?>> attributes = this.attributes;
|
||||||
|
if (attributes == null) {
|
||||||
// Not using ConcurrentHashMap due to high memory consumption.
|
// Not using ConcurrentHashMap due to high memory consumption.
|
||||||
map = new IdentityHashMap<AttributeKey<?>, Attribute<?>>(2);
|
attributes = new AtomicReferenceArray<DefaultAttribute<?>>(BUCKET_SIZE);
|
||||||
if (!updater.compareAndSet(this, null, map)) {
|
|
||||||
map = this.map;
|
if (!updater.compareAndSet(this, null, attributes)) {
|
||||||
|
attributes = this.attributes;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
synchronized (map) {
|
int i = index(key);
|
||||||
@SuppressWarnings("unchecked")
|
DefaultAttribute<?> head = attributes.get(i);
|
||||||
Attribute<T> attr = (Attribute<T>) map.get(key);
|
if (head == null) {
|
||||||
if (attr == null) {
|
// No head exists yet which means we may be able to add the attribute without synchronization and just
|
||||||
attr = new DefaultAttribute<T>(map, key);
|
// use compare and set. At worst we need to fallback to synchronization
|
||||||
map.put(key, attr);
|
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 (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;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
return attr;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@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 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 class DefaultAttribute<T> extends AtomicReference<T> implements Attribute<T> {
|
||||||
|
|
||||||
private static final long serialVersionUID = -2661411462200283011L;
|
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;
|
private final AttributeKey<T> key;
|
||||||
|
|
||||||
DefaultAttribute(Map<AttributeKey<?>, Attribute<?>> map, AttributeKey<T> key) {
|
// Double-linked list to prev and next node to allow fast removal
|
||||||
this.map = map;
|
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;
|
this.key = key;
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -97,6 +179,7 @@ public class DefaultAttributeMap implements AttributeMap {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public T getAndRemove() {
|
public T getAndRemove() {
|
||||||
|
removed = true;
|
||||||
T oldValue = getAndSet(null);
|
T oldValue = getAndSet(null);
|
||||||
remove0();
|
remove0();
|
||||||
return oldValue;
|
return oldValue;
|
||||||
@ -104,13 +187,25 @@ public class DefaultAttributeMap implements AttributeMap {
|
|||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void remove() {
|
public void remove() {
|
||||||
|
removed = true;
|
||||||
set(null);
|
set(null);
|
||||||
remove0();
|
remove0();
|
||||||
}
|
}
|
||||||
|
|
||||||
private void remove0() {
|
private void remove0() {
|
||||||
synchronized (map) {
|
synchronized (head) {
|
||||||
map.remove(key);
|
// 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;
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
@ -298,6 +298,11 @@ final class DefaultChannelHandlerContext implements ChannelHandlerContext, Resou
|
|||||||
return channel.attr(key);
|
return channel.attr(key);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public <T> boolean hasAttr(AttributeKey<T> key) {
|
||||||
|
return channel.hasAttr(key);
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public ChannelHandlerContext fireChannelRegistered() {
|
public ChannelHandlerContext fireChannelRegistered() {
|
||||||
DefaultChannelHandlerContext next = findContextInbound(MASK_CHANNEL_REGISTERED);
|
DefaultChannelHandlerContext next = findContextInbound(MASK_CHANNEL_REGISTERED);
|
||||||
|
Loading…
Reference in New Issue
Block a user