[] 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/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 dcc3b74731..d9ad8706b2 100644
--- a/akka-actor/src/main/scala/akka/AkkaApplication.scala
+++ b/akka-actor/src/main/scala/akka/AkkaApplication.scala
@@ -5,7 +5,6 @@ package akka
import akka.config._
import akka.actor._
-import akka.dispatch._
import akka.event._
import akka.util.duration._
import java.net.InetAddress
@@ -13,10 +12,8 @@ import com.eaio.uuid.UUID
import akka.dispatch.{ Dispatchers, Future }
import akka.util.Duration
import akka.util.ReflectiveAccess
-import akka.routing.Routing
import akka.serialization.Serialization
-import java.net.InetSocketAddress
-import remote.{ RemoteAddress, RemoteSupport }
+import remote.{ RemoteAddress }
object AkkaApplication {
@@ -167,8 +164,6 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
def port: Int = defaultAddress.port
- def address: String = hostname + ":" + port.toString
-
// this provides basic logging (to stdout) until .start() is called below
val mainbus = new MainBus(DebugMainBus)
mainbus.startStdoutLogger(AkkaConfig)
@@ -176,8 +171,10 @@ 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
+
+ def scheduler = provider.scheduler
// TODO think about memory consistency effects when doing funky stuff inside constructor
val reflective = new ReflectiveAccess(this)
@@ -190,6 +187,8 @@ 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
+
private class Guardian extends Actor {
def receive = {
case Terminated(_) ⇒ context.self.stop()
@@ -243,9 +242,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())
-
/**
* Create an actor path under the application supervisor (/app).
*/
diff --git a/akka-actor/src/main/scala/akka/actor/ActorCell.scala b/akka-actor/src/main/scala/akka/actor/ActorCell.scala
index f16a9867c8..68d082d4d5 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 }
@@ -79,7 +77,7 @@ private[akka] class ActorCell(
final def provider = app.provider
- var futureTimeout: Option[ScheduledFuture[AnyRef]] = None
+ var futureTimeout: Option[Cancellable] = None
var childrenRefs = emptyChildrenRefs
@@ -348,7 +346,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/ActorRef.scala b/akka-actor/src/main/scala/akka/actor/ActorRef.scala
index 19223416cf..781dd3e114 100644
--- a/akka-actor/src/main/scala/akka/actor/ActorRef.scala
+++ b/akka-actor/src/main/scala/akka/actor/ActorRef.scala
@@ -12,6 +12,7 @@ import akka.AkkaApplication
import akka.serialization.Serialization
import java.net.InetSocketAddress
import akka.remote.RemoteAddress
+import java.util.concurrent.TimeUnit
/**
* ActorRef is an immutable and serializable handle to an Actor.
@@ -171,7 +172,7 @@ class LocalActorRef private[akka] (
def name = path.name
- def address: String = _app.address + path.toString
+ def address: String = _app.defaultAddress + path.toString
private[this] val actorCell = new ActorCell(_app, this, props, _supervisor, receiveTimeout, hotswap)
actorCell.start()
@@ -379,7 +380,7 @@ class DeadLetterActorRef(val app: AkkaApplication) extends MinimalActorRef {
// FIXME (actor path): put this under the sys guardian supervisor
val path: ActorPath = app.root / "sys" / name
- def address: String = app.address + path.toString
+ def address: String = app.defaultAddress + path.toString
override def isShutdown(): Boolean = true
@@ -401,7 +402,7 @@ abstract class AskActorRef(protected val app: AkkaApplication)(timeout: Timeout
// FIXME (actor path): put this under the tmp guardian supervisor
val path: ActorPath = app.root / "tmp" / name
- def address: String = app.address + path.toString
+ def address: String = app.defaultAddress + path.toString
{
val callback: Future[Any] ⇒ Unit = { _ ⇒ app.deathWatch.publish(Terminated(AskActorRef.this)); whenDone() }
@@ -431,4 +432,4 @@ abstract class AskActorRef(protected val app: AkkaApplication)(timeout: Timeout
@throws(classOf[java.io.ObjectStreamException])
private def writeReplace(): AnyRef = app.provider.serialize(this)
-}
+}
\ No newline at end of file
diff --git a/akka-actor/src/main/scala/akka/actor/ActorRefProvider.scala b/akka-actor/src/main/scala/akka/actor/ActorRefProvider.scala
index 3c1f185a69..f5662f9f57 100644
--- a/akka-actor/src/main/scala/akka/actor/ActorRefProvider.scala
+++ b/akka-actor/src/main/scala/akka/actor/ActorRefProvider.scala
@@ -8,12 +8,13 @@ import akka.config.ConfigurationException
import akka.util.ReflectiveAccess
import akka.routing._
import akka.AkkaApplication
-import java.util.concurrent.ConcurrentHashMap
import com.eaio.uuid.UUID
import akka.AkkaException
-import akka.event.{ ActorClassification, DeathWatch, Logging }
import akka.dispatch._
import scala.annotation.tailrec
+import org.jboss.netty.akka.util.HashedWheelTimer
+import java.util.concurrent.{ TimeUnit, Executors, ConcurrentHashMap }
+import akka.event.{ LoggingAdapter, ActorClassification, DeathWatch, Logging }
/**
* Interface for all ActorRef providers to implement.
@@ -31,6 +32,8 @@ trait ActorRefProvider {
*/
private[akka] def deployer: Deployer
+ private[akka] def scheduler: Scheduler
+
private[akka] def actorOf(props: Props, supervisor: ActorRef, name: String, systemService: Boolean): ActorRef
private[akka] def actorOf(props: Props, supervisor: ActorRef, path: ActorPath, systemService: Boolean): ActorRef
@@ -101,10 +104,17 @@ class ActorRefProviderException(message: String) extends AkkaException(message)
*/
class LocalActorRefProvider(val app: AkkaApplication) extends ActorRefProvider {
+ val log = Logging(app.mainbus, this)
+
private[akka] val deployer: Deployer = new Deployer(app)
val terminationFuture = new DefaultPromise[AkkaApplication.ExitStatus](Timeout.never)(app.dispatcher)
- val log = Logging(app.mainbus, this)
+
+ private[akka] val scheduler: Scheduler = { //TODO FIXME Make this configurable
+ val s = new DefaultScheduler(new HashedWheelTimer(log, Executors.defaultThreadFactory, 100, TimeUnit.MILLISECONDS, 512))
+ terminationFuture.onComplete(_ ⇒ s.stop())
+ s
+ }
/**
* Top-level anchor for the supervision hierarchy of this actor system. Will
@@ -119,7 +129,7 @@ class LocalActorRefProvider(val app: AkkaApplication) extends ActorRefProvider {
// FIXME (actor path): move the root path to the new root guardian
val path = app.root
- val address = app.address + path.toString
+ val address = app.defaultAddress + path.toString
override def toString = name
@@ -294,3 +304,58 @@ class LocalDeathWatch extends DeathWatch with ActorClassification {
} else true
}
}
+
+import org.jboss.netty.akka.util.{ HashedWheelTimer, TimerTask }
+class DefaultScheduler(hashedWheelTimer: HashedWheelTimer) extends Scheduler {
+
+ def schedule(receiver: ActorRef, message: Any, initialDelay: Long, delay: Long, timeUnit: TimeUnit): Cancellable =
+ new DefaultCancellable(hashedWheelTimer.newTimeout(createContinuousTask(receiver, message, delay, timeUnit), initialDelay, timeUnit))
+
+ def scheduleOnce(runnable: Runnable, delay: Long, timeUnit: TimeUnit): Cancellable =
+ new DefaultCancellable(hashedWheelTimer.newTimeout(createSingleTask(runnable), delay, timeUnit))
+
+ def scheduleOnce(receiver: ActorRef, message: Any, delay: Long, timeUnit: TimeUnit): Cancellable =
+ new DefaultCancellable(hashedWheelTimer.newTimeout(createSingleTask(receiver, message), delay, timeUnit))
+
+ def schedule(f: () ⇒ Unit, initialDelay: Long, delay: Long, timeUnit: TimeUnit): Cancellable =
+ new DefaultCancellable(hashedWheelTimer.newTimeout(createContinuousTask(f, delay, timeUnit), initialDelay, timeUnit))
+
+ def scheduleOnce(f: () ⇒ Unit, delay: Long, timeUnit: TimeUnit): Cancellable =
+ new DefaultCancellable(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): TimerTask =
+ new TimerTask { def run(timeout: org.jboss.netty.akka.util.Timeout) { receiver ! message } }
+
+ private def createContinuousTask(receiver: ActorRef, message: Any, delay: Long, timeUnit: TimeUnit): TimerTask = {
+ new TimerTask {
+ def run(timeout: org.jboss.netty.akka.util.Timeout) {
+ receiver ! message
+ timeout.getTimer.newTimeout(this, delay, timeUnit)
+ }
+ }
+ }
+
+ private def createSingleTask(f: () ⇒ Unit): TimerTask =
+ new TimerTask { def run(timeout: org.jboss.netty.akka.util.Timeout) { f() } }
+
+ 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)
+ }
+ }
+ }
+
+ private[akka] def stop() = hashedWheelTimer.stop()
+}
+
+class DefaultCancellable(timeout: org.jboss.netty.akka.util.Timeout) extends Cancellable {
+ def cancel() { timeout.cancel() }
+
+ def isCancelled: Boolean = { timeout.isCancelled }
+}
+
diff --git a/akka-actor/src/main/scala/akka/actor/FSM.scala b/akka-actor/src/main/scala/akka/actor/FSM.scala
index a6a119cef6..682f5ae3c6 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[Cancellable] = _
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[Cancellable] = 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..d12dcb6329 100644
--- a/akka-actor/src/main/scala/akka/actor/Scheduler.scala
+++ b/akka-actor/src/main/scala/akka/actor/Scheduler.scala
@@ -15,132 +15,40 @@
*/
package akka.actor
-import akka.AkkaException
-import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent._
import akka.util.Duration
+import akka.AkkaException
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): Cancellable
+ def scheduleOnce(runnable: Runnable, delay: Long, timeUnit: TimeUnit): Cancellable
+ def scheduleOnce(receiver: ActorRef, message: Any, delay: Long, timeUnit: TimeUnit): Cancellable
}
abstract class Scheduler extends JScheduler {
+ def schedule(f: () ⇒ Unit, initialDelay: Long, delay: Long, timeUnit: TimeUnit): Cancellable
- 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): Cancellable
- def schedule(receiver: ActorRef, message: Any, initialDelay: Duration, delay: Duration): ScheduledFuture[AnyRef] =
+ def schedule(receiver: ActorRef, message: Any, initialDelay: Duration, delay: Duration): Cancellable =
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): Cancellable =
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): Cancellable =
scheduleOnce(receiver, message, delay.length, delay.unit)
- def scheduleOnce(f: () ⇒ Unit, delay: Duration): ScheduledFuture[AnyRef] =
+ def scheduleOnce(f: () ⇒ Unit, delay: Duration): Cancellable =
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")
- }
- }
+trait Cancellable {
+ def cancel(): Unit
- private[akka] val service = Executors.newSingleThreadScheduledExecutor(SchedulerThreadFactory)
-
- /**
- * 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)
- }
- }
-
- /**
- * 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
- }
-}
+ def isCancelled: Boolean
+}
\ 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-actor/src/main/scala/akka/remote/RemoteInterface.scala b/akka-actor/src/main/scala/akka/remote/RemoteInterface.scala
index a1d5eb57ee..46b7c1d09c 100644
--- a/akka-actor/src/main/scala/akka/remote/RemoteInterface.scala
+++ b/akka-actor/src/main/scala/akka/remote/RemoteInterface.scala
@@ -29,7 +29,8 @@ object RemoteAddress {
trait RemoteAddress extends Serializable {
def hostname: String
def port: Int
- override def toString = "" + hostname + ":" + port
+ @transient
+ override lazy val toString = "" + hostname + ":" + port
}
class RemoteException(message: String) extends AkkaException(message)
diff --git a/akka-remote/src/main/scala/akka/remote/Gossiper.scala b/akka-remote/src/main/scala/akka/remote/Gossiper.scala
index f98b2ffa8e..f158cf1b02 100644
--- a/akka-remote/src/main/scala/akka/remote/Gossiper.scala
+++ b/akka-remote/src/main/scala/akka/remote/Gossiper.scala
@@ -4,7 +4,6 @@
package akka.remote
-import akka.AkkaApplication
import akka.actor._
import akka.actor.Status._
import akka.event.Logging
@@ -107,7 +106,6 @@ class Gossiper(remote: Remote) {
private val failureDetector = remote.failureDetector
private val connectionManager = new RemoteConnectionManager(app, remote, Map.empty[RemoteAddress, ActorRef])
private val seeds = Set(address) // FIXME read in list of seeds from config
- private val scheduler = new DefaultScheduler
private val address = app.defaultAddress
private val nodeFingerprint = address.##
@@ -124,8 +122,8 @@ class Gossiper(remote: Remote) {
{
// start periodic gossip and cluster scrutinization - default is run them every second with 1/2 second in between
- scheduler schedule (() ⇒ initateGossip(), initalDelayForGossip.toSeconds, gossipFrequency.toSeconds, timeUnit)
- scheduler schedule (() ⇒ scrutinize(), initalDelayForGossip.toSeconds, gossipFrequency.toSeconds, timeUnit)
+ app.scheduler schedule (() ⇒ initateGossip(), initalDelayForGossip.toSeconds, gossipFrequency.toSeconds, timeUnit)
+ app.scheduler schedule (() ⇒ scrutinize(), initalDelayForGossip.toSeconds, gossipFrequency.toSeconds, timeUnit)
}
/**
diff --git a/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala b/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
index 496dcc051b..de31074295 100644
--- a/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
+++ b/akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
@@ -51,6 +51,8 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
def defaultDispatcher = app.dispatcher
def defaultTimeout = app.AkkaConfig.ActorTimeout
+ def scheduler: Scheduler = local.scheduler
+
private[akka] def actorOf(props: Props, supervisor: ActorRef, name: String, systemService: Boolean): ActorRef =
actorOf(props, supervisor, supervisor.path / name, systemService)