### SSH Gateway Specification Example

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Demonstrates how to specify an SSH gateway connection to a remote host. This example includes setting the Python version and changing the remote working directory.

```default
ssh=wyvern//python=python3.3//chdir=mycache
```

--------------------------------

### Socket-Based Gateway Connections with execnet

Source: https://context7.com/pytest-dev/execnet/llms.txt

Illustrates the setup for socket-based gateway connections in execnet. This allows for persistent remote connections that can be accessed via a specified port. The example points to the command-line script used to start such a server.

```python
import execnet
import time

# Start a socket server gateway (in separate process or host)
# This would typically run as: python -m execnet.script.socketserver :8888
```

--------------------------------

### SSH Gateway with Custom Port Example

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Shows how to connect to a remote host via SSH using a non-standard port. This passes arguments directly to the underlying ssh client.

```default
ssh=-p 5000 myhost
```

--------------------------------

### Simple Proxying with execnet.Group and makegateway

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_proxy.md

Demonstrates creating a gateway that proxies its I/O through a remote master gateway. This snippet shows the setup of a master gateway and a worker gateway using execnet.Group and the makegateway function, with the worker implicitly using the master's connection via the defaultspec.

```python
import execnet
group = execnet.Group()
group.defaultspec = 'popen//via=master'
master = group.makegateway('popen//id=master')
print(master)
worker = group.makegateway()
print(worker)
print(group)
```

--------------------------------

### File Synchronization using execnet.RSync

Source: https://context7.com/pytest-dev/execnet/llms.txt

Synchronizes directory structures to remote gateways using execnet.RSync. This example creates a temporary directory with files, sets up an RSync transfer to a remote destination, and verifies the files on the remote side.

```python
import execnet
import tempfile
import os

# Create source directory with files
source_dir = tempfile.mkdtemp(prefix='source_')
os.makedirs(os.path.join(source_dir, 'subdir'))
with open(os.path.join(source_dir, 'file1.txt'), 'w') as f:
    f.write('Content of file 1')
with open(os.path.join(source_dir, 'subdir', 'file2.txt'), 'w') as f:
    f.write('Content of file 2')

# Create remote gateway
gw = execnet.makegateway("popen")

# Create destination on remote
dest_dir = tempfile.mkdtemp(prefix='dest_')

# Set up RSync
rsync = execnet.RSync(source_dir, verbose=True)

# Track completion
finished = []
def on_finish():
    finished.append(True)
    print("Sync completed!")

rsync.add_target(gw, dest_dir, finishedcallback=on_finish)

# Execute synchronization
rsync.send()

# Verify on remote
channel = gw.remote_exec(f"""
    import os
    files = []
    for root, dirs, filenames in os.walk('{dest_dir}'):
        for name in filenames:
            files.append(os.path.relpath(os.path.join(root, name), '{dest_dir}'))
    channel.send(sorted(files))
""")

remote_files = channel.receive()
print(f"Files synced to remote: {remote_files}")

gw.exit()
```

--------------------------------

### Instantiate a Python Subprocess Gateway

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Creates and configures a gateway to a local Python interpreter using the 'popen' specification. This is a basic way to start a remote Python process for execution.

```python
import execnet

gateway = execnet.makegateway("popen")
```

--------------------------------

### Interacting with Java Objects from CPython using Jython

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/hybridpython.md

Shows how to use a CPython interpreter to connect to a Jython interpreter and interact with Java objects. This example demonstrates creating a `java.util.Vector` in Jython, adding elements, and then receiving those elements back in the CPython interpreter. It highlights execnet's capability to bridge CPython with other Python implementations like Jython.

```python
import execnet
gw = execnet.makegateway("popen//python=jython")
channel = gw.remote_exec("""
    from java.util import Vector
    v = Vector()
    v.add('aaa')
    v.add('bbb')
    for val in v:
        channel.send(val)
""")

for item in channel:
    print (item)
```

--------------------------------

### Remote Execution: Connect Python3 to Python2.7 with Numpy

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/hybridpython.md

Illustrates connecting a Python 3 interpreter to a Python 2.7 interpreter with the numpy library installed. This example shows how to send data to the remote interpreter, perform operations (appending to a numpy array), and receive the result (the string representation of the array). It utilizes `execnet.makegateway` to establish the connection and `gw.remote_exec` for executing code on the remote end.

```python
import execnet
gw = execnet.makegateway("popen//python=python2.7")
channel = gw.remote_exec("""
    import numpy
    array = numpy.array([1,2,3])
    while 1:
        x = channel.receive()
        if x is None:
            break
        array = numpy.append(array, x)
    channel.send(repr(array))
""")
for x in range(10):
    channel.send(x)
channel.send(None)
print (channel.receive())
```

