Repository: pytest-dev/pytest-xdist

Issue:
Running tests on multiple remote cores using socket
I have a remote server, and I would like to run tests on its cores in parallel. This is simple enough using SSH, where I can pass `n*ssh=...`. For socket I would expect it would work with `n*socket=...`. However, the `socketserver.py` script linked in the execnet repo does not support parallel clients. So whenever I try to parallelize the socket argument using `--tx 2*socket=...` my master gets stuck after initializing one worker, since the socket blocks on the first connection.

Is there a better fix to this other than modifying `socketserver.py` to fork for every connection?

Thanks.

Hints from issue comments:
@RonnyPfannschmidt: Socket server currently doesn't support multiple processes 

The easiest way would be to add hop gateways to xdist and use popen//via=hop

In this case a hop gateway wouldn't run tests itself it would just proxy to multiple popen gates started via it @Alon-L: > Socket server currently doesn't support multiple processes
> 
> The easiest way would be to add hop gateways to xdist and use popen//via=hop
> 
> In this case a hop gateway wouldn't run tests itself it would just proxy to multiple popen gates started via it

Thank you, this worked:
```
pytest -d --tx "socket=IP:PORT//id=my_proxy" --tx "5*popen//via=my_proxy"
```

However, it also treats the first `tx` argument as a remote worker, so I end up with 6 workers.
I looked at the xdist code for `NodeManager`. It seems that there's no way to call execnet's `makegateway` without registering the new gateway as a remote worker.

I'd like to create a PR to fix that, and I'm not sure which way to go for this. I'm currently thinking of adding an argument `--xspec` which is passed to the execnet's `Group` on initialization.
For example, my previous command becomes:
```
pytest -d --xspec "socket=IP:PORT//id=my_proxy" --tx "5*popen//via=my_proxy"
```
What do you think?@RonnyPfannschmidt: It needs a less generic name @RonnyPfannschmidt: Either a role=proxy on the tx or a arg name that explicitly names proxies @Alon-L: I thought this could be a more generic feature to control the `xspec` argument passed to `Group` on initialization, which can consequentially be used for proxies.
If you still think that's too generic, I think the `role=proxy` is a good idea. The only problem I see with it is that it's different than execnet's xspec specification. However, if you think that's fine then I'll add it.@RonnyPfannschmidt: Your explanation makes sense as well

For me the key is having non test gateway specs with a specific name 

Using a role key would be supported by execnet already 

However adding it with single use is potentially problematic 

Solution patch:
diff --git a/changelog/1170.feature b/changelog/1170.feature
new file mode 100644
index 00000000..d67e5947
--- /dev/null
+++ b/changelog/1170.feature
@@ -0,0 +1,4 @@
+Add the `--px` arg to create proxy gateways.
+
+Proxy gateways are passed to additional gateways using the `via` keyword.
+They can serve as a way to run multiple workers on remote machines.
diff --git a/docs/remote.rst b/docs/remote.rst
index f42fb812..6fe806f9 100644
--- a/docs/remote.rst
+++ b/docs/remote.rst
@@ -66,6 +66,20 @@ new socket host with something like this::
     pytest -d --tx socket=192.168.1.102:8888 --rsyncdir mypkg
 
 
+Using proxies to run multiple workers on remote machines
+---------------------------------------
+
+In case you want to run multiple workers on a remote machine,
+you can create a proxy gateway for the machine, and run multiple
+workers using the `via` attribute.::
+
+    pytest -d --px id=my_proxy//socket=192.168.1.102:8888 --tx 5*popen//via=my_proxy
+
+Here we declare a proxy gateway using the `--px` arg, and
+create 5 workers that run on the remote server using the proxy.
+Note that the proxy gateway does not run a worker, thus only 5
+workers are created.
+
 
 Running tests on many platforms at once
 ---------------------------------------
diff --git a/src/xdist/plugin.py b/src/xdist/plugin.py
index f670d9de..0cf90f86 100644
--- a/src/xdist/plugin.py
+++ b/src/xdist/plugin.py
@@ -139,6 +139,17 @@ def pytest_addoption(parser: pytest.Parser) -> None:
             "--tx ssh=user@codespeak.net//chdir=testcache"
         ),
     )
