More careful event dispatch (no longer global, but per endpoint)
This commit is contained in:
Robinson
parent
4b58a63dc1
commit
70825708a3
|
|
@ -422,10 +422,10 @@ open class Client<CONNECTION : Connection>(config: ClientConfiguration = ClientC
|
|||
|
||||
// on the client, we must GUARANTEE that the disconnect/close completes before NEW connect begins.
|
||||
// we will know this if we are running inside an INTERNAL dispatch that is NOT the connect dispatcher!
|
||||
val currentEvent = EventDispatcher.getCurrentEvent()
|
||||
if (currentEvent != null && currentEvent != EventDispatcher.CONNECT) {
|
||||
if (eventDispatch.isDispatch() && !eventDispatch.CONNECT.isDispatch()) {
|
||||
// only re-dispatch if we are on the event dispatch AND it's not the CONNECT one
|
||||
EventDispatcher.CONNECT.launch {
|
||||
// if we are on an "outside" thread, then we don't care.
|
||||
eventDispatch.CONNECT.launch {
|
||||
connect(
|
||||
remoteAddress = remoteAddress,
|
||||
remoteAddressString = remoteAddressString,
|
||||
|
|
|
|||
|
|
@ -283,12 +283,14 @@ open class Server<CONNECTION : Connection>(config: ServerConfiguration = ServerC
|
|||
if (mustRestartDriverOnError) {
|
||||
logger.error("Critical driver error detected, restarting server.")
|
||||
|
||||
EventDispatcher.launchSequentially(EventDispatcher.CONNECT) {
|
||||
eventDispatch.CLOSE.launch {
|
||||
waitForEndpointShutdown()
|
||||
|
||||
// also wait for everyone else to shutdown!!
|
||||
aeronDriver.internal.endPointUsages.forEach {
|
||||
it.waitForEndpointShutdown()
|
||||
if (it !== this@Server) {
|
||||
it.waitForEndpointShutdown()
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -20,7 +20,6 @@ import dorkbox.collections.ConcurrentIterator
|
|||
import dorkbox.collections.LockFreeHashSet
|
||||
import dorkbox.network.Configuration
|
||||
import dorkbox.network.connection.EndPoint
|
||||
import dorkbox.network.connection.EventDispatcher
|
||||
import dorkbox.network.connection.ListenerManager
|
||||
import dorkbox.network.connection.ListenerManager.Companion.cleanAllStackTrace
|
||||
import dorkbox.network.connection.ListenerManager.Companion.cleanStackTrace
|
||||
|
|
@ -196,14 +195,12 @@ internal class AeronDriverInternal(endPoint: EndPoint<*>?, config: Configuration
|
|||
// this must be set before anything else happens
|
||||
mustRestartDriverOnError = true
|
||||
|
||||
// we must close all the connections on a DIFFERENT thread!
|
||||
EventDispatcher.CLOSE.launch {
|
||||
endPointUsages.forEach {
|
||||
it.close(closeEverything = false,
|
||||
sendDisconnectMessage = false,
|
||||
notifyDisconnect = false,
|
||||
releaseWaitingThreads = false)
|
||||
}
|
||||
// close will make sure to run on a different thread
|
||||
endPointUsages.forEach {
|
||||
it.close(closeEverything = false,
|
||||
sendDisconnectMessage = false,
|
||||
notifyDisconnect = false,
|
||||
releaseWaitingThreads = false)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -20,7 +20,6 @@ import dorkbox.bytes.ByteArrayWrapper
|
|||
import dorkbox.collections.ConcurrentIterator
|
||||
import dorkbox.network.Configuration
|
||||
import dorkbox.network.connection.EndPoint
|
||||
import dorkbox.network.connection.EventDispatcher
|
||||
import dorkbox.util.NamedThreadFactory
|
||||
import kotlinx.atomicfu.atomic
|
||||
import org.agrona.concurrent.IdleStrategy
|
||||
|
|
@ -187,7 +186,7 @@ internal class EventPoller {
|
|||
fun close(logger: Logger, endPoint: EndPoint<*>) {
|
||||
// make sure that we close on the CLOSE dispatcher if we run on the poll dispatcher!
|
||||
if (isDispatch()) {
|
||||
EventDispatcher.CLOSE.launch {
|
||||
endPoint.eventDispatch.CLOSE.launch {
|
||||
close(logger, endPoint)
|
||||
}
|
||||
return
|
||||
|
|
|
|||
|
|
@ -337,13 +337,20 @@ open class Connection(connectionParameters: ConnectionParams<*>) {
|
|||
// the compareAndSet is used to make sure that if we call close() MANUALLY, (and later) when the auto-cleanup/disconnect is called -- it doesn't
|
||||
// try to do it again.
|
||||
|
||||
// make sure that EVERYTHING before "close()" runs before we do
|
||||
EventDispatcher.launchSequentially(EventDispatcher.CLOSE) {
|
||||
closeImmediately(
|
||||
sendDisconnectMessage = sendDisconnectMessage,
|
||||
notifyDisconnect = notifyDisconnect,
|
||||
closeEverything = closeEverything)
|
||||
// make sure that EVERYTHING before "close()" runs before we do.
|
||||
// If there are multiple clients/servers sharing the same NetworkPoller -- then they will wait on each other!
|
||||
val close = endPoint.eventDispatch.CLOSE
|
||||
if (!close.isDispatch()) {
|
||||
close.launch {
|
||||
close(sendDisconnectMessage, notifyDisconnect, closeEverything)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
closeImmediately(
|
||||
sendDisconnectMessage = sendDisconnectMessage,
|
||||
notifyDisconnect = notifyDisconnect,
|
||||
closeEverything = closeEverything)
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -100,6 +100,8 @@ abstract class EndPoint<CONNECTION : Connection> private constructor(val type: C
|
|||
|
||||
val logger: Logger = LoggerFactory.getLogger(loggerName)
|
||||
|
||||
internal val eventDispatch = EventDispatcher(loggerName)
|
||||
|
||||
private val handler = CoroutineExceptionHandler { _, exception ->
|
||||
logger.error("Uncaught Coroutine Error: ${exception.stackTraceToString()}")
|
||||
}
|
||||
|
|
@ -110,7 +112,7 @@ abstract class EndPoint<CONNECTION : Connection> private constructor(val type: C
|
|||
private val messageChannel = Channel<Paired<CONNECTION>>()
|
||||
private val pairedPool: Pool<Paired<CONNECTION>>
|
||||
|
||||
internal val listenerManager = ListenerManager<CONNECTION>(logger)
|
||||
internal val listenerManager = ListenerManager<CONNECTION>(logger, eventDispatch)
|
||||
|
||||
val connections = ConcurrentIterator<CONNECTION>()
|
||||
|
||||
|
|
|
|||
|
|
@ -23,110 +23,112 @@ import org.slf4j.LoggerFactory
|
|||
import java.util.concurrent.*
|
||||
|
||||
/**
|
||||
* Event logic throughout the network MUST be run on multiple threads! There are deadlock issues if it is only one.
|
||||
* Event logic throughout the network MUST be run on multiple threads! There are deadlock issues if it is only one, or if the client + server
|
||||
* share an event dispatcher (multiple network restarts were required to check this)
|
||||
*
|
||||
* WARNING: The logic in this class will ONLY work in this class, as it relies on this specific behavior. Do not use it elsewhere!
|
||||
*/
|
||||
enum class EventDispatcher {
|
||||
// NOTE: CLOSE must be last!
|
||||
HANDSHAKE, CONNECT, DISCONNECT, ERROR, CLOSE;
|
||||
internal class EventDispatcher(val type: String) {
|
||||
enum class EDType {
|
||||
// CLOSE must be last!
|
||||
CONNECT, ERROR, CLOSE
|
||||
}
|
||||
|
||||
companion object {
|
||||
private val DEBUG_EVENTS = false
|
||||
private val traceId = atomic(0)
|
||||
|
||||
private val logger = LoggerFactory.getLogger(EventDispatcher::class.java.simpleName)
|
||||
|
||||
private val threadIds = entries.map { atomic(0L) }.toTypedArray()
|
||||
|
||||
|
||||
private val executors = entries.map { event ->
|
||||
// It CANNOT be the default dispatch because there will be thread starvation
|
||||
// NOTE: THIS CANNOT CHANGE!! IT WILL BREAK EVERYTHING IF IT CHANGES!
|
||||
Executors.newSingleThreadExecutor(
|
||||
NamedThreadFactory("Event Dispatcher-${event.name}",
|
||||
Configuration.networkThreadGroup, Thread.NORM_PRIORITY, true) { thread ->
|
||||
// when a new thread is created, assign it to the array
|
||||
threadIds[event.ordinal].lazySet(thread.id)
|
||||
}
|
||||
)
|
||||
}.toTypedArray()
|
||||
|
||||
private val typedEntries: Array<EventDispatcher>
|
||||
private val typedEntries: Array<EDType>
|
||||
|
||||
init {
|
||||
executors.forEachIndexed { _, executor ->
|
||||
executor.submit {
|
||||
// this is to create a new thread only, so that the thread ID can be assigned
|
||||
}
|
||||
}
|
||||
typedEntries = EDType.entries.toTypedArray()
|
||||
}
|
||||
}
|
||||
|
||||
typedEntries = entries.toTypedArray()
|
||||
private val logger = LoggerFactory.getLogger("$type Dispatch")
|
||||
|
||||
private val threadIds = EDType.entries.map { atomic(0L) }.toTypedArray()
|
||||
|
||||
private val executors = EDType.entries.map { event ->
|
||||
// It CANNOT be the default dispatch because there will be thread starvation
|
||||
// NOTE: THIS CANNOT CHANGE!! IT WILL BREAK EVERYTHING IF IT CHANGES!
|
||||
Executors.newSingleThreadExecutor(
|
||||
NamedThreadFactory(
|
||||
namePrefix = "$type-${event.name}",
|
||||
group = Configuration.networkThreadGroup,
|
||||
threadPriority = Thread.NORM_PRIORITY,
|
||||
daemon = true
|
||||
) { thread ->
|
||||
// when a new thread is created, assign it to the array
|
||||
threadIds[event.ordinal].lazySet(thread.id)
|
||||
}
|
||||
)
|
||||
}.toTypedArray()
|
||||
|
||||
internal class ED(private val dispatcher: EventDispatcher, private val type: EDType) {
|
||||
fun launch(function: () -> Unit) {
|
||||
dispatcher.launch(type, function)
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if the current execution thread is running inside one of the event dispatchers listed.
|
||||
*
|
||||
* No values specified means we check ALL events
|
||||
*/
|
||||
fun isDispatch(): Boolean {
|
||||
return isCurrentEvent(*typedEntries)
|
||||
return dispatcher.isDispatch(type)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if the current execution thread is running inside one of the event dispatchers listed.
|
||||
*
|
||||
* No values specified means we check ALL events
|
||||
*/
|
||||
fun isCurrentEvent(vararg events: EventDispatcher = typedEntries): Boolean {
|
||||
val threadId = Thread.currentThread().id
|
||||
val CONNECT: ED
|
||||
val ERROR: ED
|
||||
val CLOSE: ED
|
||||
|
||||
events.forEach { event ->
|
||||
if (threadIds[event.ordinal].value == threadId) {
|
||||
return true
|
||||
}
|
||||
|
||||
init {
|
||||
executors.forEachIndexed { _, executor ->
|
||||
executor.submit {
|
||||
// this is to create a new thread only, so that the thread ID can be assigned
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if the current execution thread is NOT running inside one of the event dispatchers listed.
|
||||
*
|
||||
* No values specified means we check ALL events
|
||||
*/
|
||||
fun isNotCurrentEvent(vararg events: EventDispatcher = typedEntries): Boolean {
|
||||
val currentDispatch = getCurrentEvent() ?: return false
|
||||
CONNECT = ED(this, EDType.CONNECT)
|
||||
ERROR = ED(this, EDType.ERROR)
|
||||
CLOSE = ED(this, EDType.CLOSE)
|
||||
}
|
||||
|
||||
return events.contains(currentDispatch)
|
||||
}
|
||||
|
||||
/**
|
||||
* @return which event dispatch thread we are running in, if any
|
||||
*/
|
||||
fun getCurrentEvent(): EventDispatcher? {
|
||||
val threadId = Thread.currentThread().id
|
||||
/**
|
||||
* Checks if the current execution thread is running inside one of the event dispatchers.
|
||||
*/
|
||||
fun isDispatch(): Boolean {
|
||||
val threadId = Thread.currentThread().id
|
||||
|
||||
typedEntries.forEach { event ->
|
||||
if (threadIds[event.ordinal].value == threadId) {
|
||||
return event
|
||||
}
|
||||
typedEntries.forEach { event ->
|
||||
if (threadIds[event.ordinal].value == threadId) {
|
||||
return true
|
||||
}
|
||||
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Each event type runs inside its own coroutine dispatcher.
|
||||
*
|
||||
* We want EACH event type to run in its own dispatcher... on its OWN thread, in order to prevent deadlocks
|
||||
* This is because there are blocking dependencies: DISCONNECT -> CONNECT.
|
||||
*
|
||||
* If an event is RE-ENTRANT, then it will immediately execute!
|
||||
*/
|
||||
private fun launch(event: EventDispatcher, function: () -> Unit) {
|
||||
val eventId = event.ordinal
|
||||
return false
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if the current execution thread is running inside one of the event dispatchers.
|
||||
*/
|
||||
private fun isDispatch(type: EDType): Boolean {
|
||||
val threadId = Thread.currentThread().id
|
||||
|
||||
return threadIds[type.ordinal].value == threadId
|
||||
}
|
||||
|
||||
/**
|
||||
* Each event type runs inside its own coroutine dispatcher.
|
||||
*
|
||||
* We want EACH event type to run in its own dispatcher... on its OWN thread, in order to prevent deadlocks
|
||||
* This is because there are blocking dependencies: DISCONNECT -> CONNECT.
|
||||
*
|
||||
* If an event is RE-ENTRANT, then it will immediately execute!
|
||||
*/
|
||||
private fun launch(event: EDType, function: () -> Unit) {
|
||||
val eventId = event.ordinal
|
||||
|
||||
try {
|
||||
if (DEBUG_EVENTS) {
|
||||
val id = traceId.getAndIncrement()
|
||||
executors[eventId].submit {
|
||||
|
|
@ -141,31 +143,8 @@ enum class EventDispatcher {
|
|||
} else {
|
||||
executors[eventId].submit(function)
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
logger.error("Error during event dispatch!", e)
|
||||
}
|
||||
|
||||
|
||||
fun launchSequentially(endEvent: EventDispatcher, function: () -> Unit) {
|
||||
// If one of our callbacks requested a shutdown, we wait until all callbacks have run ... THEN shutdown
|
||||
val event = getCurrentEvent()
|
||||
|
||||
val index = event?.ordinal ?: -1
|
||||
|
||||
// This will loop through until it runs on the CLOSE EventDispatcher
|
||||
if (index < endEvent.ordinal) {
|
||||
// If this runs inside EVENT.CONNECT/DISCONNECT/ETC, we must ***WAIT*** until all listeners have been called!
|
||||
// this problem is solved by running AGAIN after we have finished running whatever event dispatcher we are currently on
|
||||
// MORE SPECIFICALLY, we must run at the end of our current one, but repeatedly until CLOSE
|
||||
|
||||
EventDispatcher.launch(typedEntries[index+1]) {
|
||||
launchSequentially(endEvent, function)
|
||||
}
|
||||
} else {
|
||||
endEvent.launch(function)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun launch(function: () -> Unit) {
|
||||
launch(this, function)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@ import kotlin.concurrent.write
|
|||
/**
|
||||
* Manages all of the different connect/disconnect/etc listeners
|
||||
*/
|
||||
internal class ListenerManager<CONNECTION: Connection>(private val logger: Logger) {
|
||||
internal class ListenerManager<CONNECTION: Connection>(private val logger: Logger, val eventDispatch: EventDispatcher) {
|
||||
companion object {
|
||||
/**
|
||||
* Specifies the load-factor for the IdentityMap used to manage keeping track of the number of connections + listeners
|
||||
|
|
@ -402,7 +402,7 @@ internal class ListenerManager<CONNECTION: Connection>(private val logger: Logge
|
|||
fun notifyConnect(connection: CONNECTION) {
|
||||
val list = onConnectList
|
||||
if (list.isNotEmpty()) {
|
||||
EventDispatcher.CONNECT.launch {
|
||||
connection.endPoint.eventDispatch.CONNECT.launch {
|
||||
list.forEach {
|
||||
try {
|
||||
it(connection)
|
||||
|
|
@ -436,7 +436,7 @@ internal class ListenerManager<CONNECTION: Connection>(private val logger: Logge
|
|||
fun directNotifyDisconnect(connection: CONNECTION) {
|
||||
val list = onDisconnectList
|
||||
if (list.isNotEmpty()) {
|
||||
EventDispatcher.DISCONNECT.launch {
|
||||
connection.endPoint.eventDispatch.CLOSE.launch {
|
||||
list.forEach {
|
||||
try {
|
||||
it(connection)
|
||||
|
|
@ -461,7 +461,7 @@ internal class ListenerManager<CONNECTION: Connection>(private val logger: Logge
|
|||
fun notifyError(connection: CONNECTION, exception: Throwable) {
|
||||
val list = onErrorList
|
||||
if (list.isNotEmpty()) {
|
||||
EventDispatcher.ERROR.launch {
|
||||
connection.endPoint.eventDispatch.ERROR.launch {
|
||||
list.forEach {
|
||||
try {
|
||||
it(connection, exception)
|
||||
|
|
@ -485,7 +485,7 @@ internal class ListenerManager<CONNECTION: Connection>(private val logger: Logge
|
|||
fun notifyError(exception: Throwable) {
|
||||
val list = onErrorGlobalList
|
||||
if (list.isNotEmpty()) {
|
||||
EventDispatcher.ERROR.launch {
|
||||
eventDispatch.ERROR.launch {
|
||||
list.forEach {
|
||||
try {
|
||||
it(exception)
|
||||
|
|
|
|||
|
|
@ -26,7 +26,6 @@ import dorkbox.network.aeron.AeronDriver
|
|||
import dorkbox.network.aeron.AeronDriver.Companion.uriHandshake
|
||||
import dorkbox.network.aeron.AeronPoller
|
||||
import dorkbox.network.connection.Connection
|
||||
import dorkbox.network.connection.EventDispatcher
|
||||
import dorkbox.network.connection.IpInfo
|
||||
import dorkbox.network.exceptions.ServerException
|
||||
import dorkbox.network.exceptions.ServerHandshakeException
|
||||
|
|
@ -121,72 +120,58 @@ internal object ServerHandshakePollers {
|
|||
}
|
||||
|
||||
// we have read all the data, now dispatch it.
|
||||
EventDispatcher.HANDSHAKE.launch {
|
||||
// HandshakeMessage.HELLO
|
||||
// HandshakeMessage.DONE
|
||||
val messageState = message.state
|
||||
val connectKey = message.connectKey
|
||||
// HandshakeMessage.HELLO
|
||||
// HandshakeMessage.DONE
|
||||
val messageState = message.state
|
||||
val connectKey = message.connectKey
|
||||
|
||||
if (messageState == HandshakeMessage.HELLO) {
|
||||
// we create a NEW publication for the handshake, which connects directly to the client handshake subscription
|
||||
if (messageState == HandshakeMessage.HELLO) {
|
||||
// we create a NEW publication for the handshake, which connects directly to the client handshake subscription
|
||||
|
||||
val publicationUri = uriHandshake(CommonContext.IPC_MEDIA, isReliable)
|
||||
val publicationUri = uriHandshake(CommonContext.IPC_MEDIA, isReliable)
|
||||
|
||||
// this will always connect to the CLIENT handshake subscription!
|
||||
val publication = try {
|
||||
driver.addExclusivePublication(publicationUri, message.streamId, logInfo, true)
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
// this will always connect to the CLIENT handshake subscription!
|
||||
val publication = try {
|
||||
driver.addExclusivePublication(publicationUri, message.streamId, logInfo, true)
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Cannot create IPC publication back to client remote process", e))
|
||||
return@launch
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Cannot create IPC publication back to client remote process", e))
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
// we actually have to wait for it to connect before we continue
|
||||
driver.waitForConnection(publication, handshakeTimeoutNs, logInfo) { cause ->
|
||||
ServerTimedoutException("$logInfo publication cannot connect with client in ${Sys.getTimePrettyFull(handshakeTimeoutNs)}", cause)
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
try {
|
||||
// we actually have to wait for it to connect before we continue
|
||||
driver.waitForConnection(publication, handshakeTimeoutNs, logInfo) { cause ->
|
||||
ServerTimedoutException("$logInfo publication cannot connect with client in ${Sys.getTimePrettyFull(handshakeTimeoutNs)}", cause)
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Cannot create IPC publication back to client remote process", e))
|
||||
return@launch
|
||||
}
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Cannot create IPC publication back to client remote process", e))
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
try {
|
||||
val success = handshake.processIpcHandshakeMessageServer(
|
||||
server = server,
|
||||
handshaker = handshaker,
|
||||
aeronDriver = driver,
|
||||
handshakePublication = publication,
|
||||
publicKey = message.publicKey!!,
|
||||
message = message,
|
||||
logInfo = logInfo,
|
||||
logger = logger
|
||||
)
|
||||
|
||||
if (success) {
|
||||
publications[connectKey] = publication
|
||||
} else {
|
||||
try {
|
||||
// we might not be able to close this connection.
|
||||
driver.close(publication, logInfo)
|
||||
}
|
||||
catch (e: Exception) {
|
||||
server.listenerManager.notifyError(e)
|
||||
}
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
try {
|
||||
val success = handshake.processIpcHandshakeMessageServer(
|
||||
server = server,
|
||||
handshaker = handshaker,
|
||||
aeronDriver = driver,
|
||||
handshakePublication = publication,
|
||||
publicKey = message.publicKey!!,
|
||||
message = message,
|
||||
logInfo = logInfo,
|
||||
logger = logger
|
||||
)
|
||||
|
||||
if (success) {
|
||||
publications[connectKey] = publication
|
||||
} else {
|
||||
try {
|
||||
// we might not be able to close this connection.
|
||||
driver.close(publication, logInfo)
|
||||
|
|
@ -194,35 +179,8 @@ internal object ServerHandshakePollers {
|
|||
catch (e: Exception) {
|
||||
server.listenerManager.notifyError(e)
|
||||
}
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Error processing IPC handshake", e))
|
||||
}
|
||||
} else {
|
||||
// HandshakeMessage.DONE
|
||||
|
||||
val publication = publications.remove(connectKey)
|
||||
if (publication == null) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] No publication back to IPC"))
|
||||
return@launch
|
||||
}
|
||||
|
||||
try {
|
||||
handshake.validateMessageTypeAndDoPending(
|
||||
server = server,
|
||||
handshaker = handshaker,
|
||||
handshakePublication = publication,
|
||||
message = message,
|
||||
logInfo = logInfo,
|
||||
logger = logger
|
||||
)
|
||||
} catch (e: Exception) {
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Error processing IPC handshake", e))
|
||||
}
|
||||
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
|
@ -234,6 +192,45 @@ internal object ServerHandshakePollers {
|
|||
catch (e: Exception) {
|
||||
server.listenerManager.notifyError(e)
|
||||
}
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Error processing IPC handshake", e))
|
||||
}
|
||||
} else {
|
||||
// HandshakeMessage.DONE
|
||||
|
||||
val publication = publications.remove(connectKey)
|
||||
if (publication == null) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] No publication back to IPC"))
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
handshake.validateMessageTypeAndDoPending(
|
||||
server = server,
|
||||
handshaker = handshaker,
|
||||
handshakePublication = publication,
|
||||
message = message,
|
||||
logInfo = logInfo,
|
||||
logger = logger
|
||||
)
|
||||
} catch (e: Exception) {
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Error processing IPC handshake", e))
|
||||
}
|
||||
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
try {
|
||||
// we might not be able to close this connection.
|
||||
driver.close(publication, logInfo)
|
||||
}
|
||||
catch (e: Exception) {
|
||||
server.listenerManager.notifyError(e)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -365,86 +362,71 @@ internal object ServerHandshakePollers {
|
|||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
|
||||
EventDispatcher.HANDSHAKE.launch {
|
||||
// HandshakeMessage.HELLO
|
||||
// HandshakeMessage.DONE
|
||||
val messageState = message.state
|
||||
val connectKey = message.connectKey
|
||||
// HandshakeMessage.HELLO
|
||||
// HandshakeMessage.DONE
|
||||
val messageState = message.state
|
||||
val connectKey = message.connectKey
|
||||
|
||||
if (messageState == HandshakeMessage.HELLO) {
|
||||
// we create a NEW publication for the handshake, which connects directly to the client handshake subscription
|
||||
if (messageState == HandshakeMessage.HELLO) {
|
||||
// we create a NEW publication for the handshake, which connects directly to the client handshake subscription
|
||||
|
||||
// we explicitly have the publisher "connect to itself", because we are using MDC to work around NAT.
|
||||
// It will "auto-connect" to the correct client port (negotiated by the MDC client subscription negotiating on the
|
||||
// control port of the server)
|
||||
val publicationUri = uriHandshake(CommonContext.UDP_MEDIA, isReliable)
|
||||
.controlEndpoint(ipInfo.getAeronPubAddress(isRemoteIpv4) + ":" + mdcPortPub)
|
||||
// we explicitly have the publisher "connect to itself", because we are using MDC to work around NAT.
|
||||
// It will "auto-connect" to the correct client port (negotiated by the MDC client subscription negotiating on the
|
||||
// control port of the server)
|
||||
val publicationUri = uriHandshake(CommonContext.UDP_MEDIA, isReliable)
|
||||
.controlEndpoint(ipInfo.getAeronPubAddress(isRemoteIpv4) + ":" + mdcPortPub)
|
||||
|
||||
|
||||
// this will always connect to the CLIENT handshake subscription!
|
||||
val publication = try {
|
||||
driver.addExclusivePublication(publicationUri, message.streamId, logInfo, false)
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
// this will always connect to the CLIENT handshake subscription!
|
||||
val publication = try {
|
||||
driver.addExclusivePublication(publicationUri, message.streamId, logInfo, false)
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Cannot create publication back to $clientAddressString", e))
|
||||
return@launch
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Cannot create publication back to $clientAddressString", e))
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
// we actually have to wait for it to connect before we continue
|
||||
driver.waitForConnection(publication, handshakeTimeoutNs, logInfo) { cause ->
|
||||
ServerTimedoutException("$logInfo publication cannot connect with client in ${Sys.getTimePrettyFull(handshakeTimeoutNs)}", cause)
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
try {
|
||||
// we actually have to wait for it to connect before we continue
|
||||
driver.waitForConnection(publication, handshakeTimeoutNs, logInfo) { cause ->
|
||||
ServerTimedoutException("$logInfo publication cannot connect with client in ${Sys.getTimePrettyFull(handshakeTimeoutNs)}", cause)
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Cannot create publication back to $clientAddressString", e))
|
||||
return
|
||||
}
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Cannot create publication back to $clientAddressString", e))
|
||||
return@launch
|
||||
}
|
||||
try {
|
||||
val success = handshake.processUdpHandshakeMessageServer(
|
||||
server = server,
|
||||
handshaker = handshaker,
|
||||
handshakePublication = publication,
|
||||
publicKey = message.publicKey!!,
|
||||
clientAddress = clientAddress,
|
||||
clientAddressString = clientAddressString,
|
||||
portPub = message.port,
|
||||
portSub = serverPortSub,
|
||||
mdcPortPub = mdcPortPub,
|
||||
isReliable = isReliable,
|
||||
message = message,
|
||||
logInfo = logInfo,
|
||||
logger = logger
|
||||
)
|
||||
|
||||
try {
|
||||
val success = handshake.processUdpHandshakeMessageServer(
|
||||
server = server,
|
||||
handshaker = handshaker,
|
||||
handshakePublication = publication,
|
||||
publicKey = message.publicKey!!,
|
||||
clientAddress = clientAddress,
|
||||
clientAddressString = clientAddressString,
|
||||
portPub = message.port,
|
||||
portSub = serverPortSub,
|
||||
mdcPortPub = mdcPortPub,
|
||||
isReliable = isReliable,
|
||||
message = message,
|
||||
logInfo = logInfo,
|
||||
logger = logger
|
||||
)
|
||||
|
||||
if (success) {
|
||||
publications[connectKey] = publication
|
||||
} else {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
try {
|
||||
// we might not be able to close this connection.
|
||||
driver.close(publication, logInfo)
|
||||
}
|
||||
catch (e: Exception) {
|
||||
server.listenerManager.notifyError(e)
|
||||
}
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
if (success) {
|
||||
publications[connectKey] = publication
|
||||
} else {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
|
@ -454,50 +436,62 @@ internal object ServerHandshakePollers {
|
|||
driver.close(publication, logInfo)
|
||||
}
|
||||
catch (e: Exception) {
|
||||
driver.close(publication, logInfo)
|
||||
server.listenerManager.notifyError(e)
|
||||
}
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Error processing IPC handshake", e))
|
||||
}
|
||||
} else {
|
||||
// HandshakeMessage.DONE
|
||||
|
||||
val publication = publications.remove(connectKey)
|
||||
|
||||
if (publication == null) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] No publication back to $clientAddressString"))
|
||||
return@launch
|
||||
}
|
||||
|
||||
try {
|
||||
handshake.validateMessageTypeAndDoPending(
|
||||
server = server,
|
||||
handshaker = handshaker,
|
||||
handshakePublication = publication,
|
||||
message = message,
|
||||
logInfo = logInfo,
|
||||
logger = logger
|
||||
)
|
||||
} catch (e: Exception) {
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Error processing IPC handshake", e))
|
||||
}
|
||||
} catch (e: Exception) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
try {
|
||||
// we might not be able to close this connection.
|
||||
driver.close(publication, logInfo)
|
||||
}
|
||||
catch (e: Exception) {
|
||||
server.listenerManager.notifyError(e)
|
||||
driver.close(publication, logInfo)
|
||||
}
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Error processing IPC handshake", e))
|
||||
}
|
||||
} else {
|
||||
// HandshakeMessage.DONE
|
||||
|
||||
val publication = publications.remove(connectKey)
|
||||
|
||||
if (publication == null) {
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] No publication back to $clientAddressString"))
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
handshake.validateMessageTypeAndDoPending(
|
||||
server = server,
|
||||
handshaker = handshaker,
|
||||
handshakePublication = publication,
|
||||
message = message,
|
||||
logInfo = logInfo,
|
||||
logger = logger
|
||||
)
|
||||
} catch (e: Exception) {
|
||||
server.listenerManager.notifyError(ServerHandshakeException("[$logInfo] Error processing IPC handshake", e))
|
||||
}
|
||||
|
||||
try {
|
||||
// we might not be able to close this connection.
|
||||
driver.close(publication, logInfo)
|
||||
}
|
||||
catch (e: Exception) {
|
||||
server.listenerManager.notifyError(e)
|
||||
}
|
||||
|
||||
// we should immediately remove the logbuffer for this! Aeron will **EVENTUALLY** remove the logbuffer, but if errors
|
||||
// and connections occur too quickly (within the cleanup/linger period), we can run out of memory!
|
||||
driver.deleteLogFile(image)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue