[SPARK-2313] PySpark pass port rather than stdin

core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala

@@ -149,7 +149,16 @@ object SparkSubmit {
     if (args.isPython) {
       if (args.primaryResource == PYSPARK_SHELL) {
         args.mainClass = "py4j.GatewayServer"
-        args.childArgs = ArrayBuffer("--die-on-broken-pipe", "0")
+
+        var gateway_port = 0;
+        // extract gateway_port from childArgs
+        for (i <- 1 until args.childArgs.length){
+          if (args.childArgs(i-1) == "--gateway_port"){
+            gateway_port = args.childArgs(i).toInt
+          }
+        }
+
+        args.childArgs = ArrayBuffer("--die-on-broken-pipe", gateway_port.toString)
       } else {
         // If a python file is provided, add it to the child arguments and list of files to deploy.
         // Usage: PythonAppRunner <main python file> <extra python files> [app arguments]

python/pyspark/java_gateway.py

@@ -20,11 +20,31 @@
 import sys
 import signal
 import shlex
+import socket
 import platform
 from subprocess import Popen, PIPE
 from threading import Thread
 from py4j.java_gateway import java_import, JavaGateway, GatewayClient
 
+def peek_free_port():
+    """
+    Check and return an available port number for binding.
+    The port number is generated with a hint from PID to avoid possible concurrent issues.
+
+    :return: available port or 0 if no ports available
+    """
+
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    port_start = 10000
+    port = os.getpid() % (0xFFFF - port_start) + port_start      # ensure port range in 10000(port_start) ~ 65535
+    for offset in range(128):   # 128 attempts max
+        try:
+            s.bind(("0.0.0.0", port + offset))
+        except socket.error:
+            continue
+        s.close()
+        return port + offset
+    return 0      # maximum number of attempts reached, let's leave it to Py4j
 
 def launch_gateway():
     SPARK_HOME = os.environ["SPARK_HOME"]
@@ -41,6 +61,11 @@ def launch_gateway():
         submit_args = submit_args if submit_args is not None else ""
         submit_args = shlex.split(submit_args)
         command = [os.path.join(SPARK_HOME, script)] + submit_args + ["pyspark-shell"]
+
+        gateway_port_candidate = peek_free_port()
+        if gateway_port_candidate>0:
+            command += ["--gateway_port", "%d" % gateway_port_candidate]
+
         if not on_windows:
             # Don't send ctrl-c / SIGINT to the Java gateway:
             def preexec_func():
@@ -51,11 +76,13 @@ def preexec_func():
         else:
             # preexec_fn not supported on Windows
             proc = Popen(command, stdout=PIPE, stdin=PIPE)
-
         try:
-            # Determine which ephemeral port the server started on:
+            # Determine which ephemeral port the server started on,
+            # or double check if gateway_port_candidate is passed in:
             gateway_port = proc.stdout.readline()
             gateway_port = int(gateway_port)
+            if gateway_port_candidate > 0 and gateway_port != gateway_port_candidate:
+                print "Warning, gateway_port_candidate != gateway_port, possible concurrent issues: %r != %r" % (gateway_port_candidate, gateway_port)
         except ValueError:
             # Grab the remaining lines of stdout
             (stdout, _) = proc.communicate()
@@ -70,7 +97,12 @@ def preexec_func():
                 error_msg += "--------------------------------------------------------------\n"
                 error_msg += gateway_port + stdout
                 error_msg += "--------------------------------------------------------------\n"
-            raise Exception(error_msg)
+            if gateway_port_candidate > 0:
+                print "Warning, parse gateway_port failed, Caused by:\n" + error_msg
+                print "However, we have a chance to assume gateway_port_candidate is used. Suppress the exception and give it a try..."
+                gateway_port = gateway_port_candidate
+            else:
+                raise Exception(error_msg)
 
         # In Windows, ensure the Java child processes do not linger after Python has exited.
         # In UNIX-based systems, the child process can kill itself on broken pipe (i.e. when