        serverSocket = new ServerSocket(0, 1, InetAddress.getByAddress(Array(127, 0, 0, 1)))
 
        // Create and start the worker
-@@ -163,13 +167,78 @@ private[spark] class PythonWorkerFactory(pythonExec: String, envVars: Map[String
+@@ -163,29 +167,97 @@ private[spark] class PythonWorkerFactory(pythonExec: String, envVars: Map[String
        workerEnv.put("PYTHON_WORKER_FACTORY_PORT", serverSocket.getLocalPort.toString)
        workerEnv.put("PYTHON_WORKER_FACTORY_SECRET", authHelper.secret)
        val worker = pb.start()
 +       */
 +
 +      simpleWorkerBuffer.synchronized {
-+        println(s"Thread ${Thread.currentThread().getId}: creating simple worker synchronized")
++        logInfo(s"Thread ${Thread.currentThread().getId}: creating simple worker synchronized")
 +        val (worker, ss) = {
 +          if (simpleWorkerBuffer.size >= maxSimpleWorker) {
 +            val p = keysIterator().next()
 +            val buffer = simpleWorkerBuffer.get(p).get
-+            println(s"scala1: waiting python's finished result.")
++            logInfo(s"scala1: waiting python's finished result.")
 +            val input = new DataInputStream(new BufferedInputStream(buffer._1.getInputStream, 1024))
 +            val r = input.readInt()
-+            println(s"scala1: python result is ${r}.")
++            logInfo(s"scala1: python result is ${r}.")
 +            require(r == SpecialLengths.FINISHED)
 +            val ss = new ServerSocket(0, 1, InetAddress.getByAddress(Array(127, 0, 0, 1)))
 +            ss.setSoTimeout(10000)
-+            println(s"scala1: new data local port ${ss.getLocalPort}--------")
++            logInfo(s"scala1: new data local port ${ss.getLocalPort}--------")
 +            val output = new DataOutputStream(
 +              new BufferedOutputStream(buffer._1.getOutputStream, 1024))
 +            output.writeInt(ss.getLocalPort)

+            }
+            val ss = new ServerSocket(0, 1, InetAddress.getByAddress(Array(127, 0, 0, 1)))
+            ss.setSoTimeout(10000)
+            println(s"scala2: new data local port ${ss.getLocalPort}--------")
+            val output = new DataOutputStream(
+              new BufferedOutputStream(manageSocket.getOutputStream, 1024))
+            output.writeInt(ss.getLocalPort)

+            val workerEnv = pb.environment()
+            workerEnv.putAll(envVars.asJava)
+            workerEnv.put("PYTHONPATH", pythonPath)
+            // This is equivalent to setting the -u flag;
+            // we use it because ipython doesn't support -u:
+            workerEnv.put("PYTHONUNBUFFERED", "YES")
+            workerEnv.put("PYTHON_WORKER_FACTORY_PORT", manageServerSocket.getLocalPort.toString)
+            workerEnv.put("PYTHON_WORKER_FACTORY_SECRET", authHelper.secret)
