Address comments and fix docs

Author: BryanCutler

core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala

@@ -138,8 +138,8 @@ private[spark] object PythonRDD extends Logging {
    * or to enable local execution.
    *
    * @return 3-tuple (as a Java array) with the port number of a local socket which serves the
-   *         data collected from this job, the secret for authentication, and a server object
-   *         that can be used to sync the JVM serving thread in Python.
+   *         data collected from this job, the secret for authentication, and a socket auth
+   *         server object that can be used to join the JVM serving thread in Python.
    */
   def runJob(
       sc: SparkContext,
@@ -158,8 +158,8 @@ private[spark] object PythonRDD extends Logging {
    * A helper function to collect an RDD as an iterator, then serve it via socket.
    *
    * @return 3-tuple (as a Java array) with the port number of a local socket which serves the
-   *         data collected from this job, the secret for authentication, and a server object
-   *         that can be used to sync the JVM serving thread in Python.
+   *         data collected from this job, the secret for authentication, and a socket auth
+   *         server object that can be used to join the JVM serving thread in Python.
    */
   def collectAndServe[T](rdd: RDD[T]): Array[Any] = {
     serveIterator(rdd.collect().iterator, s"serve RDD ${rdd.id}")
@@ -170,12 +170,12 @@ private[spark] object PythonRDD extends Logging {
    * are collected as separate jobs, by order of index. Partition data is first requested by a
    * non-zero integer to start a collection job. The response is prefaced by an integer with 1
    * meaning partition data will be served, 0 meaning the local iterator has been consumed,
-   * and -1 meaining an error occurred during collection. This function is used by
+   * and -1 meaning an error occurred during collection. This function is used by
    * pyspark.rdd._local_iterator_from_socket().
    *
    * @return 3-tuple (as a Java array) with the port number of a local socket which serves the
-   *         data collected from this job, the secret for authentication, and a server object
-   *         that can be used to sync the JVM serving thread in Python.
+   *         data collected from this job, the secret for authentication, and a socket auth
+   *         server object that can be used to join the JVM serving thread in Python.
    */
   def toLocalIteratorAndServe[T](rdd: RDD[T]): Array[Any] = {
     val handleFunc = (sock: Socket) => {
@@ -447,8 +447,8 @@ private[spark] object PythonRDD extends Logging {
    * The thread will terminate after all the data are sent or any exceptions happen.
    *
    * @return 3-tuple (as a Java array) with the port number of a local socket which serves the
-   *         data collected from this job, the secret for authentication, and a server object
-   *         that can be used to sync the JVM serving thread in Python.
+   *         data collected from this job, the secret for authentication, and a socket auth
+   *         server object that can be used to join the JVM serving thread in Python.
    */
   def serveIterator(items: Iterator[_], threadName: String): Array[Any] = {
     serveToStream(threadName) { out =>
@@ -470,8 +470,8 @@ private[spark] object PythonRDD extends Logging {
    * exceptions happen.
    *
    * @return 3-tuple (as a Java array) with the port number of a local socket which serves the
-   *         data collected from this job, the secret for authentication, and a server object
-   *         that can be used to sync the JVM serving thread in Python.
+   *         data collected from this job, the secret for authentication, and a socket auth
+   *         server object that can be used to join the JVM serving thread in Python.
    */
   private[spark] def serveToStream(
       threadName: String)(writeFunc: OutputStream => Unit): Array[Any] = {

core/src/main/scala/org/apache/spark/security/SocketAuthServer.scala

@@ -32,6 +32,9 @@ import org.apache.spark.util.{ThreadUtils, Utils}
 /**
  * Creates a server in the JVM to communicate with external processes (e.g., Python and R) for
  * handling one batch of data, with authentication and error handling.
+ *
+ * The socket server can only accept one connection, or close if no connection
+ * in 15 seconds.
  */
 private[spark] abstract class SocketAuthServer[T](
     authHelper: SocketAuthHelper,
@@ -88,9 +91,9 @@ private[spark] abstract class SocketAuthServer[T](
 }
 
 /**
- * Create a socket server class and run user function on the socket in a background thread.
- * This is the same as calling SocketAuthServer.setupOneConnectionServer except it creates
- * a server object that can then be synced from Python.
+ * Create a socket server class and run user function on the socket in a background thread
+ * that can read and write to the socket input/output streams. The function is passed in a
+ * socket that has been connected and authenticated.
  */
 private[spark] class SocketFuncServer(
     authHelper: SocketAuthHelper,
@@ -108,10 +111,15 @@ private[spark] object SocketAuthServer {
    * Convenience function to create a socket server and run a user function in a background
    * thread to write to an output stream.
    *
+   * The socket server can only accept one connection, or close if no connection
+   * in 15 seconds.
+   *
    * @param threadName Name for the background serving thread.
    * @param authHelper SocketAuthHelper for authentication
    * @param writeFunc User function to write to a given OutputStream
-   * @return
+   * @return 3-tuple (as a Java array) with the port number of a local socket which serves the
+   *         data collected from this job, the secret for authentication, and a socket auth
+   *         server object that can be used to join the JVM serving thread in Python.
    */
   def serveToStream(
       threadName: String,

python/pyspark/rdd.py

@@ -174,7 +174,7 @@ class PyLocalIterable(object):
         """ Create a synchronous local iterable over a socket """
 
         def __init__(self, _sock_info, _serializer):
-            port, auth_secret, self.jserver_obj = _sock_info
+            port, auth_secret, self.jsocket_auth_server = _sock_info
             self._sockfile = _create_local_socket((port, auth_secret))
             self._serializer = _serializer
             self._read_iter = iter([])  # Initialize as empty iterator
@@ -197,7 +197,7 @@ def __iter__(self):
 
                 # An error occurred, join serving thread and raise any exceptions from the JVM
                 elif self._read_status == -1:
-                    self.jserver_obj.getResult()
+                    self.jsocket_auth_server.getResult()
 
         def __del__(self):
             # If local iterator is not fully consumed,

python/pyspark/sql/dataframe.py

@@ -2200,14 +2200,14 @@ def _collectAsArrow(self):
         .. note:: Experimental.
         """
         with SCCallSiteSync(self._sc) as css:
-            port, auth_secret, jserver_obj = self._jdf.collectAsArrowToPython()
+            port, auth_secret, jsocket_auth_server = self._jdf.collectAsArrowToPython()
 
         # Collect list of un-ordered batches where last element is a list of correct order indices
         try:
             results = list(_load_from_socket((port, auth_secret), ArrowCollectSerializer()))
         finally:
             # Join serving thread and raise any exceptions from collectAsArrowToPython
-            jserver_obj.getResult()
+            jsocket_auth_server.getResult()
 
         # Separate RecordBatches from batch order indices in results
         batches = results[:-1]