--------------------------------

### Configure Remote SSH Gateways

Source: https://context7.com/pytest-dev/execnet/llms.txt

Establishes a connection to a remote host via SSH, allowing for specification of the interpreter path and working directory. This example shows how to retrieve system metadata from a remote environment.

```python
import execnet

# Connect to remote host with Python 3.3 in specific directory
gw = execnet.makegateway("ssh=myhost//python=python3.3//chdir=/tmp/workspace")

# Get system information from remote
channel = gw.remote_exec("""
    import sys, os
    channel.send({
        'platform': sys.platform,
        'version': tuple(sys.version_info),
        'pid': os.getpid(),
        'cwd': os.getcwd()
    })
""")

remote_info = channel.receive()
print(f"Remote platform: {remote_info['platform']}")
print(f"Remote Python: {remote_info['version']}")
print(f"Remote CWD: {remote_info['cwd']}")

gw.exit()
```

--------------------------------

### Distribute tasks across multiple gateways

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_multi.md

A comprehensive example of task distribution using a MultiChannel pattern, managing process lifecycle, and ensuring clean shutdown of worker subprocesses.

```python
from __future__ import annotations
import execnet

group = execnet.Group()
for i in range(4):
    group.makegateway()

def process_item(channel):
    import time, random
    channel.send("ready")
    for x in channel:
        if x is None: break
        time.sleep(random.randrange(3))
        channel.send(x * 10)

mch = group.remote_exec(process_item)
q = mch.make_receive_queue(endmarker=-1)
tasks = list(range(10))
terminated = 0
while 1:
    channel, item = q.get()
    if item == -1:
        terminated += 1
        if terminated == len(mch): break
        continue
    if item != "ready":
        print(f"returned {item}")
    if not tasks and tasks is not None:
        mch.send_each(None)
        tasks = None
    if tasks:
        channel.send(tasks.pop())
group.terminate()
```

--------------------------------

### Configuring Threading Models for execnet

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

execnet supports various threading models ('main_thread_only', 'thread', 'eventlet', 'gevent') for both local and remote sides. The model must be set before creating gateways using `execnet.set_execmodel()`. Requires 'eventlet' to be installed for its model.

```python
import execnet

# Example: locally use 'eventlet', remotely use 'thread' model
execnet.set_execmodel("eventlet", "thread")
gw = execnet.makegateway()
print(gw)
print(gw.remote_status())
print(gw.remote_exec("channel.send(1)").receive())
```

--------------------------------

### Asynchronous Communication with Callbacks

Source: https://context7.com/pytest-dev/execnet/llms.txt

Utilizes channel callbacks to process incoming data asynchronously, preventing the main thread from blocking on receive operations. The example captures results as they arrive until the channel terminates.

```python
import execnet

gw = execnet.makegateway()

# Execute code that sends multiple values
channel = gw.remote_exec("for i in range(10): channel.send(i ** 2)")

# Collect results using callback
results = []
channel.setcallback(results.append, endmarker=None)

# Wait for remote execution to complete
channel.waitclose()

print(f"Received squares: {results[:-1]}")  # Exclude None endmarker
# Output: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

gw.exit()
```

--------------------------------

### Working with C# Objects from CPython using IronPython

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/hybridpython.md

Provides an experimental example of using a CPython interpreter to connect to an IronPython interpreter and interact with C# classes. This snippet demonstrates instantiating a CLR Array instance (`System.Array[float]`) in IronPython and sending its string representation back to CPython. It showcases execnet's potential for interoperability with .NET environments.

```python
import execnet
gw = execnet.makegateway("popen//python=ipy")

channel = gw.remote_exec("""
    import clr
    clr.AddReference("System")
    from System import Array
    array = Array[float]([1,2])
    channel.send(str(array))
""")
print (channel.receive())
```

--------------------------------

### Reconfiguring String Coercion between Python 2 and 3

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/hybridpython.md

Explains how to reconfigure the string coercion behavior between Python 2 and Python 3 interpreters using `gw.reconfigure` and `channel.reconfigure`. This allows for customizing how strings and unicode types are handled during cross-interpreter communication, providing flexibility beyond the default settings. The example demonstrates changing coercion for both `py3str_as_py2str` and `py2str_as_py3str`.

```python
>>> import execnet
>>> gw=execnet.makegateway("popen//python=python3.2")
>>> gw.remote_exec("channel.send('hello')").receive()
u'hello'
>>> gw.reconfigure(py3str_as_py2str=True)
>>> gw.remote_exec("channel.send('hello')").receive()
'hello'
>>> ch = gw.remote_exec('channel.send(type(channel.receive()).__name__)')
>>> ch.send('a')
>>> ch.receive()
'str'
>>> ch = gw.remote_exec('channel.send(type(channel.receive()).__name__)')
>>> ch.reconfigure(py2str_as_py3str=False)
>>> ch.send('a')
>>> ch.receive()
u'bytes'
```

