[] segments = this.segments;
+ long sum = 0;
+ long check = 0;
+ int[] mc = new int[segments.length];
+ // Try a few times to get accurate count. On failure due to continuous
+ // async changes in table, resort to locking.
+ for (int k = 0; k < RETRIES_BEFORE_LOCK; ++ k) {
+ check = 0;
+ sum = 0;
+ int mcsum = 0;
+ for (int i = 0; i < segments.length; ++ i) {
+ sum += segments[i].count;
+ mcsum += mc[i] = segments[i].modCount;
+ }
+ if (mcsum != 0) {
+ for (int i = 0; i < segments.length; ++ i) {
+ check += segments[i].count;
+ if (mc[i] != segments[i].modCount) {
+ check = -1; // force retry
+ break;
+ }
+ }
+ }
+ if (check == sum) {
+ break;
+ }
+ }
+ if (check != sum) { // Resort to locking all segments
+ sum = 0;
+ for (int i = 0; i < segments.length; ++ i) {
+ segments[i].lock();
+ }
+ for (int i = 0; i < segments.length; ++ i) {
+ sum += segments[i].count;
+ }
+ for (int i = 0; i < segments.length; ++ i) {
+ segments[i].unlock();
+ }
+ }
+ if (sum > Integer.MAX_VALUE) {
+ return Integer.MAX_VALUE;
+ } else {
+ return (int) sum;
+ }
+ }
+
+ /**
+ * Returns the value to which the specified key is mapped, or {@code null}
+ * if this map contains no mapping for the key.
+ *
+ * More formally, if this map contains a mapping from a key {@code k} to
+ * a value {@code v} such that {@code key.equals(k)}, then this method
+ * returns {@code v}; otherwise it returns {@code null}. (There can be at
+ * most one such mapping.)
+ *
+ * @throws NullPointerException if the specified key is null
+ */
+ @Override
+ public V get(Object key) {
+ int hash = hashOf(key);
+ return segmentFor(hash).get(key, hash);
+ }
+
+ /**
+ * Tests if the specified object is a key in this table.
+ *
+ * @param key possible key
+ * @return true if and only if the specified object is a key in
+ * this table, as determined by the equals method;
+ * false otherwise.
+ * @throws NullPointerException if the specified key is null
+ */
+ @Override
+ public boolean containsKey(Object key) {
+ int hash = hashOf(key);
+ return segmentFor(hash).containsKey(key, hash);
+ }
+
+ /**
+ * Returns true if this map maps one or more keys to the specified
+ * value. Note: This method requires a full internal traversal of the hash
+ * table, and so is much slower than method containsKey.
+ *
+ * @param value value whose presence in this map is to be tested
+ * @return true if this map maps one or more keys to the specified
+ * value
+ * @throws NullPointerException if the specified value is null
+ */
+
+ @Override
+ public boolean containsValue(Object value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+
+ // See explanation of modCount use above
+
+ final Segment[] segments = this.segments;
+ int[] mc = new int[segments.length];
+
+ // Try a few times without locking
+ for (int k = 0; k < RETRIES_BEFORE_LOCK; ++ k) {
+ int mcsum = 0;
+ for (int i = 0; i < segments.length; ++ i) {
+ mcsum += mc[i] = segments[i].modCount;
+ if (segments[i].containsValue(value)) {
+ return true;
+ }
+ }
+ boolean cleanSweep = true;
+ if (mcsum != 0) {
+ for (int i = 0; i < segments.length; ++ i) {
+ if (mc[i] != segments[i].modCount) {
+ cleanSweep = false;
+ break;
+ }
+ }
+ }
+ if (cleanSweep) {
+ return false;
+ }
+ }
+ // Resort to locking all segments
+ for (int i = 0; i < segments.length; ++ i) {
+ segments[i].lock();
+ }
+ boolean found = false;
+ try {
+ for (int i = 0; i < segments.length; ++ i) {
+ if (segments[i].containsValue(value)) {
+ found = true;
+ break;
+ }
+ }
+ } finally {
+ for (int i = 0; i < segments.length; ++ i) {
+ segments[i].unlock();
+ }
+ }
+ return found;
+ }
+
+ /**
+ * Legacy method testing if some key maps into the specified value in this
+ * table. This method is identical in functionality to
+ * {@link #containsValue}, and exists solely to ensure full compatibility
+ * with class {@link java.util.Hashtable}, which supported this method prior to
+ * introduction of the Java Collections framework.
+ *
+ * @param value a value to search for
+ * @return true if and only if some key maps to the value
+ * argument in this table as determined by the equals
+ * method; false otherwise
+ * @throws NullPointerException if the specified value is null
+ */
+ public boolean contains(Object value) {
+ return containsValue(value);
+ }
+
+ /**
+ * Maps the specified key to the specified value in this table. Neither the
+ * key nor the value can be null.
+ *
+ * The value can be retrieved by calling the get method with a
+ * key that is equal to the original key.
+ *
+ * @param key key with which the specified value is to be associated
+ * @param value value to be associated with the specified key
+ * @return the previous value associated with key, or null
+ * if there was no mapping for key
+ * @throws NullPointerException if the specified key or value is null
+ */
+ @Override
+ public V put(K key, V value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ int hash = hashOf(key);
+ return segmentFor(hash).put(key, hash, value, false);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * @return the previous value associated with the specified key, or
+ * null if there was no mapping for the key
+ * @throws NullPointerException if the specified key or value is null
+ */
+ public V putIfAbsent(K key, V value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ int hash = hashOf(key);
+ return segmentFor(hash).put(key, hash, value, true);
+ }
+
+ /**
+ * Copies all of the mappings from the specified map to this one. These
+ * mappings replace any mappings that this map had for any of the keys
+ * currently in the specified map.
+ *
+ * @param m mappings to be stored in this map
+ */
+ @Override
+ public void putAll(Map m) {
+ for (Entry e: m.entrySet()) {
+ put(e.getKey(), e.getValue());
+ }
+ }
+
+ /**
+ * Removes the key (and its corresponding value) from this map. This method
+ * does nothing if the key is not in the map.
+ *
+ * @param key the key that needs to be removed
+ * @return the previous value associated with key, or null
+ * if there was no mapping for key
+ * @throws NullPointerException if the specified key is null
+ */
+ @Override
+ public V remove(Object key) {
+ int hash = hashOf(key);
+ return segmentFor(hash).remove(key, hash, null, false);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * @throws NullPointerException if the specified key is null
+ */
+ public boolean remove(Object key, Object value) {
+ int hash = hashOf(key);
+ if (value == null) {
+ return false;
+ }
+ return segmentFor(hash).remove(key, hash, value, false) != null;
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * @throws NullPointerException if any of the arguments are null
+ */
+ public boolean replace(K key, V oldValue, V newValue) {
+ if (oldValue == null || newValue == null) {
+ throw new NullPointerException();
+ }
+ int hash = hashOf(key);
+ return segmentFor(hash).replace(key, hash, oldValue, newValue);
+ }
+
+ /**
+ * {@inheritDoc}
+ *
+ * @return the previous value associated with the specified key, or
+ * null if there was no mapping for the key
+ * @throws NullPointerException if the specified key or value is null
+ */
+ public V replace(K key, V value) {
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ int hash = hashOf(key);
+ return segmentFor(hash).replace(key, hash, value);
+ }
+
+ /**
+ * Removes all of the mappings from this map.
+ */
+ @Override
+ public void clear() {
+ for (int i = 0; i < segments.length; ++ i) {
+ segments[i].clear();
+ }
+ }
+
+ /**
+ * Returns a {@link java.util.Set} view of the keys contained in this map. The set is
+ * backed by the map, so changes to the map are reflected in the set, and
+ * vice-versa. The set supports element removal, which removes the
+ * corresponding mapping from this map, via the Iterator.remove,
+ * Set.remove, removeAll, retainAll, and
+ * clear operations. It does not support the add or
+ * addAll operations.
+ *
+ *
The view's iterator is a "weakly consistent" iterator that
+ * will never throw {@link java.util.ConcurrentModificationException}, and guarantees
+ * to traverse elements as they existed upon construction of the iterator,
+ * and may (but is not guaranteed to) reflect any modifications subsequent
+ * to construction.
+ */
+ @Override
+ public Set keySet() {
+ Set ks = keySet;
+ return ks != null? ks : (keySet = new KeySet());
+ }
+
+ /**
+ * Returns a {@link java.util.Collection} view of the values contained in this map.
+ * The collection is backed by the map, so changes to the map are reflected
+ * in the collection, and vice-versa. The collection supports element
+ * removal, which removes the corresponding mapping from this map, via the
+ * Iterator.remove, Collection.remove, removeAll,
+ * retainAll, and clear operations. It does not support
+ * the add or addAll operations.
+ *
+ * The view's iterator is a "weakly consistent" iterator that
+ * will never throw {@link java.util.ConcurrentModificationException}, and guarantees
+ * to traverse elements as they existed upon construction of the iterator,
+ * and may (but is not guaranteed to) reflect any modifications subsequent
+ * to construction.
+ */
+ @Override
+ public Collection values() {
+ Collection vs = values;
+ return vs != null? vs : (values = new Values());
+ }
+
+ /**
+ * Returns a {@link java.util.Set} view of the mappings contained in this map.
+ * The set is backed by the map, so changes to the map are reflected in the
+ * set, and vice-versa. The set supports element removal, which removes the
+ * corresponding mapping from the map, via the Iterator.remove,
+ * Set.remove, removeAll, retainAll, and
+ * clear operations. It does not support the add or
+ * addAll operations.
+ *
+ * The view's iterator is a "weakly consistent" iterator that
+ * will never throw {@link java.util.ConcurrentModificationException}, and guarantees
+ * to traverse elements as they existed upon construction of the iterator,
+ * and may (but is not guaranteed to) reflect any modifications subsequent
+ * to construction.
+ */
+ @Override
+ public Set> entrySet() {
+ Set> es = entrySet;
+ return es != null? es : (entrySet = new EntrySet());
+ }
+
+ /**
+ * Returns an enumeration of the keys in this table.
+ *
+ * @return an enumeration of the keys in this table
+ * @see #keySet()
+ */
+ public Enumeration keys() {
+ return new KeyIterator();
+ }
+
+ /**
+ * Returns an enumeration of the values in this table.
+ *
+ * @return an enumeration of the values in this table
+ * @see #values()
+ */
+ public Enumeration elements() {
+ return new ValueIterator();
+ }
+
+ /* ---------------- Iterator Support -------------- */
+
+ abstract class HashIterator {
+ int nextSegmentIndex;
+ int nextTableIndex;
+ HashEntry[] currentTable;
+ HashEntry nextEntry;
+ HashEntry lastReturned;
+ K currentKey; // Strong reference to weak key (prevents gc)
+
+ HashIterator() {
+ nextSegmentIndex = segments.length - 1;
+ nextTableIndex = -1;
+ advance();
+ }
+
+ public void rewind() {
+ nextSegmentIndex = segments.length - 1;
+ nextTableIndex = -1;
+ currentTable = null;
+ nextEntry = null;
+ lastReturned = null;
+ currentKey = null;
+ advance();
+ }
+
+ public boolean hasMoreElements() {
+ return hasNext();
+ }
+
+ final void advance() {
+ if (nextEntry != null && (nextEntry = nextEntry.next) != null) {
+ return;
+ }
+
+ while (nextTableIndex >= 0) {
+ if ((nextEntry = currentTable[nextTableIndex --]) != null) {
+ return;
+ }
+ }
+
+ while (nextSegmentIndex >= 0) {
+ Segment seg = segments[nextSegmentIndex --];
+ if (seg.count != 0) {
+ currentTable = seg.table;
+ for (int j = currentTable.length - 1; j >= 0; -- j) {
+ if ((nextEntry = currentTable[j]) != null) {
+ nextTableIndex = j - 1;
+ return;
+ }
+ }
+ }
+ }
+ }
+
+ public boolean hasNext() {
+ while (nextEntry != null) {
+ if (nextEntry.key() != null) {
+ return true;
+ }
+ advance();
+ }
+
+ return false;
+ }
+
+ HashEntry nextEntry() {
+ do {
+ if (nextEntry == null) {
+ throw new NoSuchElementException();
+ }
+
+ lastReturned = nextEntry;
+ currentKey = lastReturned.key();
+ advance();
+ } while (currentKey == null); // Skip GC'd keys
+
+ return lastReturned;
+ }
+
+ public void remove() {
+ if (lastReturned == null) {
+ throw new IllegalStateException();
+ }
+ ConcurrentIdentityHashMap.this.remove(currentKey);
+ lastReturned = null;
+ }
+ }
+
+ final class KeyIterator
+ extends HashIterator implements ReusableIterator, Enumeration {
+
+ public K next() {
+ return super.nextEntry().key();
+ }
+
+ public K nextElement() {
+ return super.nextEntry().key();
+ }
+ }
+
+ final class ValueIterator
+ extends HashIterator implements ReusableIterator, Enumeration {
+
+ public V next() {
+ return super.nextEntry().value();
+ }
+
+ public V nextElement() {
+ return super.nextEntry().value();
+ }
+ }
+
+ /*
+ * This class is needed for JDK5 compatibility.
+ */
+ static class SimpleEntry implements Entry {
+
+ private static final long serialVersionUID = -8144765946475398746L;
+
+ private final K key;
+
+ private V value;
+
+ public SimpleEntry(K key, V value) {
+ this.key = key;
+ this.value = value;
+
+ }
+
+ public SimpleEntry(Entry entry) {
+ this.key = entry.getKey();
+ this.value = entry.getValue();
+
+ }
+
+ public K getKey() {
+ return key;
+ }
+
+ public V getValue() {
+ return value;
+ }
+
+ public V setValue(V value) {
+ V oldValue = this.value;
+ this.value = value;
+ return oldValue;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof Entry)) {
+ return false;
+ }
+ @SuppressWarnings("rawtypes")
+ Entry e = (Entry) o;
+ return eq(key, e.getKey()) && eq(value, e.getValue());
+ }
+
+ @Override
+ public int hashCode() {
+ return (key == null? 0 : key.hashCode()) ^ (value == null? 0 : value.hashCode());
+ }
+
+ @Override
+ public String toString() {
+ return key + "=" + value;
+ }
+
+ private static boolean eq(Object o1, Object o2) {
+ return o1 == null? o2 == null : o1.equals(o2);
+ }
+ }
+
+ /**
+ * Custom Entry class used by EntryIterator.next(), that relays setValue
+ * changes to the underlying map.
+ */
+ final class WriteThroughEntry extends SimpleEntry {
+
+ WriteThroughEntry(K k, V v) {
+ super(k, v);
+ }
+
+ /**
+ * Set our entry's value and write through to the map. The value to
+ * return is somewhat arbitrary here. Since a WriteThroughEntry does not
+ * necessarily track asynchronous changes, the most recent "previous"
+ * value could be different from what we return (or could even have been
+ * removed in which case the put will re-establish). We do not and can
+ * not guarantee more.
+ */
+ @Override
+ public V setValue(V value) {
+
+ if (value == null) {
+ throw new NullPointerException();
+ }
+ V v = super.setValue(value);
+ ConcurrentIdentityHashMap.this.put(getKey(), value);
+ return v;
+ }
+
+ }
+
+ final class EntryIterator extends HashIterator implements
+ ReusableIterator> {
+ public Entry next() {
+ HashEntry e = super.nextEntry();
+ return new WriteThroughEntry(e.key(), e.value());
+ }
+ }
+
+ final class KeySet extends AbstractSet {
+ @Override
+ public Iterator iterator() {
+
+ return new KeyIterator();
+ }
+
+ @Override
+ public int size() {
+ return ConcurrentIdentityHashMap.this.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return ConcurrentIdentityHashMap.this.isEmpty();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ return ConcurrentIdentityHashMap.this.containsKey(o);
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ return ConcurrentIdentityHashMap.this.remove(o) != null;
+
+ }
+
+ @Override
+ public void clear() {
+ ConcurrentIdentityHashMap.this.clear();
+ }
+ }
+
+ final class Values extends AbstractCollection {
+ @Override
+ public Iterator iterator() {
+ return new ValueIterator();
+ }
+
+ @Override
+ public int size() {
+ return ConcurrentIdentityHashMap.this.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return ConcurrentIdentityHashMap.this.isEmpty();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ return ConcurrentIdentityHashMap.this.containsValue(o);
+ }
+
+ @Override
+ public void clear() {
+ ConcurrentIdentityHashMap.this.clear();
+ }
+ }
+
+ final class EntrySet extends AbstractSet> {
+ @Override
+ public Iterator> iterator() {
+ return new EntryIterator();
+ }
+
+ @Override
+ public boolean contains(Object o) {
+ if (!(o instanceof Entry)) {
+ return false;
+ }
+ Entry e = (Entry) o;
+ V v = ConcurrentIdentityHashMap.this.get(e.getKey());
+ return v != null && v.equals(e.getValue());
+ }
+
+ @Override
+ public boolean remove(Object o) {
+ if (!(o instanceof Entry)) {
+ return false;
+ }
+ Entry e = (Entry) o;
+ return ConcurrentIdentityHashMap.this.remove(e.getKey(), e.getValue());
+ }
+
+ @Override
+ public int size() {
+ return ConcurrentIdentityHashMap.this.size();
+ }
+
+ @Override
+ public boolean isEmpty() {
+ return ConcurrentIdentityHashMap.this.isEmpty();
+ }
+
+ @Override
+ public void clear() {
+ ConcurrentIdentityHashMap.this.clear();
+ }
+ }
+}
diff --git a/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/ReusableIterator.java b/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/ReusableIterator.java
new file mode 100644
index 0000000000..210edbe65d
--- /dev/null
+++ b/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/ReusableIterator.java
@@ -0,0 +1,27 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ *
+ * Red Hat licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package org.jboss.netty.akka.util.internal;
+
+import java.util.Iterator;
+
+/**
+ * @author The Netty Project
+ * @author Trustin Lee
+ * @version $Rev: 2080 $, $Date: 2010-01-26 18:04:19 +0900 (Tue, 26 Jan 2010) $
+ */
+public interface ReusableIterator extends Iterator {
+ void rewind();
+}
diff --git a/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/SharedResourceMisuseDetector.java b/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/SharedResourceMisuseDetector.java
new file mode 100644
index 0000000000..a93db3f7ac
--- /dev/null
+++ b/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/SharedResourceMisuseDetector.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ *
+ * Red Hat licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package org.jboss.netty.akka.util.internal;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.jboss.netty.akka.logging.InternalLogger;
+import org.jboss.netty.akka.logging.InternalLoggerFactory;
+
+/**
+ * Warn when user creates too many instances to avoid {@link OutOfMemoryError}.
+ *
+ * @author The Netty Project
+ * @author Trustin Lee
+ * @version $Rev: 2234 $, $Date: 2010-04-06 18:23:25 +0900 (Tue, 06 Apr 2010) $
+ */
+public class SharedResourceMisuseDetector {
+
+ private static final int MAX_ACTIVE_INSTANCES = 256;
+ private static final InternalLogger logger =
+ InternalLoggerFactory.getInstance(SharedResourceMisuseDetector.class);
+
+ private final Class type;
+ private final AtomicLong activeInstances = new AtomicLong();
+ private final AtomicBoolean logged = new AtomicBoolean();
+
+ public SharedResourceMisuseDetector(Class type) {
+ if (type == null) {
+ throw new NullPointerException("type");
+ }
+ this.type = type;
+ }
+
+ public void increase() {
+ if (activeInstances.incrementAndGet() > MAX_ACTIVE_INSTANCES) {
+ if (logged.compareAndSet(false, true)) {
+ logger.warn(
+ "You are creating too many " + type.getSimpleName() +
+ " instances. " + type.getSimpleName() +
+ " is a shared resource that must be reused across the" +
+ " application, so that only a few instances are created.");
+ }
+ }
+ }
+
+ public void decrease() {
+ activeInstances.decrementAndGet();
+ }
+}
diff --git a/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/StackTraceSimplifier.java b/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/StackTraceSimplifier.java
new file mode 100644
index 0000000000..3a0842b409
--- /dev/null
+++ b/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/StackTraceSimplifier.java
@@ -0,0 +1,85 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ *
+ * Red Hat licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package org.jboss.netty.akka.util.internal;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Pattern;
+
+import org.jboss.netty.akka.util.DebugUtil;
+
+/**
+ * Simplifies an exception stack trace by removing unnecessary
+ * {@link StackTraceElement}s. Please note that the stack trace simplification
+ * is disabled if {@linkplain DebugUtil debug mode} is turned on.
+ *
+ * @author The Netty Project
+ * @author Trustin Lee
+ *
+ * @version $Rev: 2080 $, $Date: 2010-01-26 18:04:19 +0900 (Tue, 26 Jan 2010) $
+ *
+ */
+public class StackTraceSimplifier {
+
+ private static final boolean SIMPLIFY_STACK_TRACE = !DebugUtil.isDebugEnabled();
+ private static final Pattern EXCLUDED_STACK_TRACE =
+ Pattern.compile(
+ "^org\\.jboss\\.netty\\." +
+ "(util\\.(ThreadRenamingRunnable|internal\\.DeadLockProofWorker)" +
+ "|channel\\.(SimpleChannel(Upstream|Downstream)?Handler|(Default|Static)ChannelPipeline.*))(\\$.*)?$");
+
+ /**
+ * Removes unnecessary {@link StackTraceElement}s from the specified
+ * exception. {@link ThreadRenamingRunnable}, {@link SimpleChannelHandler},
+ * {@link DefaultChannelPipeline}, and {@link StaticChannelPipeline}
+ * will be dropped from the trace.
+ */
+ public static void simplify(Throwable e) {
+ if (!SIMPLIFY_STACK_TRACE) {
+ return;
+ }
+
+ if (e.getCause() != null) {
+ simplify(e.getCause());
+ }
+
+ StackTraceElement[] trace = e.getStackTrace();
+ if (trace == null || trace.length == 0) {
+ return;
+ }
+
+ // Perhaps Netty bug. Let us not strip things out.
+ if (EXCLUDED_STACK_TRACE.matcher(trace[0].getClassName()).matches()) {
+ return;
+ }
+
+ List simpleTrace =
+ new ArrayList(trace.length);
+
+ simpleTrace.add(trace[0]);
+
+ // Remove unnecessary stack trace elements.
+ for (int i = 1; i < trace.length; i ++) {
+ if (EXCLUDED_STACK_TRACE.matcher(trace[i].getClassName()).matches()) {
+ continue;
+ }
+ simpleTrace.add(trace[i]);
+ }
+
+ e.setStackTrace(
+ simpleTrace.toArray(new StackTraceElement[simpleTrace.size()]));
+ }
+}
diff --git a/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/SystemPropertyUtil.java b/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/SystemPropertyUtil.java
new file mode 100644
index 0000000000..bf3e2ac571
--- /dev/null
+++ b/akka-actor/src/main/java/org/jboss/netty/akka/util/internal/SystemPropertyUtil.java
@@ -0,0 +1,89 @@
+/*
+ * Copyright 2009 Red Hat, Inc.
+ *
+ * Red Hat licenses this file to you under the Apache License, version 2.0
+ * (the "License"); you may not use this file except in compliance with the
+ * License. You may obtain a copy of the License at:
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package org.jboss.netty.akka.util.internal;
+
+import java.util.regex.Pattern;
+
+/**
+ * Accesses the system property swallowing a {@link SecurityException}.
+ *
+ * @author The Netty Project
+ * @author Trustin Lee
+ *
+ * @version $Rev: 2161 $, $Date: 2010-02-18 11:12:15 +0900 (Thu, 18 Feb 2010) $
+ *
+ */
+public class SystemPropertyUtil {
+
+ /**
+ * Returns the value of the Java system property with the specified
+ * {@code key}.
+ *
+ * @return the property value.
+ * {@code null} if there's no such property or if an access to the
+ * specified property is not allowed.
+ */
+ public static String get(String key) {
+ try {
+ return System.getProperty(key);
+ } catch (Exception e) {
+ return null;
+ }
+ }
+
+ /**
+ * Returns the value of the Java system property with the specified
+ * {@code key}, while falling back to the specified default value if
+ * the property access fails.
+ *
+ * @return the property value.
+ * {@code def} if there's no such property or if an access to the
+ * specified property is not allowed.
+ */
+ public static String get(String key, String def) {
+ String value = get(key);
+ if (value == null) {
+ value = def;
+ }
+ return value;
+ }
+
+ /**
+ * Returns the value of the Java system property with the specified
+ * {@code key}, while falling back to the specified default value if
+ * the property access fails.
+ *
+ * @return the property value.
+ * {@code def} if there's no such property or if an access to the
+ * specified property is not allowed.
+ */
+ public static int get(String key, int def) {
+ String value = get(key);
+ if (value == null) {
+ return def;
+ }
+
+ if (Pattern.matches("-?[0-9]+", value)) {
+ return Integer.parseInt(value);
+ } else {
+ return def;
+ }
+ }
+
+ private SystemPropertyUtil() {
+ // Unused
+ }
+}
diff --git a/akka-actor/src/main/scala/akka/AkkaApplication.scala b/akka-actor/src/main/scala/akka/AkkaApplication.scala
index 7ec88972fe..31b328ad49 100644
--- a/akka-actor/src/main/scala/akka/AkkaApplication.scala
+++ b/akka-actor/src/main/scala/akka/AkkaApplication.scala
@@ -17,6 +17,7 @@ import akka.routing.Routing
import akka.remote.RemoteSupport
import akka.serialization.Serialization
import java.net.InetSocketAddress
+import org.jboss.netty.akka.util.HashedWheelTimer
object AkkaApplication {
@@ -174,8 +175,11 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
// TODO correctly pull its config from the config
val dispatcherFactory = new Dispatchers(this)
+
implicit val dispatcher = dispatcherFactory.defaultGlobalDispatcher
- def terminationFuture: Future[ExitStatus] = provider.terminationFuture
+
+ // Start the scheduler before the provider (to prevent null pointers from happening in e.g. Gossiper)
+ val scheduler = new DefaultScheduler(new HashedWheelTimer)
// TODO think about memory consistency effects when doing funky stuff inside constructor
val reflective = new ReflectiveAccess(this)
@@ -183,6 +187,10 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
// TODO think about memory consistency effects when doing funky stuff inside constructor
val provider: ActorRefProvider = reflective.createProvider
+ def terminationFuture: Future[ExitStatus] = provider.terminationFuture
+
+ terminationFuture.onComplete(_ ⇒ scheduler.stop())
+
private class Guardian extends Actor {
def receive = {
case Terminated(_) ⇒ context.self.stop()
@@ -236,9 +244,6 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
// TODO think about memory consistency effects when doing funky stuff inside constructor
val serialization = new Serialization(this)
- val scheduler = new DefaultScheduler
- terminationFuture.onComplete(_ ⇒ scheduler.shutdown())
-
// TODO shutdown all that other stuff, whatever that may be
def stop(): Unit = {
guardian.stop()
diff --git a/akka-actor/src/main/scala/akka/actor/ActorCell.scala b/akka-actor/src/main/scala/akka/actor/ActorCell.scala
index 16f1ec7535..71c9991c84 100644
--- a/akka-actor/src/main/scala/akka/actor/ActorCell.scala
+++ b/akka-actor/src/main/scala/akka/actor/ActorCell.scala
@@ -5,11 +5,9 @@
package akka.actor
import akka.dispatch._
-import akka.util._
import scala.annotation.tailrec
import scala.collection.immutable.{ Stack, TreeMap }
-import scala.collection.JavaConverters
-import java.util.concurrent.{ ScheduledFuture, TimeUnit }
+import java.util.concurrent.TimeUnit
import akka.AkkaApplication
import akka.event.Logging.{ Debug, Warning, Error }
@@ -77,7 +75,7 @@ private[akka] class ActorCell(
final def provider = app.provider
- var futureTimeout: Option[ScheduledFuture[AnyRef]] = None
+ var futureTimeout: Option[org.jboss.netty.akka.util.Timeout] = None
var _children = emptyChildren //Reuse same empty instance to avoid allocating new instance of the Ordering and the actual empty instance for every actor
@@ -335,7 +333,7 @@ private[akka] class ActorCell(
final def cancelReceiveTimeout() {
if (futureTimeout.isDefined) {
- futureTimeout.get.cancel(true)
+ futureTimeout.get.cancel()
futureTimeout = None
}
}
diff --git a/akka-actor/src/main/scala/akka/actor/FSM.scala b/akka-actor/src/main/scala/akka/actor/FSM.scala
index a6a119cef6..a49a6605f3 100644
--- a/akka-actor/src/main/scala/akka/actor/FSM.scala
+++ b/akka-actor/src/main/scala/akka/actor/FSM.scala
@@ -4,11 +4,10 @@
package akka.actor
import akka.util._
-import akka.event.Logging
import scala.collection.mutable
-import java.util.concurrent.ScheduledFuture
import akka.AkkaApplication
+import akka.event.Logging
object FSM {
@@ -31,7 +30,7 @@ object FSM {
case class TimeoutMarker(generation: Long)
case class Timer(name: String, msg: Any, repeat: Boolean, generation: Int)(implicit app: AkkaApplication) {
- private var ref: Option[ScheduledFuture[AnyRef]] = _
+ private var ref: Option[org.jboss.netty.akka.util.Timeout] = _
def schedule(actor: ActorRef, timeout: Duration) {
if (repeat) {
@@ -43,7 +42,7 @@ object FSM {
def cancel {
if (ref.isDefined) {
- ref.get.cancel(true)
+ ref.get.cancel()
ref = None
}
}
@@ -393,7 +392,7 @@ trait FSM[S, D] extends ListenerManagement {
* FSM State data and current timeout handling
*/
private var currentState: State = _
- private var timeoutFuture: Option[ScheduledFuture[AnyRef]] = None
+ private var timeoutFuture: Option[org.jboss.netty.akka.util.Timeout] = None
private var generation: Long = 0L
/*
@@ -458,7 +457,7 @@ trait FSM[S, D] extends ListenerManagement {
case t @ Timer(name, msg, repeat, gen) ⇒
if ((timers contains name) && (timers(name).generation == gen)) {
if (timeoutFuture.isDefined) {
- timeoutFuture.get.cancel(true)
+ timeoutFuture.get.cancel()
timeoutFuture = None
}
generation += 1
@@ -476,7 +475,7 @@ trait FSM[S, D] extends ListenerManagement {
removeListener(actorRef)
case value ⇒ {
if (timeoutFuture.isDefined) {
- timeoutFuture.get.cancel(true)
+ timeoutFuture.get.cancel()
timeoutFuture = None
}
generation += 1
diff --git a/akka-actor/src/main/scala/akka/actor/Scheduler.scala b/akka-actor/src/main/scala/akka/actor/Scheduler.scala
index ae9b86695d..be6c37cfe1 100644
--- a/akka-actor/src/main/scala/akka/actor/Scheduler.scala
+++ b/akka-actor/src/main/scala/akka/actor/Scheduler.scala
@@ -15,132 +15,82 @@
*/
package akka.actor
-import akka.AkkaException
-import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent._
import akka.util.Duration
+import org.jboss.netty.akka.util.{ HashedWheelTimer, TimerTask }
+import akka.AkkaException
+import org.jboss.netty.akka.util.{ Timeout ⇒ TimeOut }
case class SchedulerException(msg: String, e: Throwable) extends AkkaException(msg, e) {
def this(msg: String) = this(msg, null)
}
trait JScheduler {
- def schedule(receiver: ActorRef, message: Any, initialDelay: Long, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef]
- def scheduleOnce(runnable: Runnable, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef]
- def scheduleOnce(receiver: ActorRef, message: Any, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef]
+ def schedule(receiver: ActorRef, message: Any, initialDelay: Long, delay: Long, timeUnit: TimeUnit): TimeOut
+ def scheduleOnce(runnable: Runnable, delay: Long, timeUnit: TimeUnit): TimeOut
+ def scheduleOnce(receiver: ActorRef, message: Any, delay: Long, timeUnit: TimeUnit): TimeOut
}
abstract class Scheduler extends JScheduler {
+ def schedule(f: () ⇒ Unit, initialDelay: Long, delay: Long, timeUnit: TimeUnit): TimeOut
- def schedule(f: () ⇒ Unit, initialDelay: Long, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef]
- def scheduleOnce(f: () ⇒ Unit, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef]
+ def scheduleOnce(f: () ⇒ Unit, delay: Long, timeUnit: TimeUnit): TimeOut
- def schedule(receiver: ActorRef, message: Any, initialDelay: Duration, delay: Duration): ScheduledFuture[AnyRef] =
+ def schedule(receiver: ActorRef, message: Any, initialDelay: Duration, delay: Duration): TimeOut =
schedule(receiver, message, initialDelay.toNanos, delay.toNanos, TimeUnit.NANOSECONDS)
- def schedule(f: () ⇒ Unit, initialDelay: Duration, delay: Duration): ScheduledFuture[AnyRef] =
+ def schedule(f: () ⇒ Unit, initialDelay: Duration, delay: Duration): TimeOut =
schedule(f, initialDelay.toNanos, delay.toNanos, TimeUnit.NANOSECONDS)
- def scheduleOnce(receiver: ActorRef, message: Any, delay: Duration): ScheduledFuture[AnyRef] =
+ def scheduleOnce(receiver: ActorRef, message: Any, delay: Duration): TimeOut =
scheduleOnce(receiver, message, delay.length, delay.unit)
- def scheduleOnce(f: () ⇒ Unit, delay: Duration): ScheduledFuture[AnyRef] =
+ def scheduleOnce(f: () ⇒ Unit, delay: Duration): TimeOut =
scheduleOnce(f, delay.length, delay.unit)
}
-class DefaultScheduler extends Scheduler {
- private def createSendRunnable(receiver: ActorRef, message: Any, throwWhenReceiverExpired: Boolean): Runnable = new Runnable {
- def run = {
- receiver ! message
- if (throwWhenReceiverExpired && receiver.isShutdown) throw new ActorKilledException("Receiver was terminated")
+class DefaultScheduler(hashedWheelTimer: HashedWheelTimer) extends Scheduler {
+ def schedule(receiver: ActorRef, message: Any, initialDelay: Long, delay: Long, timeUnit: TimeUnit): TimeOut =
+ hashedWheelTimer.newTimeout(createContinuousTask(receiver, message, delay, timeUnit), initialDelay, timeUnit)
+
+ def scheduleOnce(runnable: Runnable, delay: Long, timeUnit: TimeUnit): TimeOut =
+ hashedWheelTimer.newTimeout(createSingleTask(runnable), delay, timeUnit)
+
+ def scheduleOnce(receiver: ActorRef, message: Any, delay: Long, timeUnit: TimeUnit): TimeOut =
+ hashedWheelTimer.newTimeout(createSingleTask(receiver, message), delay, timeUnit)
+
+ def schedule(f: () ⇒ Unit, initialDelay: Long, delay: Long, timeUnit: TimeUnit): TimeOut =
+ hashedWheelTimer.newTimeout(createContinuousTask(f, delay, timeUnit), initialDelay, timeUnit)
+
+ def scheduleOnce(f: () ⇒ Unit, delay: Long, timeUnit: TimeUnit): TimeOut =
+ hashedWheelTimer.newTimeout(createSingleTask(f), delay, timeUnit)
+
+ private def createSingleTask(runnable: Runnable): TimerTask =
+ new TimerTask() { def run(timeout: org.jboss.netty.akka.util.Timeout) { runnable.run() } }
+
+ private def createSingleTask(receiver: ActorRef, message: Any) =
+ new TimerTask { def run(timeout: org.jboss.netty.akka.util.Timeout) { receiver ! message } }
+
+ private def createContinuousTask(receiver: ActorRef, message: Any, delay: Long, timeUnit: TimeUnit) = {
+ new TimerTask {
+ def run(timeout: org.jboss.netty.akka.util.Timeout) {
+ receiver ! message
+ timeout.getTimer.newTimeout(this, delay, timeUnit)
+ }
}
}
- private[akka] val service = Executors.newSingleThreadScheduledExecutor(SchedulerThreadFactory)
+ private def createSingleTask(f: () ⇒ Unit): TimerTask =
+ new TimerTask { def run(timeout: org.jboss.netty.akka.util.Timeout) { f() } }
- /**
- * Schedules to send the specified message to the receiver after initialDelay and then repeated after delay.
- * The returned java.util.concurrent.ScheduledFuture can be used to cancel the
- * send of the message.
- */
- def schedule(receiver: ActorRef, message: Any, initialDelay: Long, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef] = {
- try {
- service.scheduleAtFixedRate(createSendRunnable(receiver, message, true), initialDelay, delay, timeUnit).asInstanceOf[ScheduledFuture[AnyRef]]
- } catch {
- case e: Exception ⇒ throw SchedulerException(message + " could not be scheduled on " + receiver, e)
+ private def createContinuousTask(f: () ⇒ Unit, delay: Long, timeUnit: TimeUnit): TimerTask = {
+ new TimerTask {
+ def run(timeout: org.jboss.netty.akka.util.Timeout) {
+ f()
+ timeout.getTimer.newTimeout(this, delay, timeUnit)
+ }
}
}
- /**
- * Schedules to run specified function to the receiver after initialDelay and then repeated after delay,
- * avoid blocking operations since this is executed in the schedulers thread.
- * The returned java.util.concurrent.ScheduledFuture can be used to cancel the
- * execution of the function.
- */
- def schedule(f: () ⇒ Unit, initialDelay: Long, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef] =
- schedule(new Runnable { def run = f() }, initialDelay, delay, timeUnit)
-
- /**
- * Schedules to run specified runnable to the receiver after initialDelay and then repeated after delay,
- * avoid blocking operations since this is executed in the schedulers thread.
- * The returned java.util.concurrent.ScheduledFuture can be used to cancel the
- * execution of the runnable.
- */
- def schedule(runnable: Runnable, initialDelay: Long, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef] = {
- try {
- service.scheduleAtFixedRate(runnable, initialDelay, delay, timeUnit).asInstanceOf[ScheduledFuture[AnyRef]]
- } catch {
- case e: Exception ⇒ throw SchedulerException("Failed to schedule a Runnable", e)
- }
- }
-
- /**
- * Schedules to send the specified message to the receiver after delay.
- * The returned java.util.concurrent.ScheduledFuture can be used to cancel the
- * send of the message.
- */
- def scheduleOnce(receiver: ActorRef, message: Any, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef] = {
- try {
- service.schedule(createSendRunnable(receiver, message, false), delay, timeUnit).asInstanceOf[ScheduledFuture[AnyRef]]
- } catch {
- case e: Exception ⇒ throw SchedulerException(message + " could not be scheduleOnce'd on " + receiver, e)
- }
- }
-
- /**
- * Schedules a function to be run after delay,
- * avoid blocking operations since the runnable is executed in the schedulers thread.
- * The returned java.util.concurrent.ScheduledFuture can be used to cancel the
- * execution of the function.
- */
- def scheduleOnce(f: () ⇒ Unit, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef] =
- scheduleOnce(new Runnable { def run = f() }, delay, timeUnit)
-
- /**
- * Schedules a runnable to be run after delay,
- * avoid blocking operations since the runnable is executed in the schedulers thread.
- * The returned java.util.concurrent.ScheduledFuture can be used to cancel the
- * execution of the runnable.
- */
- def scheduleOnce(runnable: Runnable, delay: Long, timeUnit: TimeUnit): ScheduledFuture[AnyRef] = {
- try {
- service.schedule(runnable, delay, timeUnit).asInstanceOf[ScheduledFuture[AnyRef]]
- } catch {
- case e: Exception ⇒ throw SchedulerException("Failed to scheduleOnce a Runnable", e)
- }
- }
-
- private[akka] def shutdown() { service.shutdownNow() }
-}
-
-private object SchedulerThreadFactory extends ThreadFactory {
- private val count = new AtomicLong(0)
- val threadFactory = Executors.defaultThreadFactory()
-
- def newThread(r: Runnable): Thread = {
- val thread = threadFactory.newThread(r)
- thread.setName("akka:scheduler-" + count.incrementAndGet())
- thread.setDaemon(true)
- thread
- }
-}
+ private[akka] def stop() = hashedWheelTimer.stop()
+}
\ No newline at end of file
diff --git a/akka-actor/src/main/scala/akka/dispatch/Future.scala b/akka-actor/src/main/scala/akka/dispatch/Future.scala
index 8051bd2336..945c3df5b1 100644
--- a/akka-actor/src/main/scala/akka/dispatch/Future.scala
+++ b/akka-actor/src/main/scala/akka/dispatch/Future.scala
@@ -21,7 +21,7 @@ import java.util.{ LinkedList ⇒ JLinkedList }
import scala.annotation.tailrec
import scala.collection.mutable.Stack
import akka.util.{ Switch, Duration, BoxedType }
-import java.util.concurrent.atomic.{ AtomicReferenceFieldUpdater, AtomicInteger, AtomicReference, AtomicBoolean }
+import java.util.concurrent.atomic.{ AtomicReferenceFieldUpdater, AtomicInteger, AtomicBoolean }
class FutureTimeoutException(message: String, cause: Throwable = null) extends AkkaException(message, cause) {
def this(message: String) = this(message, null)
@@ -985,7 +985,7 @@ class DefaultPromise[T](val timeout: Timeout)(implicit val dispatcher: MessageDi
}
}
val timeoutFuture = dispatcher.app.scheduler.scheduleOnce(runnable, timeLeft(), NANOS)
- onComplete(_ ⇒ timeoutFuture.cancel(true))
+ onComplete(_ ⇒ timeoutFuture.cancel())
false
} else true
} else false
diff --git a/akka-remote/src/main/scala/akka/remote/Gossiper.scala b/akka-remote/src/main/scala/akka/remote/Gossiper.scala
index 4a5efc4f40..fba6bfd04b 100644
--- a/akka-remote/src/main/scala/akka/remote/Gossiper.scala
+++ b/akka-remote/src/main/scala/akka/remote/Gossiper.scala
@@ -109,7 +109,7 @@ class Gossiper(remote: Remote) {
private val failureDetector = remote.failureDetector
private val connectionManager = new RemoteConnectionManager(app, remote, Map.empty[InetSocketAddress, ActorRef])
private val seeds = Set(address) // FIXME read in list of seeds from config
- private val scheduler = new DefaultScheduler
+ private val scheduler = app.scheduler
private val address = app.defaultAddress
private val nodeFingerprint = address.##