+    group.addoption(
+        "--px",
+        dest="px",
+        action="append",
+        default=[],
+        metavar="xspec",
+        help=(
+            "Add a proxy gateway to pass to test execution environments using `via`. Example:\n"
+            "--px id=my_proxy//socket=192.168.1.102:8888 --tx 5*popen//via=my_proxy"
+        ),
+    )
     group._addoption(
         "-d",
         action="store_true",
diff --git a/src/xdist/scheduler/each.py b/src/xdist/scheduler/each.py
index 8552b53f..bf914519 100644
--- a/src/xdist/scheduler/each.py
+++ b/src/xdist/scheduler/each.py
@@ -6,7 +6,7 @@
 
 from xdist.remote import Producer
 from xdist.report import report_collection_diff
-from xdist.workermanage import parse_spec_config
+from xdist.workermanage import parse_tx_spec_config
 from xdist.workermanage import WorkerController
 
 
@@ -26,7 +26,7 @@ class EachScheduling:
 
     def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
         self.config = config
-        self.numnodes = len(parse_spec_config(config))
+        self.numnodes = len(parse_tx_spec_config(config))
         self.node2collection: dict[WorkerController, list[str]] = {}
         self.node2pending: dict[WorkerController, list[int]] = {}
         self._started: list[WorkerController] = []
diff --git a/src/xdist/scheduler/load.py b/src/xdist/scheduler/load.py
index ac011e57..56f6092f 100644
--- a/src/xdist/scheduler/load.py
+++ b/src/xdist/scheduler/load.py
@@ -7,7 +7,7 @@
 
 from xdist.remote import Producer
 from xdist.report import report_collection_diff
-from xdist.workermanage import parse_spec_config
+from xdist.workermanage import parse_tx_spec_config
 from xdist.workermanage import WorkerController
 
 
@@ -58,7 +58,7 @@ class LoadScheduling:
     """
 
     def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
-        self.numnodes = len(parse_spec_config(config))
+        self.numnodes = len(parse_tx_spec_config(config))
         self.node2collection: dict[WorkerController, list[str]] = {}
         self.node2pending: dict[WorkerController, list[int]] = {}
         self.pending: list[int] = []
diff --git a/src/xdist/scheduler/loadscope.py b/src/xdist/scheduler/loadscope.py
index ee3f2fdf..114561b4 100644
--- a/src/xdist/scheduler/loadscope.py
+++ b/src/xdist/scheduler/loadscope.py
@@ -8,7 +8,7 @@
 
 from xdist.remote import Producer
 from xdist.report import report_collection_diff
-from xdist.workermanage import parse_spec_config
+from xdist.workermanage import parse_tx_spec_config
 from xdist.workermanage import WorkerController
 
 
@@ -91,7 +91,7 @@ class LoadScopeScheduling:
     """
 
     def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
-        self.numnodes = len(parse_spec_config(config))
+        self.numnodes = len(parse_tx_spec_config(config))
         self.collection: list[str] | None = None
 
         self.workqueue: OrderedDict[str, dict[str, bool]] = OrderedDict()
diff --git a/src/xdist/scheduler/worksteal.py b/src/xdist/scheduler/worksteal.py
index 28708fcc..550372ee 100644
--- a/src/xdist/scheduler/worksteal.py
+++ b/src/xdist/scheduler/worksteal.py
@@ -7,7 +7,7 @@
 
 from xdist.remote import Producer
 from xdist.report import report_collection_diff
-from xdist.workermanage import parse_spec_config
+from xdist.workermanage import parse_tx_spec_config
 from xdist.workermanage import WorkerController
 
 
@@ -65,7 +65,7 @@ class WorkStealingScheduling:
     """
 
     def __init__(self, config: pytest.Config, log: Producer | None = None) -> None:
-        self.numnodes = len(parse_spec_config(config))
+        self.numnodes = len(parse_tx_spec_config(config))
         self.node2collection: dict[WorkerController, list[str]] = {}
         self.node2pending: dict[WorkerController, list[int]] = {}
         self.pending: list[int] = []
diff --git a/src/xdist/workermanage.py b/src/xdist/workermanage.py
index 08ba243f..201c8e71 100644
--- a/src/xdist/workermanage.py
+++ b/src/xdist/workermanage.py
@@ -23,7 +23,7 @@
 from xdist.remote import WorkerInfo
 
 
-def parse_spec_config(config: pytest.Config) -> list[str]:
+def parse_tx_spec_config(config: pytest.Config) -> list[str]:
     xspeclist = []
     tx: list[str] = config.getvalue("tx")
     for xspec in tx:
@@ -57,8 +57,17 @@ def __init__(
         if self.testrunuid is None:
             self.testrunuid = uuid.uuid4().hex
         self.group = execnet.Group(execmodel="main_thread_only")
+        for proxy_spec in self._getpxspecs():
+            # Proxy gateways do not run workers, and are meant to be passed with the `via` attribute
+            # to additional gateways.
+            # They are useful for running multiple workers on remote machines.
+            if getattr(proxy_spec, "id", None) is None:
+                raise pytest.UsageError(
+                    f"Proxy gateway {proxy_spec} must include an id"
+                )
+            self.group.makegateway(proxy_spec)
         if specs is None:
-            specs = self._getxspecs()
+            specs = self._gettxspecs()
         self.specs: list[execnet.XSpec] = []
         for spec in specs:
             if not isinstance(spec, execnet.XSpec):
@@ -107,8 +116,11 @@ def setup_node(
     def teardown_nodes(self) -> None:
         self.group.terminate(self.EXIT_TIMEOUT)
 
-    def _getxspecs(self) -> list[execnet.XSpec]:
-        return [execnet.XSpec(x) for x in parse_spec_config(self.config)]
+    def _gettxspecs(self) -> list[execnet.XSpec]:
+        return [execnet.XSpec(x) for x in parse_tx_spec_config(self.config)]
+
+    def _getpxspecs(self) -> list[execnet.XSpec]:
+        return [execnet.XSpec(x) for x in self.config.getoption("px")]
 
     def _getrsyncdirs(self) -> list[Path]:
         for spec in self.specs:


Test patch:
diff --git a/testing/test_plugin.py b/testing/test_plugin.py
index f3526670..81d9e873 100644
--- a/testing/test_plugin.py
+++ b/testing/test_plugin.py
@@ -295,7 +295,7 @@ class TestDistOptions:
     def test_getxspecs(self, pytester: pytest.Pytester) -> None:
         config = pytester.parseconfigure("--tx=popen", "--tx", "ssh=xyz")
         nodemanager = NodeManager(config)
-        xspecs = nodemanager._getxspecs()
+        xspecs = nodemanager._gettxspecs()
         assert len(xspecs) == 2
         print(xspecs)
         assert xspecs[0].popen
@@ -303,7 +303,7 @@ def test_getxspecs(self, pytester: pytest.Pytester) -> None:
 
     def test_xspecs_multiplied(self, pytester: pytest.Pytester) -> None:
         config = pytester.parseconfigure("--tx=3*popen")
-        xspecs = NodeManager(config)._getxspecs()
+        xspecs = NodeManager(config)._gettxspecs()
         assert len(xspecs) == 3
         assert xspecs[1].popen
 
diff --git a/testing/test_workermanage.py b/testing/test_workermanage.py
index 491d0424..6fb2795c 100644
--- a/testing/test_workermanage.py
+++ b/testing/test_workermanage.py
@@ -368,6 +368,34 @@ def test_one():
         (rep,) = reprec.getreports("pytest_runtest_logreport")
         assert rep.passed
 
+    def test_proxy_gateway_setup_nodes(self, pytester: pytest.Pytester) -> None:
+        nodemanager = NodeManager(
+            pytester.parseconfig(
+                "--px", "popen//id=my_proxy", "--tx", "popen//via=my_proxy"
+            )
+        )
+        nodes = nodemanager.setup_nodes(None)  # type: ignore[arg-type]
+
+        # Proxy gateways are considered as an execnet gateway
+        assert len(nodemanager.group) == 2
+
+        # Proxy gateways do not run workers
+        assert len(nodes) == 1
+
+    def test_proxy_gateway(self, pytester: pytest.Pytester) -> None:
+        pytester.makepyfile(
+            __init__="",
+            test_x="""
+            def test_one():
+                pass
+        """,
+        )
+        reprec = pytester.inline_run(
+            "-d", "--px", "popen//id=my_proxy", "--tx", "popen//via=my_proxy"
+        )
+        rep = reprec.getreports("pytest_runtest_logreport")
+        assert rep[1].passed
+
 
 class MyWarning(UserWarning):
     pass