--------------------------------

### Remote Module Execution with execnet

Source: https://context7.com/pytest-dev/execnet/llms.txt

Executes a Python module's source code on a remote gateway. The example shows a 'remote_task.py' module that processes data received through a channel. This is useful for organizing and reusing code in distributed environments.

```python
# File: remote_task.py
"""Remote module that performs specific tasks."""

def process_data(items):
    return [x * 2 for x in items]

if __name__ == '__channelexec__':
    # This block executes when run via remote_exec
    channel.send("Remote module initialized")

    while True:
        try:
            data = channel.receive()
            if data == 'STOP':
                break
            result = process_data(data)
            channel.send(result)
        except EOFError:
            break

    channel.send("Remote module shutting down")
```

```python
# Main code:
import execnet
import remote_task

gw = execnet.makegateway()
channel = gw.remote_exec(remote_task)

# Wait for initialization
print(channel.receive())

# Send tasks
channel.send([1, 2, 3, 4])
print(f"Result: {channel.receive()}")

channel.send([10, 20, 30])
print(f"Result: {channel.receive()}")

# Signal completion
channel.send('STOP')
print(channel.receive())

gw.exit()
```

--------------------------------

### Compare local and remote working directories

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_info.md

Shows how to retrieve the current working directory from a remote process to compare it against the initiating process's directory.

```python
import execnet, os
gw = execnet.makegateway()
ch = gw.remote_exec("import os; channel.send(os.getcwd())")
res = ch.receive()
assert res == os.getcwd()
```

--------------------------------

### Configure Default Specifications for Gateway Factories

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_group.md

Demonstrates setting a default specification on a group to allow the creation of configured gateways without exposing implementation details to the caller.

```python
import execnet
group = execnet.Group()
group.defaultspec = "ssh=localhost//chdir=mytmp//nice=20"
gw = group.makegateway()
ch = gw.remote_exec("import os.path; channel.send(os.path.basename(os.getcwd()))")
print(ch.receive())
```

--------------------------------

### Bootstrap execnet Gateway via Socket

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_info.md

This section explains how to instantiate an execnet gateway using a socket connection when direct SSH access is not feasible. It involves running a `socketserver.py` script on the remote machine and then connecting to it from the local machine using `execnet.makegateway()`.

```bash
python socketserver.py :8888   # bind to all IPs, port 8888
```

```python
import execnet
gw = execnet.makegateway("socket=TARGET-IP:8888")
```

--------------------------------

### Transfer channels over existing channels

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_info.md

Demonstrates the ability to create new channels and send them over an established gateway channel to facilitate sub-communication paths.

```python
import execnet
gw = execnet.makegateway()
channel = gw.remote_exec("""
       ch1, ch2 = channel.receive()
       ch2.send("world")
       ch1.send("hello")
""")
c1 = gw.newchannel()
c2 = gw.newchannel()
channel.send((c1, c2))
print(c1.receive())
print(c2.receive())
```

--------------------------------

### Create and Manage Subprocess Gateways with execnet

Source: https://context7.com/pytest-dev/execnet/llms.txt

Demonstrates how to initiate a local subprocess gateway and perform bidirectional data exchange using channels. The process involves creating a gateway, executing remote code, and cleaning up the connection.

```python
import execnet

# Create a simple subprocess gateway
gw = execnet.makegateway()

# Execute code remotely and communicate
channel = gw.remote_exec("channel.send(channel.receive() + 1)")
channel.send(1)
result = channel.receive()
print(result)  # Output: 2

# Clean up
gw.exit()
```

--------------------------------

### RSync filesystem synchronization with execnet

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Demonstrates how to use the execnet.RSync class to synchronize a local directory with a remote destination. It involves creating an RSync object, adding a remote target, and sending the files. This is useful for deploying code or data to remote servers.

```python
import execnet

rsync = execnet.RSync('/tmp/source')
gw = execnet.makegateway()
rsync.add_target(gw, '/tmp/dest')
rsync.send()
```

--------------------------------

### Vagrant SSH Gateway Specification

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Illustrates how to establish an SSH connection to a Vagrant VM. This uses Vagrant's `vagrant ssh` command and supports standard SSH parameters.

```default
vagrant_ssh=default
```

--------------------------------

### Multi-Channel Pattern for Parallel Processing in Python

Source: https://context7.com/pytest-dev/execnet/llms.txt

