raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Removing InetSocketAddress as much as possible from the remoting, switching to RemoteAddress for an easier way forward with different transports. Also removing quite a few allocations internally in the remoting as a side-efect of this.

Browse source
This commit is contained in:
Viktor Klang 2011-11-10 17:39:04 +01:00
parent 0800511ac9
commit ba9281e267
12 changed files with 122 additions and 146 deletions
Download patch file Download diff file Expand all files Collapse all files

8
akka-actor/src/main/scala/akka/AkkaApplication.scala
View file

@ -14,9 +14,9 @@ import akka.dispatch.{ Dispatchers, Future }
import akka.util.Duration
import akka.util.ReflectiveAccess
import akka.routing.Routing
import akka.remote.RemoteSupport
import akka.serialization.Serialization
import java.net.InetSocketAddress
import remote.{ RemoteAddress, RemoteSupport }
object AkkaApplication {
@ -155,7 +155,7 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
case value ⇒ value
}
val defaultAddress = new InetSocketAddress(System.getProperty("akka.remote.hostname") match {
val defaultAddress = RemoteAddress(System.getProperty("akka.remote.hostname") match {
case null | "" ⇒ InetAddress.getLocalHost.getHostAddress
case value ⇒ value
}, System.getProperty("akka.remote.port") match {
@ -163,9 +163,9 @@ class AkkaApplication(val name: String, val config: Configuration) extends Actor
case value ⇒ value.toInt
})
def hostname: String = defaultAddress.getAddress.getHostAddress
def hostname: String = defaultAddress.hostname
def port: Int = defaultAddress.getPort
def port: Int = defaultAddress.port
// this provides basic logging (to stdout) until .start() is called below
val mainbus = new MainBus(DebugMainBus)

4
akka-actor/src/main/scala/akka/actor/ActorRef.scala
View file

@ -13,6 +13,7 @@ import akka.event.ActorEventBus
import akka.serialization.Serialization
import akka.actor.DeadLetterActorRef.SerializedDeadLetterActorRef
import java.net.InetSocketAddress
import akka.remote.RemoteAddress
/**
* ActorRef is an immutable and serializable handle to an Actor.
@ -286,7 +287,8 @@ trait ScalaActorRef { ref: ActorRef ⇒
case class SerializedActorRef(address: String, hostname: String, port: Int) {
import akka.serialization.Serialization.app
def this(address: String, inet: InetSocketAddress) = this(address, inet.getAddress.getHostAddress, inet.getPort)
def this(address: String, remoteAddress: RemoteAddress) = this(address, remoteAddress.hostname, remoteAddress.port)
def this(address: String, remoteAddress: InetSocketAddress) = this(address, remoteAddress.getAddress.getHostAddress, remoteAddress.getPort) //TODO FIXME REMOVE
@throws(classOf[java.io.ObjectStreamException])
def readResolve(): AnyRef = {

52
akka-actor/src/main/scala/akka/remote/RemoteInterface.scala
View file

@ -8,9 +8,29 @@ import akka.actor._
import akka.{ AkkaException, AkkaApplication }
import scala.reflect.BeanProperty
import java.io.{ PrintWriter, PrintStream }
import java.net.InetSocketAddress
import java.io.{ PrintWriter, PrintStream }
object RemoteAddress {
def apply(host: String, port: Int): RemoteAddress = apply(new InetSocketAddress(host, port))
def apply(inetAddress: InetSocketAddress): RemoteAddress = inetAddress match {
case null ⇒ null
case inet ⇒
val host = inet.getAddress.getHostAddress
val portNo = inet.getPort
new RemoteAddress {
def hostname = host
def port = portNo
}
}
}
trait RemoteAddress {
def hostname: String
def port: Int
override def toString = "" + hostname + ":" + port
}
class RemoteException(message: String) extends AkkaException(message)
@ -27,35 +47,35 @@ sealed trait RemoteLifeCycleEvent
* Life-cycle events for RemoteClient.
*/
trait RemoteClientLifeCycleEvent extends RemoteLifeCycleEvent {
def remoteAddress: InetSocketAddress
def remoteAddress: RemoteAddress
}
case class RemoteClientError(
@BeanProperty cause: Throwable,
@BeanProperty remote: RemoteSupport,
@BeanProperty remoteAddress: InetSocketAddress) extends RemoteClientLifeCycleEvent
@BeanProperty remoteAddress: RemoteAddress) extends RemoteClientLifeCycleEvent
case class RemoteClientDisconnected(
@BeanProperty remote: RemoteSupport,
@BeanProperty remoteAddress: InetSocketAddress) extends RemoteClientLifeCycleEvent
@BeanProperty remoteAddress: RemoteAddress) extends RemoteClientLifeCycleEvent
case class RemoteClientConnected(
@BeanProperty remote: RemoteSupport,
@BeanProperty remoteAddress: InetSocketAddress) extends RemoteClientLifeCycleEvent
@BeanProperty remoteAddress: RemoteAddress) extends RemoteClientLifeCycleEvent
case class RemoteClientStarted(
@BeanProperty remote: RemoteSupport,
@BeanProperty remoteAddress: InetSocketAddress) extends RemoteClientLifeCycleEvent
@BeanProperty remoteAddress: RemoteAddress) extends RemoteClientLifeCycleEvent
case class RemoteClientShutdown(
@BeanProperty remote: RemoteSupport,
@BeanProperty remoteAddress: InetSocketAddress) extends RemoteClientLifeCycleEvent
@BeanProperty remoteAddress: RemoteAddress) extends RemoteClientLifeCycleEvent
case class RemoteClientWriteFailed(
@BeanProperty request: AnyRef,
@BeanProperty cause: Throwable,
@BeanProperty remote: RemoteSupport,
@BeanProperty remoteAddress: InetSocketAddress) extends RemoteClientLifeCycleEvent
@BeanProperty remoteAddress: RemoteAddress) extends RemoteClientLifeCycleEvent
/**
* Life-cycle events for RemoteServer.
@ -71,18 +91,18 @@ case class RemoteServerError(
@BeanProperty remote: RemoteSupport) extends RemoteServerLifeCycleEvent
case class RemoteServerClientConnected(
@BeanProperty remote: RemoteSupport,
@BeanProperty val clientAddress: Option[InetSocketAddress]) extends RemoteServerLifeCycleEvent
@BeanProperty val clientAddress: Option[RemoteAddress]) extends RemoteServerLifeCycleEvent
case class RemoteServerClientDisconnected(
@BeanProperty remote: RemoteSupport,
@BeanProperty val clientAddress: Option[InetSocketAddress]) extends RemoteServerLifeCycleEvent
@BeanProperty val clientAddress: Option[RemoteAddress]) extends RemoteServerLifeCycleEvent
case class RemoteServerClientClosed(
@BeanProperty remote: RemoteSupport,
@BeanProperty val clientAddress: Option[InetSocketAddress]) extends RemoteServerLifeCycleEvent
@BeanProperty val clientAddress: Option[RemoteAddress]) extends RemoteServerLifeCycleEvent
case class RemoteServerWriteFailed(
@BeanProperty request: AnyRef,
@BeanProperty cause: Throwable,
@BeanProperty server: RemoteSupport,
@BeanProperty remoteAddress: Option[InetSocketAddress]) extends RemoteServerLifeCycleEvent
@BeanProperty remoteAddress: Option[RemoteAddress]) extends RemoteServerLifeCycleEvent
/**
* Thrown for example when trying to send a message using a RemoteClient that is either not started or shut down.
@ -90,7 +110,7 @@ case class RemoteServerWriteFailed(
class RemoteClientException private[akka] (
message: String,
@BeanProperty val client: RemoteSupport,
val remoteAddress: InetSocketAddress, cause: Throwable = null) extends AkkaException(message, cause)
val remoteAddress: RemoteAddress, cause: Throwable = null) extends AkkaException(message, cause)
/**
* Thrown when the remote server actor dispatching fails for some reason.
@ -127,18 +147,18 @@ abstract class RemoteSupport(val app: AkkaApplication) {
/**
* Shuts down a specific client connected to the supplied remote address returns true if successful
*/
def shutdownClientConnection(address: InetSocketAddress): Boolean
def shutdownClientConnection(address: RemoteAddress): Boolean
/**
* Restarts a specific client connected to the supplied remote address, but only if the client is not shut down
*/
def restartClientConnection(address: InetSocketAddress): Boolean
def restartClientConnection(address: RemoteAddress): Boolean
/** Methods that needs to be implemented by a transport **/
protected[akka] def send(message: Any,
senderOption: Option[ActorRef],
remoteAddress: InetSocketAddress,
remoteAddress: RemoteAddress,
recipient: ActorRef,
loader: Option[ClassLoader]): Unit

