SI-5734 Allow setting of socket timeout for remote actors

TcpService is locked while connections are being established, preventing
messages being sent to remote nodes. There was no way to specify a connect
timeout, so the service was waiting for a ConnectException to be thrown if
the remote node was unavailable (which can take several minutes).

Added a system property to allow setting a socket connect timeout to
prevent the TcpService from being locked for minutes at a time.

Default behaviour is unchanged, all tests pass and the patch has been
running in production without issue.

1 files changed, 27 insertions, 2 deletions

diff --git a/src/actors/scala/actors/remote/TcpService.scala b/src/actors/scala/actors/remote/TcpService.scala
index bde05fd816..028dd3a083 100644
--- a/src/actors/scala/actors/remote/TcpService.scala
+++ b/src/actors/scala/actors/remote/TcpService.scala
@@ -14,7 +14,7 @@ package remote
 
 import java.io.{DataInputStream, DataOutputStream, IOException}
 import java.lang.{Thread, SecurityException}
-import java.net.{InetAddress, ServerSocket, Socket, UnknownHostException}
+import java.net.{InetAddress, InetSocketAddress, ServerSocket, Socket, SocketTimeoutException, UnknownHostException}
 
 import scala.collection.mutable
 import scala.util.Random
@@ -59,6 +59,23 @@ object TcpService {
     portnum
   }
 
+  private val connectTimeoutMillis = {
+    val propName = "scala.actors.tcpSocket.connectTimeoutMillis"
+    val defaultTimeoutMillis = 0
+    sys.props get propName flatMap {
+      timeout =>
+        try {
+          val to = timeout.toInt
+          Debug.info("Using socket timeout $to")
+          Some(to)
+        } catch {
+          case e: NumberFormatException =>
+            Debug.warning(s"""Could not parse $propName = "$timeout" as an Int""")
+            None
+        }
+    } getOrElse defaultTimeoutMillis
+  }
+
   var BufSize: Int = 65536
 }
 
@@ -176,7 +193,15 @@ class TcpService(port: Int, cl: ClassLoader) extends Thread with Service {
   }
 
   def connect(n: Node): TcpServiceWorker = synchronized {
-    val socket = new Socket(n.address, n.port)
+    val socket = new Socket()
+    val start = System.nanoTime
+    try {
+      socket.connect(new InetSocketAddress(n.address, n.port), TcpService.connectTimeoutMillis)
+    } catch {
+      case e: SocketTimeoutException =>
+        Debug.warning(f"Timed out connecting to $n after ${(System.nanoTime - start) / math.pow(10, 9)}%.3f seconds")
+        throw e
+    }
     val worker = new TcpServiceWorker(this, socket)
     worker.sendNode(n)
     worker.start()