Distributed Workflows with Generic Multinode

Fuzzball supports a generic multinode implementation that allows you to run custom commands or scripts across multiple nodes without being tied to a specific parallel computing framework implementation. This is useful when you want full control over how processes are launched and coordinated across nodes, or when your application uses its own distributed communication mechanism.

Unlike the openmpi, mpich, or gasnet implementations where Fuzzball automatically wraps your command with the appropriate launcher (e.g., mpirun), the generic implementation gives you direct access to the multinode environment through a set of environment variables. You are responsible for launching and coordinating processes on remote nodes yourself.

When using the generic implementation, the following environment variables are available both as shell variables and for expansion within command arguments:

The MULTINODE_SSH_WRAPPER variable points to a wrapper that allows you to execute commands on remote nodes. Use it like: $MULTINODE_SSH_WRAPPER <hostname> <command> [args...].

By default, only the head node’s standard output and standard error are captured in the workflow logs. If you need output from other nodes to appear in the logs, set the MULTINODE_WRAPPER_FORWARD_STREAMS environment variable to 1. This forwards stdout and stderr from all worker nodes to the head node for centralized log collection.

The generic implementation supports variable expansion in command arguments. This allows you to reference multinode environment variables directly in your command line, and they will be expanded before execution.

Here is an example Fuzzfile that uses the generic implementation to run an MPI application with explicit control over how mpirun is invoked:

version: v4
jobs:
  generic-multinode:
    image:
      uri: oras://docker.io/anderbubble/openmpi-hello-world.sif@sha256:dc92ea0c541a8d9f30b4d6b78bfc57bd2fe9806689d93ece6069c6ed6519fa9a
    env:
      - PATH=/usr/lib64/openmpi/bin:/usr/local/bin:/usr/bin:/bin
    command:
      - mpirun
      - -H
      - $MULTINODE_HOSTLIST
      - --mca
      - plm_rsh_agent
      - $MULTINODE_SSH_WRAPPER
      - -np
      - $MULTINODE_TOTAL_SLOTS
      - /usr/lib64/openmpi/bin/mpi_hello_world
    resource:
      cpu:
        cores: 2
        affinity: NUMA
      memory:
        size: 1GB
    multinode:
      nodes: 2
      implementation: generic

In this example, the command field references $MULTINODE_HOSTLIST, $MULTINODE_SSH_WRAPPER, and $MULTINODE_TOTAL_SLOTS directly. These are expanded by the generic wrapper before the command is executed. This gives you full control over how the MPI launcher is configured, including the ability to pass custom options.

You can also use a script to coordinate work across nodes. In this mode, the multinode environment variables are available as standard shell environment variables within your script.

version: v4
jobs:
  generic-multinode-script:
    image:
      uri: docker://mirror.gcr.io/library/alpine:3.16
    env:
      - MULTINODE_WRAPPER_FORWARD_STREAMS=1
    script: |
      #!/bin/sh
      HOSTNAME=$(hostname)

      sleep_func() {
	      echo "$HOSTNAME is the head node"
        sleep 3
      }

      sleep_func &
      pid=$!

      IFS=','

      for HOST in ${MULTINODE_HOSTLIST_NOSLOTS}; do
        if [ "$HOSTNAME" != "$HOST" ]; then
          $MULTINODE_SSH_WRAPPER $HOST echo "$HOST is a worker node"
        fi
      done

      wait $pid
    resource:
      cpu:
        cores: 1
        affinity: NUMA
      memory:
        size: 1GB
    multinode:
      nodes: 2
      implementation: generic

In this example, the script iterates over the hostlist and uses $MULTINODE_SSH_WRAPPER to run a command on each remote node. The MULTINODE_WRAPPER_FORWARD_STREAMS=1 environment variable ensures that output from the worker nodes is forwarded to the head node and captured in the workflow logs.

After running this workflow, the logs will contain output from all nodes:

generic-multinode-script-0 is the head node
generic-multinode-script-1 is a worker node

Managing IFS, SSH wrapper loops, and background processes by hand across every script is repetitive and error-prone. The generic implementation provides a helper script at /multinode/generic-helper that encapsulates these patterns into four reusable shell functions.

/multinode/generic-helper is only available when multinode.implementation is set to generic. It is not present for openmpi, mpich, or gasnet implementations.

Source the helper at the top of your script before calling any of its functions:

#!/bin/sh
. /multinode/generic-helper

Once sourced, the helper captures the current node’s hostname as the rank 0 (head node) identifier. All subsequent function calls use this to distinguish the head node from worker nodes.

Runs a command or script on rank 0 (the head node) in the background, allowing worker node launches to proceed concurrently. The background process ID is stored in $rank0_pid for use with wait_rank0.

exec_rank0 'command to run on rank 0'

The argument is a shell expression that is evaluated on rank 0. To pass a multi-line script, either pass a single quoted string or use - to read from stdin:

exec_rank0 - <<'EOF'
echo "starting head node"
my-app --role head --addr "$MULTINODE_NODE_IP"
EOF

Runs a command or script on every worker node (rank 1, 2, …) sequentially using $MULTINODE_SSH_WRAPPER. The script is forwarded verbatim and interpreted by each worker’s shell, where $HOSTNAME and $RANK are set to that node’s hostname and numeric rank.

