Skip to content

Commit

Permalink
remove setReuseAddress
Browse files Browse the repository at this point in the history
  • Loading branch information
Davies Liu committed Apr 2, 2015
1 parent 7977c2f commit e5a51a2
Show file tree
Hide file tree
Showing 2 changed files with 2 additions and 15 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -605,7 +605,6 @@ private[spark] object PythonRDD extends Logging {
*/
private def serveIterator[T](items: Iterator[T], threadName: String): Int = {
val serverSocket = new ServerSocket(0, 1)
serverSocket.setReuseAddress(true)
// Close the socket if no connection in 3 seconds
serverSocket.setSoTimeout(3000)

Expand Down
16 changes: 2 additions & 14 deletions python/pyspark/rdd.py
Original file line number Diff line number Diff line change
Expand Up @@ -113,23 +113,11 @@ def _parse_memory(s):

def _load_from_socket(port, serializer):
sock = socket.socket()
sock.settimeout(1)
sock.settimeout(3)
try:
sock.connect(("localhost", port))
rf = sock.makefile("rb", 65536)
iter = serializer.load_stream(rf)
try:
yield next(iter)
except socket.timeout as e:
# the connection is not acknowledged by JVM, retry
# server will be closed after 3 seconds, then it will be refused
for v in _load_from_socket(port, serializer):
yield v
return

# increase the timeout, because the server side may be slowed down by GC
sock.settimeout(10)
for item in iter:
for item in serializer.load_stream(rf):
yield item
finally:
sock.close()
Expand Down

0 comments on commit e5a51a2

Please sign in to comment.