9
akka-actor/src/main/scala/akka/routing/ConnectionManager.scala
View file

@ -10,6 +10,7 @@ import scala.annotation.tailrec
import java.util.concurrent.atomic.{ AtomicReference, AtomicInteger }
import java.net.InetSocketAddress
import akka.remote.RemoteAddress
/**
* An Iterable that also contains a version.
@ -71,12 +72,12 @@ trait ConnectionManager {
/**
* Creates a new connection (ActorRef) if it didn't exist. Atomically.
*/
def putIfAbsent(address: InetSocketAddress, newConnectionFactory: () ⇒ ActorRef): ActorRef
def putIfAbsent(address: RemoteAddress, newConnectionFactory: () ⇒ ActorRef): ActorRef
/**
* Fails over connections from one address to another.
*/
def failOver(from: InetSocketAddress, to: InetSocketAddress)
def failOver(from: RemoteAddress, to: RemoteAddress)
}
/**
@ -120,9 +121,9 @@ class LocalConnectionManager(initialConnections: Iterable[ActorRef]) extends Con
}
}
def failOver(from: InetSocketAddress, to: InetSocketAddress) {} // do nothing here
def failOver(from: RemoteAddress, to: RemoteAddress) {} // do nothing here
def putIfAbsent(address: InetSocketAddress, newConnectionFactory: () ⇒ ActorRef): ActorRef = {
def putIfAbsent(address: RemoteAddress, newConnectionFactory: () ⇒ ActorRef): ActorRef = {
throw new UnsupportedOperationException("Not supported")
}
}

29
akka-actor/src/main/scala/akka/util/RemoteAddress.scala
View file

@ -1,29 +0,0 @@
/**
* Copyright (C) 2009-2011 Typesafe Inc. <http://www.typesafe.com>
*/
package akka.util
import java.net.InetSocketAddress
object RemoteAddress {
def apply(hostname: String, port: Int) = new RemoteAddress(hostname, port)
def apply(inetAddress: InetSocketAddress): RemoteAddress = inetAddress match {
case null ⇒ null
case inet ⇒ new RemoteAddress(inet.getAddress.getHostAddress, inet.getPort)
}
}
class RemoteAddress(val hostname: String, val port: Int) {
override val hashCode: Int = {
var result = HashCode.SEED
result = HashCode.hash(result, hostname)
result = HashCode.hash(result, port)
result
}
override def equals(that: Any): Boolean = {
that.isInstanceOf[RemoteAddress] &&
that.asInstanceOf[RemoteAddress].hostname == hostname &&
that.asInstanceOf[RemoteAddress].port == port
}
}

