You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ru...@apache.org on 2026/09/02 14:24:17 UTC
(spark) branch master updated: [SPARK-59070][PYTHON][ML] Guard TorchDistributor log socket getpeername() against OSError
This is an automated email from the ASF dual-hosted git repository.
zhengruifeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new b519a3fdf48d [SPARK-59070][PYTHON][ML] Guard TorchDistributor log socket getpeername() against OSError
b519a3fdf48d is described below
commit b519a3fdf48d2889817e15f0a2aeb34e59f0a923
Author: Tomu Hirata <To...@users.noreply.github.com>
AuthorDate: Wed Sep 2 22:23:46 2026 +0800
[SPARK-59070][PYTHON][ML] Guard TorchDistributor log socket getpeername() against OSError
### What changes were proposed in this pull request?
In `TorchDistributor._execute_command`, the loop that streams subprocess output compares `log_streaming_client.sock.getsockname()[0]` with `log_streaming_client.sock.getpeername()[0]` to decide whether to echo each log line to stdout (avoiding duplication when client and server run on the same node). This comparison is now wrapped in a `try/except OSError` block. If the socket call raises, `same_node` falls back to `False`, and the line is written to stdout.
### Why are the changes needed?
The log-streaming socket is best effort and can be dropped by a cloud-provider idle-timeout while torch/NCCL initializes (roughly 24 seconds of silence). The next log line then calls `getpeername()` on the dead socket, raising `OSError: [Errno 107] Transport endpoint is not connected`. Because `_execute_command` runs inside a barrier task, this uncaught error aborts the entire multi-node training job approximately 30 seconds in.
`LogStreamingClient.send()` in `log_communication.py` already guards against this case (try/except → mark client failed); the getsockname/getpeername comparison was the only remaining unguarded call site.
Reproduced on Databricks Runtime 17.3.19 ML GPU against Apache Spark 4.0.
### Does this PR introduce _any_ user-facing change?
No API change. Multi-node `TorchDistributor` training that previously crashed with `OSError: [Errno 107] Transport endpoint is not connected` ~30 s into a run will now degrade the best-effort log socket gracefully and continue training.
### How was this patch tested?
Added `TorchDistributorBaselineUnitTests.test_execute_command_survives_log_socket_drop` in `python/pyspark/ml/torch/tests/test_distributor.py`. The test configures a mock `LogStreamingClient` whose `getpeername()` raises `OSError(107, ...)`, then asserts that `_execute_command` still completes and the output line reaches stdout.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude (claude-sonnet-4-6)
Closes #58375 from TomeHirata/SPARK-59070-torchdistributor-log-socket-oserror.
Authored-by: Tomu Hirata <To...@users.noreply.github.com>
Signed-off-by: Ruifeng Zheng <ru...@apache.org>
---
python/pyspark/ml/torch/distributor.py | 23 +++++++++++++++--------
python/pyspark/ml/torch/tests/test_distributor.py | 20 +++++++++++++++++++-
2 files changed, 34 insertions(+), 9 deletions(-)
diff --git a/python/pyspark/ml/torch/distributor.py b/python/pyspark/ml/torch/distributor.py
index 8d5b1a74b9ae..697272909853 100644
--- a/python/pyspark/ml/torch/distributor.py
+++ b/python/pyspark/ml/torch/distributor.py
@@ -485,14 +485,21 @@ class TorchDistributor(Distributor):
decoded = line.decode("utf-8")
tail.append(decoded)
if redirect_to_stdout:
- if (
- log_streaming_client
- and not log_streaming_client.failed
- and (
- log_streaming_client.sock.getsockname()[0]
- == log_streaming_client.sock.getpeername()[0]
- )
- ):
+ same_node = False
+ if log_streaming_client and not log_streaming_client.failed:
+ try:
+ same_node = (
+ log_streaming_client.sock.getsockname()[0]
+ == log_streaming_client.sock.getpeername()[0]
+ )
+ except OSError:
+ # The log-streaming socket is best effort and can be
+ # dropped (e.g. a cloud-provider idle-timeout) while
+ # torch/NCCL initializes. getpeername() then raises
+ # OSError; a barrier task must not die over log
+ # redirection, so fall back to writing to stdout.
+ same_node = False
+ if same_node:
# If log_streaming_client and log_stream_server are in the same
# node (typical case is spark local mode),
# server side will redirect the log to STDOUT,
diff --git a/python/pyspark/ml/torch/tests/test_distributor.py b/python/pyspark/ml/torch/tests/test_distributor.py
index 0149b869fc53..4f3891c64cb6 100644
--- a/python/pyspark/ml/torch/tests/test_distributor.py
+++ b/python/pyspark/ml/torch/tests/test_distributor.py
@@ -27,7 +27,7 @@ import time
import unittest
from io import StringIO
from typing import Any, Callable, Dict
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
from pyspark import SparkConf, SparkContext
from pyspark.ml.torch.distributor import TorchDistributor, _get_gpus_owned
@@ -257,6 +257,24 @@ class TorchDistributorBaselineUnitTestsMixin:
error_command = ["bash", "-c", "'abc''def'"]
TorchDistributor._execute_command(error_command)
+ def test_execute_command_survives_log_socket_drop(self) -> None:
+ """A dropped log-streaming socket must not kill the task.
+
+ The socket is best effort and can be closed (e.g. a cloud-provider
+ idle-timeout) mid-run, making getpeername() raise OSError. The command
+ must still complete and its logs fall back to stdout.
+ """
+ client = MagicMock()
+ client.failed = False
+ client.sock.getsockname.return_value = ("10.0.0.1", 12345)
+ client.sock.getpeername.side_effect = OSError(107, "Transport endpoint is not connected")
+
+ with patch_stdout() as output:
+ TorchDistributor._execute_command(
+ ["echo", "hello_after_socket_drop"], log_streaming_client=client
+ )
+ self.assertIn("hello_after_socket_drop", output.getvalue().strip())
+
def test_create_torchrun_command(self) -> None:
train_path = "train.py"
args_string = ["1", "3"]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org