Demonstrates parallel processing using multiple gateway channels with execnet. It creates several worker gateways, distributes tasks, collects results, and ensures proper cleanup.

```python
import execnet

# Create multiple gateways
group = execnet.Group()
for i in range(4):
    group.makegateway(f"popen//id=worker{i}")

# Define work function
work_code = """
    import time
    task_id = channel.receive()
    # Simulate work
    time.sleep(0.1)
    result = task_id ** 2
    channel.send({'task': task_id, 'result': result})
"""

# Execute work on all gateways
multi_channel = group.remote_exec(work_code)

# Send different tasks to each channel
for idx, channel in enumerate(multi_channel):
    channel.send(idx * 10)

# Collect results as they arrive
results = []
for channel in multi_channel:
    result = channel.receive()
    results.append(result)
    print(f"Task {result['task']} completed: {result['result']}")

# Cleanup
multi_channel.waitclose()
group.terminate(timeout=3.0)
```

--------------------------------

### Manage multiple remote channels with MultiChannel

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_multi.md

Demonstrates how to aggregate multiple remote channels into a single MultiChannel container for collective operations, such as receiving results from several sources simultaneously.

```python
import execnet
ch1 = execnet.makegateway().remote_exec("channel.send(1)")
ch2 = execnet.makegateway().remote_exec("channel.send(2)")
mch = execnet.MultiChannel([ch1, ch2])
len(mch)
# Output: 2
sum(mch.receive_each())
# Output: 3
```

--------------------------------

### Use channel callbacks for asynchronous data

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_info.md

Configures a callback function on a channel to process incoming data immediately as it arrives, rather than waiting for explicit receive calls.

```python
import execnet
gw = execnet.makegateway()
channel = gw.remote_exec("for i in range(10): channel.send(i)")
l = []
channel.setcallback(l.append, endmarker=None)
channel.waitclose()
```

--------------------------------

### Execute remote source code in a subprocess

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_info.md

Demonstrates how to instantiate a subprocess gateway and use remote_exec to run code that communicates bidirectionally through a channel.

```python
import execnet
gw = execnet.makegateway()
channel = gw.remote_exec("channel.send(channel.receive()+1)")
channel.send(1)
channel.receive()
```

--------------------------------

### Popen Gateway to Avoid Bytecode Writing

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Sets up a 'popen' gateway using the current Python executable and passes the '-B' flag to prevent writing of .pyc or .pyo files.

```default
popen//dont_write_bytecode
```

--------------------------------

### Threading Models

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Configuration for execution models across different thread environments.

```APIDOC
## execnet.set_execmodel(local_model, remote_model)

### Description
Configures the threading model for both sides of the gateway.

### Parameters
- **local_model** (str) - Required - Model for local side (gevent, eventlet, thread, main_thread_only).
- **remote_model** (str) - Required - Model for remote side (gevent, eventlet, thread, main_thread_only).
```

--------------------------------

### Understand execnet.default_group

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_group.md

Explains the implicit behavior of execnet.makegateway() and its association with the internal default_group singleton.

```python
import execnet
gw = execnet.makegateway()
print(gw in execnet.default_group)
print(execnet.default_group.defaultspec)
```

--------------------------------

### Assign Custom Gateway IDs

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_group.md

Shows how to assign specific identifiers to gateways during initialization for easier management and lookup within a group.

```python
import execnet
group = execnet.Group()
gw = group.makegateway("popen//id=sub1")
assert gw.id == "sub1"
print(group['sub1'])
```

--------------------------------

### Manage Gateway Membership and Lifecycle with execnet.Group

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_group.md

Demonstrates initializing a group of gateways, accessing them by index or ID, and performing bulk termination of member processes.

```python
import execnet
group = execnet.Group(['popen'] * 2)
len(group)
print(list(group))
group.terminate()
```

--------------------------------

### Setting Custom Execution Models in Python with execnet

Source: https://context7.com/pytest-dev/execnet/llms.txt

Configures execnet to use alternative threading models like 'thread' for gateway execution. It demonstrates setting the execution model, creating a gateway with it, and performing concurrent operations.

```python
import execnet

# Set execution model before creating gateways
# Options: 'thread', 'eventlet', 'gevent', 'main_thread_only'
execnet.set_execmodel("thread", "thread")

# Create gateway with specified model
gw = execnet.makegateway("popen//execmodel=thread")

# Check gateway configuration
print(gw)
# Output: <Gateway id='gw0' receive-live, thread model, 0 active channels>

# Get remote status
status = gw.remote_status()
print(f"Remote channels: {status.numchannels}")
print(f"Remote executing: {status.numexecuting}")

# Execute multiple concurrent operations
channels = []
for i in range(5):
    ch = gw.remote_exec(f"""
        import time
        time.sleep(0.1)
        channel.send({i} * 2)
    """)
    channels.append(ch)

# Collect results
for ch in channels:
    result = ch.receive()
    print(f"Result: {result}")

gw.exit()
```