exec_rankN 'command to run on each worker node'

Worker nodes are derived from $MULTINODE_HOSTLIST_NOSLOTS with rank 0 excluded. The function also accepts - to read a script from stdin, identical to exec_rank0.

exec_rankN dispatches the command to each worker node and returns without waiting for it to complete; the remote exit status is not reported back to the calling script. If a worker command fails, the failure is propagated asynchronously to the head node and fails the job.

Runs a command or script on all nodes: rank 0 in the background (via exec_rank0), then each worker node (via exec_rankN), and finally waits for rank 0 to finish. This is the single-call equivalent of exec_rank0 + exec_rankN + wait_rank0.

exec_all 'command to run on every node'

Use exec_all when all nodes run the same command, for example launching a distributed framework that handles rank detection internally.

For multi-line scripts, pass the body via a heredoc. When called with no arguments, exec_all reads the script from stdin automatically:

exec_all <<'EOF'
echo "starting on $HOSTNAME"
my-distributed-app --rank "$RANK"
EOF

To also pass positional arguments to the script, use - as the first argument (to explicitly signal stdin reading) followed by the argument values. Inside the heredoc, reference them as $1, $2, or $@:

MODEL_PATH=/models/my-model
BATCH_SIZE=32

exec_all - "$MODEL_PATH" "$BATCH_SIZE" "$MULTINODE_NODE_IP" <<'EOF'
echo "node $HOSTNAME loading model $1 with batch size $2"
my-distributed-app --model "$1" --batch-size "$2" --head-addr "$3"
EOF

The argument values are expanded before the SSH wrapper forwards the script to each remote node, so $1, $2, and $@ resolve to the same values on all nodes.

Waits for the rank 0 background process (started by exec_rank0) to complete, then exits the calling script with rank 0’s exit status. This is analogous to wait $pid but uses the $rank0_pid set internally by exec_rank0.

wait_rank0

Always call wait_rank0 after exec_rank0 (unless you use exec_all, which calls it automatically) to avoid the head node script exiting before rank 0’s work finishes.

Scripts passed to exec_rankN and exec_all are forwarded to worker nodes verbatim, so shell quoting decides where variables expand:

• Single quotes (or a quoted heredoc such as <<'EOF') defer expansion to the node where the script runs: $HOSTNAME, $RANK, and any variables defined in the job’s env section resolve to each node’s own values.
• Double quotes (or an unquoted heredoc) expand on the head node before forwarding, so every node receives the same value.

The runtime-computed MULTINODE_* variables (such as $MULTINODE_NODE_IP and $MULTINODE_HOSTLIST) are only set in the head node’s script environment. To use them in a worker script, expand them on the head node with double quotes, or pass them as positional arguments.

The following workflow is equivalent to the manual script example above but uses the helper to eliminate explicit IFS management, the SSH wrapper loop, and background PID tracking:

version: v4
jobs:
  generic-multinode-helper:
    image:
      uri: docker://mirror.gcr.io/library/alpine:3.16
    env:
      - MULTINODE_WRAPPER_FORWARD_STREAMS=1
    script: |
      #!/bin/sh
      . /multinode/generic-helper

      exec_rank0 'echo "$HOSTNAME is the head node"; sleep 3'
      exec_rankN 'echo "$HOSTNAME is a worker node"'
      wait_rank0
    resource:
      cpu:
        cores: 1
        affinity: NUMA
      memory:
        size: 1GB
    multinode:
      nodes: 2
      implementation: generic

After running this workflow, the logs will contain output from all nodes:

generic-multinode-helper-0 is the head node
generic-multinode-helper-1 is a worker node

When the head node and worker nodes run different commands — a common pattern for inference servers and distributed training — use exec_rank0 and exec_rankN separately:

script: |
  #!/bin/sh
  . /multinode/generic-helper

  exec_rank0 'my-server --role head --addr "$MULTINODE_NODE_IP"'
  exec_rankN "my-server --role worker --head-addr $MULTINODE_NODE_IP"
  wait_rank0

Note the quoting: the exec_rankN command uses double quotes so $MULTINODE_NODE_IP expands on the head node — every worker receives the head node’s address. The exec_rank0 command runs on the head node itself, where $MULTINODE_NODE_IP is available either way.

When all nodes run the same command and the application determines its own role (e.g. by reading its hostname or an externally set rank), use exec_all:

script: |
  #!/bin/sh
  . /multinode/generic-helper
  exec_all "my-app --hostlist $MULTINODE_HOSTLIST --head-addr $MULTINODE_NODE_IP"

The generic implementation is a good fit when:

• You need to run a distributed framework that doesn’t match one of the supported implementations.
• You want explicit control over how processes are launched across nodes.
• Your application has its own mechanism for distributed coordination.
• You want to run custom shell scripts that orchestrate work across multiple nodes.

For standard MPI workloads, the openmpi or mpich implementations are recommended as they handle the mpirun invocation automatically. See Distributed Workflows with MPI for details.

You can also create long-running distributed services like inference servers or interactive computing environments. Services use the same multinode configuration but add network endpoints and persistence options.

For details on configuring host networking for multinode services, including:

• When and why to enable host networking
• How to expose endpoints on rank 0
• Example distributed inference service configurations

See Host Networking for Multinode Services.