15
akka-remote/src/main/scala/akka/remote/AccrualFailureDetector.scala
View file

@ -4,7 +4,6 @@
package akka.remote
import java.net.InetSocketAddress
import java.util.concurrent.atomic.AtomicReference
import scala.collection.immutable.Map
@ -42,9 +41,9 @@ class AccrualFailureDetector(val threshold: Int = 8, val maxSampleSize: Int = 10
*/
private case class State(
version: Long = 0L,
failureStats: Map[InetSocketAddress, FailureStats] = Map.empty[InetSocketAddress, FailureStats],
intervalHistory: Map[InetSocketAddress, Vector[Long]] = Map.empty[InetSocketAddress, Vector[Long]],
timestamps: Map[InetSocketAddress, Long] = Map.empty[InetSocketAddress, Long])
failureStats: Map[RemoteAddress, FailureStats] = Map.empty[RemoteAddress, FailureStats],
intervalHistory: Map[RemoteAddress, Vector[Long]] = Map.empty[RemoteAddress, Vector[Long]],
timestamps: Map[RemoteAddress, Long] = Map.empty[RemoteAddress, Long])
private val state = new AtomicReference[State](State())
@ -52,13 +51,13 @@ class AccrualFailureDetector(val threshold: Int = 8, val maxSampleSize: Int = 10
* Returns true if the connection is considered to be up and healthy
* and returns false otherwise.
*/
def isAvailable(connection: InetSocketAddress): Boolean = phi(connection) < threshold
def isAvailable(connection: RemoteAddress): Boolean = phi(connection) < threshold
/**
* Records a heartbeat for a connection.
*/
@tailrec
final def heartbeat(connection: InetSocketAddress) {
final def heartbeat(connection: RemoteAddress) {
val oldState = state.get
val latestTimestamp = oldState.timestamps.get(connection)
@ -139,7 +138,7 @@ class AccrualFailureDetector(val threshold: Int = 8, val maxSampleSize: Int = 10
* Implementations of 'Cumulative Distribution Function' for Exponential Distribution.
* For a discussion on the math read [https://issues.apache.org/jira/browse/CASSANDRA-2597].
*/
def phi(connection: InetSocketAddress): Double = {
def phi(connection: RemoteAddress): Double = {
val oldState = state.get
val oldTimestamp = oldState.timestamps.get(connection)
if (oldTimestamp.isEmpty) 0.0D // treat unmanaged connections, e.g. with zero heartbeats, as healthy connections
@ -154,7 +153,7 @@ class AccrualFailureDetector(val threshold: Int = 8, val maxSampleSize: Int = 10
* Removes the heartbeat management for a connection.
*/
@tailrec
final def remove(connection: InetSocketAddress) {
final def remove(connection: RemoteAddress) {
val oldState = state.get
if (oldState.failureStats.contains(connection)) {

28
akka-remote/src/main/scala/akka/remote/Gossiper.scala
View file

@ -12,9 +12,7 @@ import akka.util.duration._
import akka.remote.RemoteProtocol._
import akka.remote.RemoteProtocol.RemoteSystemDaemonMessageType._
import java.net.InetSocketAddress
import java.util.concurrent.atomic.AtomicReference
import java.util.concurrent.TimeUnit
import java.security.SecureRandom
import System.{ currentTimeMillis ⇒ newTimestamp }
@ -27,8 +25,8 @@ import com.google.protobuf.ByteString
* Interface for node membership change listener.
*/
trait NodeMembershipChangeListener {
def nodeConnected(node: InetSocketAddress)
def nodeDisconnected(node: InetSocketAddress)
def nodeConnected(node: RemoteAddress)
def nodeDisconnected(node: RemoteAddress)
}
/**
@ -36,23 +34,23 @@ trait NodeMembershipChangeListener {
*/
case class Gossip(
version: VectorClock,
node: InetSocketAddress,
availableNodes: Set[InetSocketAddress] = Set.empty[InetSocketAddress],
unavailableNodes: Set[InetSocketAddress] = Set.empty[InetSocketAddress])
node: RemoteAddress,
availableNodes: Set[RemoteAddress] = Set.empty[RemoteAddress],
unavailableNodes: Set[RemoteAddress] = Set.empty[RemoteAddress])
/*
// ====== NEW GOSSIP IMPLEMENTATION ======
case class Gossip(
version: VectorClock,
node: InetSocketAddress,
leader: InetSocketAddress, // FIXME leader is always head of 'members', so we probably don't need this field
node: RemoteAddress,
leader: RemoteAddress, // FIXME leader is always head of 'members', so we probably don't need this field
members: SortedSet[Member] = SortetSet.empty[Member](Ordering.fromLessThan[String](_ > _)), // sorted set of members with their status, sorted by name
seen: Map[Member, VectorClock] = Map.empty[Member, VectorClock], // for ring convergence
pendingChanges: Option[Vector[PendingPartitioningChange]] = None, // for handoff
meta: Option[Map[String, Array[Byte]]] = None) // misc meta-data
case class Member(address: InetSocketAddress, status: MemberStatus)
case class Member(address: RemoteAddress, status: MemberStatus)
sealed trait MemberStatus
object MemberStatus {
@ -73,8 +71,8 @@ case class Gossip(
type VNodeMod = AnyRef
case class PendingPartitioningChange(
owner: InetSocketAddress,
nextOwner: InetSocketAddress,
owner: RemoteAddress,
nextOwner: RemoteAddress,
changes: Vector[VNodeMod],
status: PendingPartitioningStatus)
*/
@ -107,7 +105,7 @@ class Gossiper(remote: Remote) {
private val app = remote.app
private val log = Logging(app, this)
private val failureDetector = remote.failureDetector
private val connectionManager = new RemoteConnectionManager(app, remote, Map.empty[InetSocketAddress, ActorRef])
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
@ -231,7 +229,7 @@ class Gossiper(remote: Remote) {
/**
* Gossips set of nodes passed in as argument. Returns 'true' if it gossiped to a "seed" node.
*/
private def gossipTo(nodes: Set[InetSocketAddress]): Boolean = {
private def gossipTo(nodes: Set[RemoteAddress]): Boolean = {
val peers = nodes filter (_ != address) // filter out myself
val peer = selectRandomNode(peers)
val oldState = state.get
@ -323,7 +321,7 @@ class Gossiper(remote: Remote) {
}
}
private def selectRandomNode(nodes: Set[InetSocketAddress]): InetSocketAddress = {
private def selectRandomNode(nodes: Set[RemoteAddress]): RemoteAddress = {
nodes.toList(random.nextInt(nodes.size))
}
}

14
akka-remote/src/main/scala/akka/remote/NetworkEventStream.scala
View file

@ -4,9 +4,7 @@
package akka.remote
import akka.dispatch.PinnedDispatcher
import scala.collection.mutable
import java.net.InetSocketAddress
import akka.actor.{ LocalActorRef, Actor, ActorRef, Props, newUuid }
import akka.actor.Actor._
import akka.AkkaApplication
@ -21,10 +19,10 @@ object NetworkEventStream {
private sealed trait NetworkEventStreamEvent
private case class Register(listener: Listener, connectionAddress: InetSocketAddress)
private case class Register(listener: Listener, connectionAddress: RemoteAddress)
extends NetworkEventStreamEvent
private case class Unregister(listener: Listener, connectionAddress: InetSocketAddress)
private case class Unregister(listener: Listener, connectionAddress: RemoteAddress)
extends NetworkEventStreamEvent
/**
@ -39,8 +37,8 @@ object NetworkEventStream {
*/
private class Channel extends Actor {
val listeners = new mutable.HashMap[InetSocketAddress, mutable.Set[Listener]]() {
override def default(k: InetSocketAddress) = mutable.Set.empty[Listener]
val listeners = new mutable.HashMap[RemoteAddress, mutable.Set[Listener]]() {
override def default(k: RemoteAddress) = mutable.Set.empty[Listener]
}
def receive = {
@ -72,12 +70,12 @@ class NetworkEventStream(val app: AkkaApplication) {
/**
* Registers a network event stream listener (asyncronously).
*/
def register(listener: Listener, connectionAddress: InetSocketAddress) =
def register(listener: Listener, connectionAddress: RemoteAddress) =
sender ! Register(listener, connectionAddress)
/**
* Unregisters a network event stream listener (asyncronously) .
*/
def unregister(listener: Listener, connectionAddress: InetSocketAddress) =
def unregister(listener: Listener, connectionAddress: RemoteAddress) =
sender ! Unregister(listener, connectionAddress)
}

18
akka-remote/src/main/scala/akka/remote/RemoteActorRefProvider.scala
View file

@ -111,9 +111,9 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
case RouterType.Custom(implClass) ⇒ () ⇒ Routing.createCustomRouter(implClass)
}
val connections = (Map.empty[InetSocketAddress, ActorRef] /: remoteAddresses) { (conns, a) ⇒
val inetAddr = new InetSocketAddress(a.hostname, a.port)
conns + (inetAddr -> RemoteActorRef(remote.server, inetAddr, address, None))
val connections = (Map.empty[RemoteAddress, ActorRef] /: remoteAddresses) { (conns, a) ⇒
val remoteAddress = RemoteAddress(a.hostname, a.port)
conns + (remoteAddress -> RemoteActorRef(remote.server, remoteAddress, address, None))
}
val connectionManager = new RemoteConnectionManager(app, remote, connections)
@ -169,19 +169,19 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
}
private[akka] def deserialize(actor: SerializedActorRef): Option[ActorRef] = {
if (optimizeLocalScoped_? && (actor.hostname == app.hostname || actor.hostname == app.defaultAddress.getHostName) && actor.port == app.port) {
val remoteAddress = RemoteAddress(actor.hostname, actor.port)
if (optimizeLocalScoped_? && remoteAddress == app.defaultAddress) {
local.actorFor(actor.address)
} else {
val remoteInetSocketAddress = new InetSocketAddress(actor.hostname, actor.port) //FIXME Drop the InetSocketAddresses and use RemoteAddress
log.debug("{}: Creating RemoteActorRef with address [{}] connected to [{}]", app.defaultAddress, actor.address, remoteInetSocketAddress)
Some(RemoteActorRef(remote.server, remoteInetSocketAddress, actor.address, None)) //Should it be None here
log.debug("{}: Creating RemoteActorRef with address [{}] connected to [{}]", app.defaultAddress, actor.address, remoteAddress)
Some(RemoteActorRef(remote.server, remoteAddress, actor.address, None)) //Should it be None here
}
}
/**
* Using (checking out) actor on a specific node.
*/
def useActorOnNode(remoteAddress: InetSocketAddress, actorAddress: String, actorFactory: () ⇒ Actor) {
def useActorOnNode(remoteAddress: RemoteAddress, actorAddress: String, actorFactory: () ⇒ Actor) {
log.debug("[{}] Instantiating Actor [{}] on node [{}]", app.defaultAddress, actorAddress, remoteAddress)
val actorFactoryBytes =
@ -244,7 +244,7 @@ class RemoteActorRefProvider(val app: AkkaApplication) extends ActorRefProvider
*/
private[akka] case class RemoteActorRef private[akka] (
remote: RemoteSupport,
remoteAddress: InetSocketAddress,
remoteAddress: RemoteAddress,
address: String,
loader: Option[ClassLoader])
extends ActorRef with ScalaActorRef {

17
akka-remote/src/main/scala/akka/remote/RemoteConnectionManager.scala
View file

@ -12,7 +12,6 @@ import akka.event.Logging
import scala.collection.immutable.Map
import scala.annotation.tailrec
import java.net.InetSocketAddress
import java.util.concurrent.atomic.AtomicReference
/**
@ -23,13 +22,13 @@ import java.util.concurrent.atomic.AtomicReference
class RemoteConnectionManager(
app: AkkaApplication,
remote: Remote,
initialConnections: Map[InetSocketAddress, ActorRef] = Map.empty[InetSocketAddress, ActorRef])
initialConnections: Map[RemoteAddress, ActorRef] = Map.empty[RemoteAddress, ActorRef])
extends ConnectionManager {
val log = Logging(app, this)
// FIXME is this VersionedIterable really needed? It is not used I think. Complicates API. See 'def connections' etc.
case class State(version: Long, connections: Map[InetSocketAddress, ActorRef])
case class State(version: Long, connections: Map[RemoteAddress, ActorRef])
extends VersionedIterable[ActorRef] {
def iterable: Iterable[ActorRef] = connections.values
}
@ -55,7 +54,7 @@ class RemoteConnectionManager(
def size: Int = connections.connections.size
def connectionFor(address: InetSocketAddress): Option[ActorRef] = connections.connections.get(address)
def connectionFor(address: RemoteAddress): Option[ActorRef] = connections.connections.get(address)
def isEmpty: Boolean = connections.connections.isEmpty
@ -64,7 +63,7 @@ class RemoteConnectionManager(
}
@tailrec
final def failOver(from: InetSocketAddress, to: InetSocketAddress) {
final def failOver(from: RemoteAddress, to: RemoteAddress) {
log.debug("Failing over connection from [{}] to [{}]", from, to)
val oldState = state.get
@ -95,8 +94,8 @@ class RemoteConnectionManager(
val oldState = state.get()
var changed = false
var faultyAddress: InetSocketAddress = null
var newConnections = Map.empty[InetSocketAddress, ActorRef]
var faultyAddress: RemoteAddress = null
var newConnections = Map.empty[RemoteAddress, ActorRef]
oldState.connections.keys foreach { address ⇒
val actorRef: ActorRef = oldState.connections.get(address).get
@ -122,7 +121,7 @@ class RemoteConnectionManager(
}
@tailrec
final def putIfAbsent(address: InetSocketAddress, newConnectionFactory: () ⇒ ActorRef): ActorRef = {
final def putIfAbsent(address: RemoteAddress, newConnectionFactory: () ⇒ ActorRef): ActorRef = {
val oldState = state.get()
val oldConnections = oldState.connections
@ -149,7 +148,7 @@ class RemoteConnectionManager(
}
}
private[remote] def newConnection(actorAddress: String, inetSocketAddress: InetSocketAddress) = {
private[remote] def newConnection(actorAddress: String, inetSocketAddress: RemoteAddress) = {
RemoteActorRef(remote.server, inetSocketAddress, actorAddress, None)
}
}

69
akka-remote/src/main/scala/akka/remote/netty/NettyRemoteSupport.scala
View file

@ -36,13 +36,11 @@ class RemoteClientMessageBufferException(message: String, cause: Throwable = nul
*/
abstract class RemoteClient private[akka] (
val remoteSupport: NettyRemoteSupport,
val remoteAddress: InetSocketAddress) {
val remoteAddress: RemoteAddress) {
val log = Logging(remoteSupport.app, this)
val name = simpleName(this) + "@" +
remoteAddress.getAddress.getHostAddress + "::" +
remoteAddress.getPort
val name = simpleName(this) + "@" + remoteAddress
private[remote] val runSwitch = new Switch()
@ -54,18 +52,13 @@ abstract class RemoteClient private[akka] (
def shutdown(): Boolean
def isBoundTo(address: InetSocketAddress): Boolean = currentChannel.getRemoteAddress match {
case remoteAddress: InetSocketAddress ⇒
address.getAddress.getHostAddress == remoteAddress.getAddress.getHostAddress && address.getPort == remoteAddress.getPort
case _ ⇒ false
}
def isBoundTo(address: RemoteAddress): Boolean = remoteAddress == address
/**
* Converts the message to the wireprotocol and sends the message across the wire
*/
def send(message: Any, senderOption: Option[ActorRef], recipient: ActorRef) {
def send(message: Any, senderOption: Option[ActorRef], recipient: ActorRef): Unit =
send(remoteSupport.createRemoteMessageProtocolBuilder(Left(recipient), Right(message), senderOption).build)
}
/**
* Sends the message across the wire
@ -101,8 +94,7 @@ abstract class RemoteClient private[akka] (
class PassiveRemoteClient(val currentChannel: Channel,
remoteSupport: NettyRemoteSupport,
remoteAddress: InetSocketAddress,
val loader: Option[ClassLoader] = None)
remoteAddress: RemoteAddress)
extends RemoteClient(remoteSupport, remoteAddress) {
def connect(reconnectIfAlreadyConnected: Boolean = false): Boolean = runSwitch switchOn {
@ -125,7 +117,7 @@ class PassiveRemoteClient(val currentChannel: Channel,
*/
class ActiveRemoteClient private[akka] (
remoteSupport: NettyRemoteSupport,
remoteAddress: InetSocketAddress,
remoteAddress: RemoteAddress,
val loader: Option[ClassLoader] = None)
extends RemoteClient(remoteSupport, remoteAddress) {
@ -167,7 +159,7 @@ class ActiveRemoteClient private[akka] (
def attemptReconnect(): Boolean = {
log.debug("Remote client reconnecting to [{}]", remoteAddress)
val connection = bootstrap.connect(remoteAddress)
val connection = bootstrap.connect(new InetSocketAddress(remoteAddress.hostname, remoteAddress.port))
openChannels.add(connection.awaitUninterruptibly.getChannel) // Wait until the connection attempt succeeds or fails.
if (!connection.isSuccess) {
@ -190,7 +182,7 @@ class ActiveRemoteClient private[akka] (
log.debug("Starting remote client connection to [{}]", remoteAddress)
connection = bootstrap.connect(remoteAddress)
connection = bootstrap.connect(new InetSocketAddress(remoteAddress.hostname, remoteAddress.port))
val channel = connection.awaitUninterruptibly.getChannel
openChannels.add(channel)
@ -253,7 +245,7 @@ class ActiveRemoteClient private[akka] (
class ActiveRemoteClientPipelineFactory(
name: String,
bootstrap: ClientBootstrap,
remoteAddress: InetSocketAddress,
remoteAddress: RemoteAddress,
timer: HashedWheelTimer,
client: ActiveRemoteClient) extends ChannelPipelineFactory {
@ -278,7 +270,7 @@ class ActiveRemoteClientPipelineFactory(
class ActiveRemoteClientHandler(
val name: String,
val bootstrap: ClientBootstrap,
val remoteAddress: InetSocketAddress,
val remoteAddress: RemoteAddress,
val timer: HashedWheelTimer,
val client: ActiveRemoteClient)
extends SimpleChannelUpstreamHandler {
@ -366,26 +358,25 @@ class NettyRemoteSupport(_app: AkkaApplication) extends RemoteSupport(_app) with
protected[akka] def send(message: Any,
senderOption: Option[ActorRef],
recipientAddress: InetSocketAddress,
recipientAddress: RemoteAddress,
recipient: ActorRef,
loader: Option[ClassLoader]): Unit = {
val key = RemoteAddress(recipientAddress)
clientsLock.readLock.lock
try {
val client = remoteClients.get(key) match {
val client = remoteClients.get(recipientAddress) match {
case Some(client) ⇒ client
case None ⇒
clientsLock.readLock.unlock
clientsLock.writeLock.lock //Lock upgrade, not supported natively
try {
try {
remoteClients.get(key) match {
remoteClients.get(recipientAddress) match {
//Recheck for addition, race between upgrades
case Some(client) ⇒ client //If already populated by other writer
case None ⇒ //Populate map
val client = new ActiveRemoteClient(this, recipientAddress, loader)
client.connect()
remoteClients += key -> client
remoteClients += recipientAddress -> client
client
}
} finally {
@ -401,31 +392,30 @@ class NettyRemoteSupport(_app: AkkaApplication) extends RemoteSupport(_app) with
}
}
def bindClient(inetAddress: InetSocketAddress, client: RemoteClient, putIfAbsent: Boolean = false): Boolean = clientsLock withWriteGuard {
val address = RemoteAddress(inetAddress)
if (putIfAbsent && remoteClients.contains(address)) false
def bindClient(remoteAddress: RemoteAddress, client: RemoteClient, putIfAbsent: Boolean = false): Boolean = clientsLock withWriteGuard {
if (putIfAbsent && remoteClients.contains(remoteAddress)) false
else {
client.connect()
remoteClients.put(address, client).foreach(_.shutdown())
remoteClients.put(remoteAddress, client).foreach(_.shutdown())
true
}
}
def unbindClient(inetAddress: InetSocketAddress): Unit = clientsLock withWriteGuard {
def unbindClient(remoteAddress: RemoteAddress): Unit = clientsLock withWriteGuard {
remoteClients.foreach {
case (k, v) ⇒ if (v.isBoundTo(inetAddress)) { v.shutdown(); remoteClients.remove(k) }
case (k, v) ⇒ if (v.isBoundTo(remoteAddress)) { v.shutdown(); remoteClients.remove(k) }
}
}
def shutdownClientConnection(address: InetSocketAddress): Boolean = clientsLock withWriteGuard {
remoteClients.remove(RemoteAddress(address)) match {
def shutdownClientConnection(remoteAddress: RemoteAddress): Boolean = clientsLock withWriteGuard {
remoteClients.remove(remoteAddress) match {
case Some(client) ⇒ client.shutdown()
case None ⇒ false
}
}
def restartClientConnection(address: InetSocketAddress): Boolean = clientsLock withReadGuard {
remoteClients.get(RemoteAddress(address)) match {
def restartClientConnection(remoteAddress: RemoteAddress): Boolean = clientsLock withReadGuard {
remoteClients.get(remoteAddress) match {
case Some(client) ⇒ client.connect(reconnectIfAlreadyConnected = true)
case None ⇒ false
}
@ -469,8 +459,9 @@ class NettyRemoteSupport(_app: AkkaApplication) extends RemoteSupport(_app) with
class NettyRemoteServer(val remoteSupport: NettyRemoteSupport, val loader: Option[ClassLoader]) {
val log = Logging(remoteSupport.app, this)
import remoteSupport.serverSettings._
import remoteSupport.app.defaultAddress
val name = "NettyRemoteServer@" + remoteSupport.app.hostname + ":" + remoteSupport.app.port
val name = "NettyRemoteServer@" + defaultAddress
private val factory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool, Executors.newCachedThreadPool)
@ -487,7 +478,7 @@ class NettyRemoteServer(val remoteSupport: NettyRemoteSupport, val loader: Optio
bootstrap.setOption("child.reuseAddress", true)
bootstrap.setOption("child.connectTimeoutMillis", CONNECTION_TIMEOUT.toMillis)
openChannels.add(bootstrap.bind(remoteSupport.app.defaultAddress))
openChannels.add(bootstrap.bind(new InetSocketAddress(defaultAddress.hostname, defaultAddress.port)))
remoteSupport.notifyListeners(RemoteServerStarted(remoteSupport))
def shutdown() {
@ -620,8 +611,8 @@ class RemoteServerHandler(
instruction.getCommandType match {
case CommandType.CONNECT if USE_PASSIVE_CONNECTIONS ⇒
val origin = instruction.getOrigin
val inbound = new InetSocketAddress(origin.getHostname, origin.getPort)
val client = new PassiveRemoteClient(event.getChannel, remoteSupport, inbound, applicationLoader)
val inbound = RemoteAddress(origin.getHostname, origin.getPort)
val client = new PassiveRemoteClient(event.getChannel, remoteSupport, inbound)
remoteSupport.bindClient(inbound, client)
case CommandType.SHUTDOWN ⇒ //TODO FIXME Dispose passive connection here
case _ ⇒ //Unknown command
@ -637,9 +628,9 @@ class RemoteServerHandler(
event.getChannel.close
}
private def getClientAddress(c: Channel): Option[InetSocketAddress] =
private def getClientAddress(c: Channel): Option[RemoteAddress] =
c.getRemoteAddress match {
case inet: InetSocketAddress ⇒ Some(inet)
case inet: InetSocketAddress ⇒ Some(RemoteAddress(inet))
case _ ⇒ None
}
}

5
akka-remote/src/test/scala/akka/remote/AccrualFailureDetectorSpec.scala
View file

@ -8,10 +8,10 @@ import java.net.InetSocketAddress
class AccrualFailureDetectorSpec extends WordSpec with MustMatchers {
"An AccrualFailureDetector" should {
val conn = RemoteAddress(new InetSocketAddress("localhost", 2552))
"mark node as available after a series of successful heartbeats" in {
val fd = new AccrualFailureDetector
val conn = new InetSocketAddress("localhost", 2552)
fd.heartbeat(conn)
@ -27,7 +27,6 @@ class AccrualFailureDetectorSpec extends WordSpec with MustMatchers {
// FIXME how should we deal with explicit removal of connection? - if triggered as failure then we have a problem in boostrap - see line 142 in AccrualFailureDetector
"mark node as dead after explicit removal of connection" ignore {
val fd = new AccrualFailureDetector
val conn = new InetSocketAddress("localhost", 2552)
fd.heartbeat(conn)
@ -46,7 +45,6 @@ class AccrualFailureDetectorSpec extends WordSpec with MustMatchers {
"mark node as dead if heartbeat are missed" in {
val fd = new AccrualFailureDetector(threshold = 3)
val conn = new InetSocketAddress("localhost", 2552)
fd.heartbeat(conn)
@ -65,7 +63,6 @@ class AccrualFailureDetectorSpec extends WordSpec with MustMatchers {
"mark node as available if it starts heartbeat again after being marked dead due to detection of failure" in {
val fd = new AccrualFailureDetector(threshold = 3)
val conn = new InetSocketAddress("localhost", 2552)
fd.heartbeat(conn)