--------------------------------

### Cross-interpreter Serialization with execnet.dumps/loads

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/hybridpython.md

Demonstrates how to serialize Python objects using execnet.dumps and deserialize them with execnet.loads, enabling data exchange between different Python interpreter versions (Python 2 and 3). This method is a safe and fast alternative to standard library modules like pickle and marshal for inter-version compatibility. It supports builtin Python data structures but not user-level instances.

```python
# using python2
import execnet
with open("data.py23", "wb") as f:
    f.write(execnet.dumps(["hello", "world"]))

# using Python3
import execnet
with open("data.py23", "rb") as f:
    val = execnet.loads(f.read(), py2str_as_py3str=True)
assert val == ["hello", "world"]
```

--------------------------------

### Retrieve Remote File Contents using servefiles

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_info.md

This code demonstrates how to retrieve the contents of remote files using a custom `servefiles.py` module over SSH with execnet. The server code reads specified files and sends their content, while the client code connects to the remote gateway, sends file paths, and receives the content.

```python
# content of servefiles.py

def servefiles(channel):
    for fn in channel:
        f = open(fn, "rb")
        channel.send(f.read())
        f.close()

if __name__ == "__channelexec__":
    servefiles(channel)  # type: ignore[name-defined]
```

```python
import execnet
import servefiles
gw = execnet.makegateway("ssh=codespeak.net")
channel = gw.remote_exec(servefiles)

for fn in ('/etc/passwd', '/etc/group'):
    channel.send(fn)
    content = channel.receive()
    print(fn)
    print(content)
```

--------------------------------

### Cross-Version Communication with Gateways

Source: https://context7.com/pytest-dev/execnet/llms.txt

Demonstrates inter-process communication using execnet gateways. A Python dictionary is sent to a remote process, processed (sum of numbers multiplied by 2), and the result is sent back. This showcases remote execution and channel communication.

```python
import execnet

data = {
    'numbers': [1, 2, 3, 4, 5],
    'nested': {'key': 'value', 'items': (10, 20, 30)},
    'mixed': [{'a': 1}, {'b': 2}]
}

gw = execnet.makegateway("popen")
channel = gw.remote_exec("""
    data = channel.receive()
    # Process data on remote side
    result = sum(data['numbers']) * 2
    channel.send(result)
""")

channel.send(data)
result = channel.receive()
print(f"Remote calculation result: {result}")

gw.exit()
```

--------------------------------

### Implement asynchronous callbacks for channels

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_multi.md

Uses setcallback to process channel data immediately upon receipt without blocking the main execution thread.

```python
import execnet
gw = execnet.makegateway()
ch = gw.remote_exec("channel.receive() ; channel.send(42)")
l = []
ch.setcallback(l.append)
ch.send(1)
ch.waitclose()
assert l == [42]
```

--------------------------------

### Connect to Socket Gateway from Client

Source: https://context7.com/pytest-dev/execnet/llms.txt

Establishes a connection to a socket gateway running on localhost and executes remote code. It sends information about the remote node and its system. Handles HostNotFound exceptions.

```python
import execnet
try:
    gw = execnet.makegateway("socket=localhost:8888")

    # Use the gateway normally
    channel = gw.remote_exec("""
        import platform
        channel.send({
            'node': platform.node(),
            'system': platform.system(),
            'machine': platform.machine()
        })
    """)

    info = channel.receive()
    print(f"Connected to: {info['node']}")
    print(f"System: {info['system']} on {info['machine']}")

    gw.exit()

except execnet.HostNotFound as e:
    print(f"Could not connect: {e}")
```

--------------------------------

### Popen Gateway with Specific Python and Nice Level

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Configures a 'popen' gateway to use a specific Python executable (python2.7) and sets a lower process priority using the 'nice' option.

```default
popen//python=python2.7//nice=20
```

--------------------------------

### Popen Gateway with Eventlet Execution Model

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Specifies a 'popen' gateway that uses the 'eventlet' execution model for handling I/O and dispatching threads in the remote process.

```default
popen//execmodel=eventlet
```

--------------------------------

### Enable execnet tracing via environment variables

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_debug.md

Demonstrates how to enable verbose debugging for execnet operations. Setting EXECNET_DEBUG to 2 redirects trace output to stderr, which is useful for monitoring gateway message passing.

```bash
EXECNET_DEBUG=2
python -c 'import execnet ; execnet.makegateway().remote_exec("42")'
```

--------------------------------

