Skip to content

Commit

Permalink
Merge pull request corda#2477 from corda/aslemmer-corda/issues/2300
Browse files Browse the repository at this point in the history 
Add RPC deduplication to client and server
  • Loading branch information
@exFalso
exFalso authored Feb 19, 2018
2 parents c704ff6 + 32bcf0a commit dc268eb
Show file tree
Hide file tree
Showing 8 changed files with 376 additions and 221 deletions.
83 changes: 79 additions & 4 deletions client/rpc/src/integration-test/kotlin/net/corda/client/rpc/RPCStabilityTests.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ import net.corda.core.utilities.*
import net.corda.node.services.messaging.RPCServerConfiguration
import net.corda.nodeapi.RPCApi
import net.corda.testing.core.SerializationEnvironmentRule
import net.corda.testing.internal.*
import net.corda.testing.internal.testThreadFactory
import net.corda.testing.node.internal.*
import org.apache.activemq.artemis.api.core.SimpleString
import org.junit.After
Expand All @@ -30,6 +30,7 @@ import java.util.concurrent.Executors
import java.util.concurrent.ScheduledExecutorService
import java.util.concurrent.TimeUnit
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicLong

class RPCStabilityTests {
@Rule
Expand Down Expand Up @@ -127,7 +128,7 @@ class RPCStabilityTests {
rpcDriver {
fun startAndCloseServer(broker: RpcBrokerHandle) {
startRpcServerWithBrokerRunning(
configuration = RPCServerConfiguration.default.copy(consumerPoolSize = 1, producerPoolBound = 1),
configuration = RPCServerConfiguration.default,
ops = DummyOps,
brokerHandle = broker
).rpcServer.close()
Expand All @@ -148,7 +149,7 @@ class RPCStabilityTests {
@Test
fun `rpc client close doesnt leak broker resources`() {
rpcDriver {
val server = startRpcServer(configuration = RPCServerConfiguration.default.copy(consumerPoolSize = 1, producerPoolBound = 1), ops = DummyOps).get()
val server = startRpcServer(configuration = RPCServerConfiguration.default, ops = DummyOps).get()
RPCClient<RPCOps>(server.broker.hostAndPort!!).start(RPCOps::class.java, rpcTestUser.username, rpcTestUser.password).close()
val initial = server.broker.getStats()
repeat(100) {
Expand Down Expand Up @@ -337,11 +338,12 @@ class RPCStabilityTests {
val request = RPCApi.ClientToServer.RpcRequest(
clientAddress = SimpleString(myQueue),
methodName = SlowConsumerRPCOps::streamAtInterval.name,
serialisedArguments = listOf(10.millis, 123456).serialize(context = SerializationDefaults.RPC_SERVER_CONTEXT).bytes,
serialisedArguments = listOf(10.millis, 123456).serialize(context = SerializationDefaults.RPC_SERVER_CONTEXT),
replyId = Trace.InvocationId.newInstance(),
sessionId = Trace.SessionId.newInstance()
)
request.writeToClientMessage(message)
message.putLongProperty(RPCApi.DEDUPLICATION_SEQUENCE_NUMBER_FIELD_NAME, 0)
producer.send(message)
session.commit()

Expand All @@ -350,6 +352,79 @@ class RPCStabilityTests {
}
}

@Test
fun `deduplication in the server`() {
rpcDriver {
val server = startRpcServer(ops = SlowConsumerRPCOpsImpl()).getOrThrow()

// Construct an RPC client session manually
val myQueue = "${RPCApi.RPC_CLIENT_QUEUE_NAME_PREFIX}.test.${random63BitValue()}"
val session = startArtemisSession(server.broker.hostAndPort!!)
session.createTemporaryQueue(myQueue, myQueue)
val consumer = session.createConsumer(myQueue, null, -1, -1, false)
val replies = ArrayList<Any>()
consumer.setMessageHandler {
replies.add(it)
it.acknowledge()
}

val producer = session.createProducer(RPCApi.RPC_SERVER_QUEUE_NAME)
session.start()

pollUntilClientNumber(server, 1)

val message = session.createMessage(false)
val request = RPCApi.ClientToServer.RpcRequest(
clientAddress = SimpleString(myQueue),
methodName = DummyOps::protocolVersion.name,
serialisedArguments = emptyList<Any>().serialize(context = SerializationDefaults.RPC_SERVER_CONTEXT),
replyId = Trace.InvocationId.newInstance(),
sessionId = Trace.SessionId.newInstance()
)
request.writeToClientMessage(message)
message.putLongProperty(RPCApi.DEDUPLICATION_SEQUENCE_NUMBER_FIELD_NAME, 0)
producer.send(message)
// duplicate the message
producer.send(message)

pollUntilTrue("Number of replies is 1") {
replies.size == 1
}.getOrThrow()
}
}

@Test
fun `deduplication in the client`() {
rpcDriver {
val broker = startRpcBroker().getOrThrow()

// Construct an RPC server session manually
val session = startArtemisSession(broker.hostAndPort!!)
val consumer = session.createConsumer(RPCApi.RPC_SERVER_QUEUE_NAME)
val producer = session.createProducer()
val dedupeId = AtomicLong(0)
consumer.setMessageHandler {
it.acknowledge()
val request = RPCApi.ClientToServer.fromClientMessage(it)
when (request) {
is RPCApi.ClientToServer.RpcRequest -> {
val reply = RPCApi.ServerToClient.RpcReply(request.replyId, Try.Success(0), "server")
val message = session.createMessage(false)
reply.writeToClientMessage(SerializationDefaults.RPC_SERVER_CONTEXT, message)
message.putLongProperty(RPCApi.DEDUPLICATION_SEQUENCE_NUMBER_FIELD_NAME, dedupeId.getAndIncrement())
producer.send(request.clientAddress, message)
// duplicate the reply
producer.send(request.clientAddress, message)
}
is RPCApi.ClientToServer.ObservablesClosed -> {
}
}
}
session.start()

startRpcClient<RPCOps>(broker.hostAndPort!!).getOrThrow()
}
}
}

fun RPCDriverDSL.pollUntilClientNumber(server: RpcServerHandle, expected: Int) {
Expand Down
11 changes: 6 additions & 5 deletions client/rpc/src/main/kotlin/net/corda/client/rpc/internal/RPCClient.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -44,8 +44,6 @@ data class RPCClientConfiguration(
val reapInterval: Duration,
/** The number of threads to use for observations (for executing [Observable.onNext]) */
val observationExecutorPoolSize: Int,
/** The maximum number of producers to create to handle outgoing messages */
val producerPoolBound: Int,
/**
* Determines the concurrency level of the Observable Cache. This is exposed because it implicitly determines
* the limit on the number of leaked observables reaped because of garbage collection per reaping.
Expand All @@ -58,9 +56,12 @@ data class RPCClientConfiguration(
val connectionRetryIntervalMultiplier: Double,
/** Maximum retry interval */
val connectionMaxRetryInterval: Duration,
/** Maximum reconnect attempts on failover */
val maxReconnectAttempts: Int,
/** Maximum file size */
val maxFileSize: Int
val maxFileSize: Int,
/** The cache expiry of a deduplication watermark per client. */
val deduplicationCacheExpiry: Duration
) {
companion object {
val unlimitedReconnectAttempts = -1
Expand All @@ -70,14 +71,14 @@ data class RPCClientConfiguration(
trackRpcCallSites = false,
reapInterval = 1.seconds,
observationExecutorPoolSize = 4,
producerPoolBound = 1,
cacheConcurrencyLevel = 8,
connectionRetryInterval = 5.seconds,
connectionRetryIntervalMultiplier = 1.5,
connectionMaxRetryInterval = 3.minutes,
maxReconnectAttempts = unlimitedReconnectAttempts,
/** 10 MiB maximum allowed file size for attachments, including message headers. TODO: acquire this value from Network Map when supported. */
maxFileSize = 10485760
maxFileSize = 10485760,
deduplicationCacheExpiry = 1.days
)
}
}
Expand Down
111 changes: 56 additions & 55 deletions client/rpc/src/main/kotlin/net/corda/client/rpc/internal/RPCClientProxyHandler.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@ import net.corda.client.rpc.RPCSinceVersion
import net.corda.core.context.Actor
import net.corda.core.context.Trace
import net.corda.core.context.Trace.InvocationId
import net.corda.core.internal.LazyPool
import net.corda.core.internal.LazyStickyPool
import net.corda.core.internal.LifeCycle
import net.corda.core.internal.ThreadBox
Expand All @@ -26,14 +25,12 @@ import net.corda.core.utilities.Try
import net.corda.core.utilities.contextLogger
import net.corda.core.utilities.debug
import net.corda.core.utilities.getOrThrow
import net.corda.nodeapi.ArtemisConsumer
import net.corda.nodeapi.ArtemisProducer
import net.corda.nodeapi.RPCApi
import net.corda.nodeapi.internal.DeduplicationChecker
import org.apache.activemq.artemis.api.core.RoutingType
import org.apache.activemq.artemis.api.core.SimpleString
import org.apache.activemq.artemis.api.core.client.*
import org.apache.activemq.artemis.api.core.client.ActiveMQClient.DEFAULT_ACK_BATCH_SIZE
import org.apache.activemq.artemis.api.core.client.ClientMessage
import org.apache.activemq.artemis.api.core.client.ServerLocator
import rx.Notification
import rx.Observable
import rx.subjects.UnicastSubject
Expand All @@ -43,6 +40,7 @@ import java.time.Instant
import java.util.*
import java.util.concurrent.*
import java.util.concurrent.atomic.AtomicInteger
import java.util.concurrent.atomic.AtomicLong
import kotlin.reflect.jvm.javaMethod

/**
Expand Down Expand Up @@ -111,6 +109,8 @@ class RPCClientProxyHandler(

// Used for reaping
private var reaperExecutor: ScheduledExecutorService? = null
// Used for sending
private var sendExecutor: ExecutorService? = null

// A sticky pool for running Observable.onNext()s. We need the stickiness to preserve the observation ordering.
private val observationExecutorThreadFactory = ThreadFactoryBuilder().setNameFormat("rpc-client-observation-pool-%d").setDaemon(true).build()
Expand Down Expand Up @@ -161,22 +161,14 @@ class RPCClientProxyHandler(
build()
}

// We cannot pool consumers as we need to preserve the original muxed message order.
// TODO We may need to pool these somehow anyway, otherwise if the server sends many big messages in parallel a
// single consumer may be starved for flow control credits. Recheck this once Artemis's large message streaming is
// integrated properly.
private var sessionAndConsumer: ArtemisConsumer? = null
// Pool producers to reduce contention on the client side.
private val sessionAndProducerPool = LazyPool(bound = rpcConfiguration.producerPoolBound) {
// Note how we create new sessions *and* session factories per producer.
// We cannot simply pool producers on one session because sessions are single threaded.
// We cannot simply pool sessions on one session factory because flow control credits are tied to factories, so
// sessions tend to starve each other when used concurrently.
val sessionFactory = serverLocator.createSessionFactory()
val session = sessionFactory.createSession(rpcUsername, rpcPassword, false, true, true, false, DEFAULT_ACK_BATCH_SIZE)
session.start()
ArtemisProducer(sessionFactory, session, session.createProducer(RPCApi.RPC_SERVER_QUEUE_NAME))
}
private var sessionFactory: ClientSessionFactory? = null
private var producerSession: ClientSession? = null
private var consumerSession: ClientSession? = null
private var rpcProducer: ClientProducer? = null
private var rpcConsumer: ClientConsumer? = null

private val deduplicationChecker = DeduplicationChecker(rpcConfiguration.deduplicationCacheExpiry)
private val deduplicationSequenceNumber = AtomicLong(0)

/**
* Start the client. This creates the per-client queue, starts the consumer session and the reaper.
Expand All @@ -187,22 +179,25 @@ class RPCClientProxyHandler(
1,
ThreadFactoryBuilder().setNameFormat("rpc-client-reaper-%d").setDaemon(true).build()
)
sendExecutor = Executors.newSingleThreadExecutor(
ThreadFactoryBuilder().setNameFormat("rpc-client-sender-%d").build()
)
reaperScheduledFuture = reaperExecutor!!.scheduleAtFixedRate(
this::reapObservablesAndNotify,
rpcConfiguration.reapInterval.toMillis(),
rpcConfiguration.reapInterval.toMillis(),
TimeUnit.MILLISECONDS
)
sessionAndProducerPool.run {
it.session.createTemporaryQueue(clientAddress, RoutingType.ANYCAST, clientAddress)
}
val sessionFactory = serverLocator.createSessionFactory()
val session = sessionFactory.createSession(rpcUsername, rpcPassword, false, true, true, false, DEFAULT_ACK_BATCH_SIZE)
val consumer = session.createConsumer(clientAddress)
consumer.setMessageHandler(this@RPCClientProxyHandler::artemisMessageHandler)
sessionAndConsumer = ArtemisConsumer(sessionFactory, session, consumer)
sessionFactory = serverLocator.createSessionFactory()
producerSession = sessionFactory!!.createSession(rpcUsername, rpcPassword, false, true, true, false, DEFAULT_ACK_BATCH_SIZE)
rpcProducer = producerSession!!.createProducer(RPCApi.RPC_SERVER_QUEUE_NAME)
consumerSession = sessionFactory!!.createSession(rpcUsername, rpcPassword, false, true, true, false, DEFAULT_ACK_BATCH_SIZE)
consumerSession!!.createTemporaryQueue(clientAddress, RoutingType.ANYCAST, clientAddress)
rpcConsumer = consumerSession!!.createConsumer(clientAddress)
rpcConsumer!!.setMessageHandler(this::artemisMessageHandler)
lifeCycle.transition(State.UNSTARTED, State.SERVER_VERSION_NOT_SET)
session.start()
consumerSession!!.start()
producerSession!!.start()
}

// This is the general function that transforms a client side RPC to internal Artemis messages.
Expand All @@ -212,31 +207,28 @@ class RPCClientProxyHandler(
if (method == toStringMethod) {
return "Client RPC proxy for $rpcOpsClass"
}
if (sessionAndConsumer!!.session.isClosed) {
if (consumerSession!!.isClosed) {
throw RPCException("RPC Proxy is closed")
}

val replyId = InvocationId.newInstance()
callSiteMap?.set(replyId, Throwable("<Call site of root RPC '${method.name}'>"))
try {
val serialisedArguments = (arguments?.toList() ?: emptyList()).serialize(context = serializationContextWithObservableContext)
val request = RPCApi.ClientToServer.RpcRequest(clientAddress, method.name, serialisedArguments.bytes, replyId, sessionId, externalTrace, impersonatedActor)
val request = RPCApi.ClientToServer.RpcRequest(
clientAddress,
method.name,
serialisedArguments,
replyId,
sessionId,
externalTrace,
impersonatedActor
)
val replyFuture = SettableFuture.create<Any>()
sessionAndProducerPool.run {
val message = it.session.createMessage(false)
request.writeToClientMessage(message)

log.debug {
val argumentsString = arguments?.joinToString() ?: ""
"-> RPC(${replyId.value}) -> ${method.name}($argumentsString): ${method.returnType}"
}

require(rpcReplyMap.put(replyId, replyFuture) == null) {
"Generated several RPC requests with same ID $replyId"
}
it.producer.send(message)
it.session.commit()
require(rpcReplyMap.put(replyId, replyFuture) == null) {
"Generated several RPC requests with same ID $replyId"
}
sendMessage(request)
return replyFuture.getOrThrow()
} catch (e: RuntimeException) {
// Already an unchecked exception, so just rethrow it
Expand All @@ -249,9 +241,24 @@ class RPCClientProxyHandler(
}
}

private fun sendMessage(message: RPCApi.ClientToServer) {
val artemisMessage = producerSession!!.createMessage(false)
message.writeToClientMessage(artemisMessage)
sendExecutor!!.submit {
artemisMessage.putLongProperty(RPCApi.DEDUPLICATION_SEQUENCE_NUMBER_FIELD_NAME, deduplicationSequenceNumber.getAndIncrement())
log.debug { "-> RPC -> $message" }
rpcProducer!!.send(artemisMessage)
}
}

// The handler for Artemis messages.
private fun artemisMessageHandler(message: ClientMessage) {
val serverToClient = RPCApi.ServerToClient.fromClientMessage(serializationContextWithObservableContext, message)
val deduplicationSequenceNumber = message.getLongProperty(RPCApi.DEDUPLICATION_SEQUENCE_NUMBER_FIELD_NAME)
if (deduplicationChecker.checkDuplicateMessageId(serverToClient.deduplicationIdentity, deduplicationSequenceNumber)) {
log.info("Message duplication detected, discarding message")
return
}
log.debug { "Got message from RPC server $serverToClient" }
when (serverToClient) {
is RPCApi.ServerToClient.RpcReply -> {
Expand Down Expand Up @@ -325,14 +332,12 @@ class RPCClientProxyHandler(
* @param notify whether to notify observables or not.
*/
private fun close(notify: Boolean = true) {
sessionAndConsumer?.sessionFactory?.close()
sessionFactory?.close()
reaperScheduledFuture?.cancel(false)
observableContext.observableMap.invalidateAll()
reapObservables(notify)
reaperExecutor?.shutdownNow()
sessionAndProducerPool.close().forEach {
it.sessionFactory.close()
}
sendExecutor?.shutdownNow()
// Note the ordering is important, we shut down the consumer *before* the observation executor, otherwise we may
// leak borrowed executors.
val observationExecutors = observationExecutorPool.close()
Expand Down Expand Up @@ -385,11 +390,7 @@ class RPCClientProxyHandler(
}
if (observableIds != null) {
log.debug { "Reaping ${observableIds.size} observables" }
sessionAndProducerPool.run {
val message = it.session.createMessage(false)
RPCApi.ClientToServer.ObservablesClosed(observableIds).writeToClientMessage(message)
it.producer.send(message)
}
sendMessage(RPCApi.ClientToServer.ObservablesClosed(observableIds))
}
}
}
Expand Down
Loading

0 comments on commit dc268eb

Please sign in to comment.