### Handle termination with endmarkers

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_multi.md

Configures a receive queue with an endmarker to detect when remote channels close or gateways terminate, ensuring robust process monitoring.

```python
group = execnet.Group(['popen'] * 3)
mch = group.remote_exec("channel.send(channel.receive()+1)")
queue = mch.make_receive_queue(endmarker=42)
mch[0].send(1)
chan1, res1 = queue.get()
group.terminate(timeout=1)
for i in range(3):
    chan1, res1 = queue.get()
    assert res1 == 42
```

--------------------------------

### Execute Python Code Remotely with remotecmd

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_info.md

This snippet demonstrates how to execute Python functions defined in a remote module (`remotecmd.py`) using execnet. It shows the remote server-side code that evaluates incoming commands and the client-side code to establish a gateway, send commands, and receive results.

```python
def simple(arg):
    return arg + 1

def listdir(path):
    return os.listdir(path)

if __name__ == "__channelexec__":
    for item in channel:  # type: ignore[name-defined]
        channel.send(eval(item))  # type: ignore[name-defined]
```

```python
import execnet, remotecmd
gw = execnet.makegateway()
ch = gw.remote_exec(remotecmd)
ch.send('simple(10)') # execute func-call remotely
print(ch.receive())
```

--------------------------------

### Query remote SSH environment information

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_info.md

Uses a gateway connection over SSH to execute code remotely and retrieve system details like platform information, version, and process ID.

```python
import execnet, os
gw = execnet.makegateway("ssh=codespeak.net")
channel = gw.remote_exec("""
    import sys, os
    channel.send((sys.platform, tuple(sys.version_info), os.getpid()))
""")
platform, version_info, remote_pid = channel.receive()
```

--------------------------------

### Remote-execute a module object

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_info.md

Demonstrates sending a module object to a remote process. The remote side can detect execution via the __channelexec__ special attribute.

```python
# content of a module remote1.py
if __name__ == "__channelexec__":
    channel.send("initialization complete")

# Execution
import execnet, remote1
gw = execnet.makegateway()
ch = gw.remote_exec(remote1)
print(ch.receive())
```

--------------------------------

### Process channel results using a queue

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_multi.md

Utilizes MultiChannel.make_receive_queue() to consolidate data from multiple channels into a single, blocking queue for ordered result retrieval.

```python
ch1 = execnet.makegateway().remote_exec("channel.send(1)")
ch2 = execnet.makegateway().remote_exec("channel.send(2)")
mch = execnet.MultiChannel([ch1, ch2])
queue = mch.make_receive_queue()
chan1, res1 = queue.get()
chan2, res2 = queue.get(timeout=3)
print(res1 + res2)
```

--------------------------------

### Cross-interpreter serialization with execnet.dumps and execnet.loads

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Shows how to serialize Python objects into bytestrings using execnet.dumps and deserialize them using execnet.loads. This functionality is crucial for exchanging data between different Python interpreters (e.g., Python 2 and Python 3, PyPy and CPython). The functions support conversion of string types between Python 2 and Python 3.

```python
import execnet

data = [1, 2, 3]
dumped_data = execnet.dumps(data)
loaded_data = execnet.loads(dumped_data)
print(loaded_data)

# Example with string conversion
# On Python 2, dumping a str and loading as py3str
# On Python 3, dumping a str and loading as py2str

```

--------------------------------

### Group Gateway Management

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Methods for managing gateway groups and enforcing robust process termination.

```APIDOC
## Group.terminate(timeout=None)

### Description
Triggers exit of all member gateways and waits for sub-process termination.

### Parameters
- **timeout** (float) - Optional - Seconds to wait before attempting to kill subprocesses.
```

--------------------------------

### Pre-allocate Gateway IDs

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_group.md

Provides a mechanism to determine a gateway's unique identifier before the actual instantiation of the gateway process.

```python
import execnet
group = execnet.Group()
spec = execnet.XSpec("popen")
group.allocate_id(spec)
allocated_id = spec.id
gw = group.makegateway(spec)
assert gw.id == allocated_id
```

--------------------------------

### Grouped Gateways and Termination

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Gateway instances are managed by groups, with `execnet.default_group` being the default. The `Group.terminate()` method ensures all member gateways and associated subprocesses are shut down gracefully, with an optional timeout for forceful killing.

```python
group.terminate(timeout=None)
```

--------------------------------

### Popen Gateway with Environment Variable

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Configures a 'popen' gateway to set a specific environment variable (NAME=value) in the remote process.

```default
popen//env:NAME=value
```

--------------------------------

### Remote Module Execution via Gateway

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Executes the source code of a Python module on a remote gateway. The `channel` object is placed in the module's global namespace.

```python
import my_module

gateway = execnet.makegateway("popen")
channel = gateway.remote_exec(my_module)
```

--------------------------------

### Coordinate Multiple Gateways with Groups

Source: https://context7.com/pytest-dev/execnet/llms.txt

Manages a collection of gateways using a Group object to facilitate parallel execution and centralized resource cleanup. This is ideal for performing identical tasks across multiple isolated environments.

```python
import execnet

# Create a group with multiple gateways
group = execnet.Group()

# Add different types of gateways
gw1 = group.makegateway("popen//id=local1")
gw2 = group.makegateway("popen//id=local2")
gw3 = group.makegateway("popen//python=python3//id=local3")

# Execute same code on all gateways in parallel
multi_channel = group.remote_exec("""
    import os
    channel.send({'id': channel.gateway.id, 'pid': os.getpid()})
""")

# Receive results from all gateways
for channel in multi_channel:
    info = channel.receive()
    print(f"Gateway {info['id']}: PID {info['pid']}")

# Terminate all gateways with timeout
group.terminate(timeout=5.0)
```

--------------------------------

### Command Service Pattern with execnet

Source: https://context7.com/pytest-dev/execnet/llms.txt

Implements a remote command service that evaluates function calls sent from the local side. The remote module exposes functions like 'get_info', 'add', 'multiply', and 'list_files', allowing the main script to execute them remotely and receive results.

```python
# File: command_service.py
"""Remote command service module."""
import os
import sys

def get_info():
    return {
        'platform': sys.platform,
        'python_version': sys.version,
        'cwd': os.getcwd()
    }

def add(a, b):
    return a + b

def multiply(a, b):
    return a * b

def list_files(path='.'):
    return os.listdir(path)

if __name__ == '__channelexec__':
    channel.send({'status': 'ready', 'available_commands': ['get_info', 'add', 'multiply', 'list_files']})

    for command in channel:
        try:
            result = eval(command)
            channel.send({'status': 'success', 'result': result})
        except Exception as e:
            channel.send({'status': 'error', 'message': str(e)})
```

```python
# Main code:
import execnet
import command_service

gw = execnetnet.makegateway()
channel = gw.remote_exec(command_service)

# Wait for service ready
init_msg = channel.receive()
print(f"Service status: {init_msg['status']}")
print(f"Available: {init_msg['available_commands']}")

# Execute commands
channel.send('add(10, 20)')
response = channel.receive()
print(f"add(10, 20) = {response['result']}")

channel.send('multiply(5, 7)')
response = channel.receive()
print(f"multiply(5, 7) = {response['result']}")

channel.send('get_info()')
response = channel.receive()
print(f"Remote info: {response['result']['platform']}")

channel.close()
gw.exit()
```

--------------------------------

### Remote Code Execution via Gateway

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Executes a string of Python source code on a remote gateway. The `channel` object is made available in the global namespace of the remote execution context.

```python
gateway = execnet.makegateway("popen")
channel = gateway.remote_exec("print('hello from remote')")
```

--------------------------------

### Serialize and Deserialize Data with execnet

Source: https://context7.com/pytest-dev/execnet/llms.txt

Serializes Python data structures to bytes using execnet.dumps and deserializes them back using execnet.loads. This process is compatible across different Python versions (2/3, PyPy, Jython).

```python
import execnet

data = {
    'numbers': [1, 2, 3, 4, 5],
    'nested': {'key': 'value', 'items': (10, 20, 30)},
    'mixed': [{'a': 1}, {'b': 2}]
}

serialized = execnet.dumps(data)
print(f"Serialized to {len(serialized)} bytes")

restored = execnet.loads(serialized)
print(f"Restored: {restored}")
assert restored == data
```

--------------------------------

### Execute Parametrized Functions Remotely

Source: https://context7.com/pytest-dev/execnet/llms.txt

Executes defined functions on a remote gateway by passing arguments, which is more robust than using inline code strings. This approach maintains a persistent connection for sending data during function execution.

```python
import execnet

def multiplier(channel, factor):
    """Remote function that multiplies received values by a factor."""
    while not channel.isclosed():
        try:
            param = channel.receive()
            channel.send(param * factor)
        except EOFError:
            break

# Create gateway and execute function with keyword arguments
gw = execnet.makegateway()
channel = gw.remote_exec(multiplier, factor=10)

# Send values and receive multiplied results
for i in range(5):
    channel.send(i)
    result = channel.receive()
    print(f"{i} * 10 = {result}")

channel.close()
gw.exit()
```

--------------------------------

### Socket Gateway Specification

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Defines a gateway using a socket connection, specifying the host and port for the remote Python server process.

```default
socket=192.168.1.4:8888
```

--------------------------------

### Channel Data Exchange

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Methods for sending and receiving data across remote channels, including callback management and file-like access.

```APIDOC
## Channel.send(item)

### Description
Sends a simple Python type item to the other side of the channel.

### Parameters
- **item** (any) - Required - The data item to send.

## Channel.receive(timeout=None)

### Description
Receives a data item from the other side. Blocks until an item is available or timeout occurs.

### Parameters
- **timeout** (float) - Optional - Seconds to wait before raising TimeoutError.

## Channel.setcallback(callback, endmarker=_NOENDMARKER)

### Description
Configures a callback function for asynchronous item reception.

## Channel.makefile(mode, proxyclose=False)

### Description
Returns a file-like object for stream-based reading or writing.

### Parameters
- **mode** (str) - Required - 'r' for reading, 'w' for writing.
- **proxyclose** (bool) - Optional - Whether closing the file also closes the channel.
```

--------------------------------

### Remote-execute a parametrized function

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_info.md

Executes a pure function remotely by passing it to remote_exec along with keyword arguments, maintaining a communication loop until the channel is closed.

```python
import execnet

def multiplier(channel, factor):
    while not channel.isclosed():
        param = channel.receive()
        channel.send(param * factor)

gw = execnet.makegateway()
channel = gw.remote_exec(multiplier, factor=10)

for i in range(5):
    channel.send(i)
    result = channel.receive()
    assert result == i * 10

gw.exit()
```

--------------------------------

### Perform Robust Gateway Termination

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_group.md

Utilizes group.terminate(timeout) to safely shut down gateways and ensure that remote processes are cleaned up properly after a specified timeout period.

```python
import execnet
group = execnet.Group()
gw = group.makegateway("popen//id=sleeper")
ch = gw.remote_exec("import time ; time.sleep(2.0)")
group.terminate(timeout=1.0)
```

--------------------------------

### Client: Retrieve Remote File Contents via SSH using execnet

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_ssh_fileserver.md

This Python client code utilizes execnet to connect to a remote SSH server and retrieve file contents. It establishes an SSH gateway, remotely executes the server-side `servefiles` function, sends a list of filenames to the server, and receives and prints the content for each file.

```python
import execnet
import sys

# Assuming servefiles.py is available on the remote host or can be sent
gw = execnet.makegateway("ssh=your_remote_user@your_remote_host") # Replace with your remote details
channel = gw.remote_exec(open(__file__).read() + "\nservefiles(channel)\nif __name__ == \"__channelexec__\":\n    servefiles(channel)")

files_to_retrieve = ['/etc/passwd', '/etc/group', '/path/to/nonexistent_file.txt'] # Example files

for fn in files_to_retrieve:
    channel.send(fn)
    content = channel.receive()
    print(f"--- Content of {fn} ---")
    print(content.decode('utf-8', errors='replace')) # Decode bytes to string for printing
    print("\n")

gw.exit()
```

--------------------------------

### Remote Function Execution via Gateway

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Remotely executes a Python function, passing any keyword arguments. A `channel` object is automatically added to the function's keyword arguments.

```python
def my_remote_func(x, channel):
    channel.send(x * 2)

gateway = execnet.makegateway("popen")
channel = gateway.remote_exec(my_remote_func, x=10)
result = channel.receive()
```

--------------------------------

### Channel Operations for Data Exchange

Source: https://github.com/pytest-dev/execnet/blob/master/doc/basics.md

Channel objects facilitate sending and receiving data between asynchronously running programs. Data is copied by value and must be simple Python types. Errors like OSError or channel.RemoteError can occur during operations.

```python
channel.send(item)
channel.receive(timeout=None)
channel.setcallback(callback, endmarker=_NOENDMARKER)
channel.makefile(mode='r', proxyclose=False)
channel.makefile(mode='w', proxyclose=False)
channel.close(error=None)
channel.waitclose(timeout=None)

# Exception types
channel.RemoteError
channel.TimeoutError
```

--------------------------------

### Server: Read Remote Files via SSH using execnet

Source: https://github.com/pytest-dev/execnet/blob/master/doc/example/test_ssh_fileserver.md

This Python code defines a function to read files from a remote server accessible via SSH. It iterates through filenames received on a channel, opens each file in binary read mode, sends its content, and closes the file. It's designed to be executed remotely using execnet.

```python
import execnet
import sys

def servefiles(channel):
    for fn in channel:
        try:
            with open(fn, "rb") as f:
                channel.send(f.read())
        except IOError as e:
            channel.send(f"Error reading {fn}: {e}")

if __name__ == "__channelexec__":
    servefiles(channel)  # type: ignore[name-defined]
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.