Enable hybrid CUDA IPC and EFA GPU exchange

presto/docker/config/template/etc_coordinator/config_native.properties

@@ -23,6 +23,10 @@ memory.heap-headroom-per-node={{ .HeadroomGb }}GB
 node-scheduler.max-pending-splits-per-task=2000
 # Cap concurrent splits per node for balanced scheduling.
 node-scheduler.max-splits-per-node=2000
+# Preserve deterministic split-to-worker affinity for hot cache reuse. Use the
+# scheduler default-sized ring so a larger worker set is not underrepresented.
+node-scheduler.node-selection-hash-strategy=CONSISTENT_HASHING
+node-scheduler.consistent-hashing-min-virtual-node-count=1000
 
 # Optimizer flags
 # Use known constraints to simplify plan and filters.

presto/docker/config/template/overrides/gpu/etc_worker/config_native.properties

@@ -11,13 +11,17 @@ cudf.jit_expression_enabled=false
 # Turn on to use intra-node exchange optimization.
 # NOTE: In cudf exchange 20260212 branch, this is needed for UCX to use nvlink.
 cudf.intra_node_exchange=true
+# EFA/SRD requires active UCXX progress polling and transports that do not
+# provide peer-failure handling.
+ucxx.error_handling=false
+ucxx.blocking_polling=false
 
-# Use 100M rows per chunk for cudf partitioned output.
+# Bound partition output for SF3K Q18 memory pressure.
 # NOTE: This is not yet propagated to the worker properly because only a fixed set of query configs are supported.
 #       https://github.com/prestodb/presto/blob/a62672886152c8c6b61cf301d246f217d850e357/presto-native-execution/presto_cpp/main/PrestoToVeloxQueryConfig.cpp#L106-L224
 #       As a result, this needs to be hardcoded right now.
-cudf.partitioned_output_batch_rows=100000000
+cudf.partitioned_output_batch_rows=10000000
 
 # Enable cudf rebatching before aggregations.
 cudf.concat_optimization_enabled=true
-cudf.batch_size_min_threshold=100000000
+cudf.batch_size_min_threshold=40000000

presto/docker/docker-compose.common.yml

@@ -42,6 +42,7 @@ services:
         - VELOX_SHA=${VELOX_SHA:-}
         - VELOX_BRANCH=${VELOX_BRANCH:-}
         - VELOX_REPOSITORY=${VELOX_REPO:-}
+        - UCX_VERSION=${UCX_VERSION:-1.22.0}
     environment:
       - GLOG_logtostderr=1
       - SERVER_START_TIMESTAMP=${SERVER_START_TIMESTAMP:-}

presto/docker/docker-compose/template/docker-compose.native-gpu.yml.jinja

@@ -30,19 +30,14 @@ x-presto-native-worker-gpu: &gpu_worker_base
             - driver: nvidia
               count: all
               capabilities: [gpu]
-  # devices:
-    #   # required for GDS
-    #   - /dev/infiniband/rdma_cm
-    #   - /dev/infiniband/uverbs0
-    #   - /dev/infiniband/uverbs1
-    #   - /dev/infiniband/uverbs2
-    #   - /dev/infiniband/uverbs3
-    #   - /dev/infiniband/uverbs4
-    #   - /dev/infiniband/uverbs5
-    #   - /dev/infiniband/uverbs6
-    #   - /dev/infiniband/uverbs7
-    #   - /dev/infiniband/uverbs8
-    #   - /dev/infiniband/uverbs9
+{% if ucx_efa %}
+  devices:
+    - /dev/infiniband/rdma_cm
+    - /dev/infiniband/uverbs0
+    - /dev/infiniband/uverbs1
+    - /dev/infiniband/uverbs2
+    - /dev/infiniband/uverbs3
+{% endif %}
   depends_on:
     - presto-coordinator
   volumes:
@@ -70,18 +65,29 @@ services:
   presto-native-worker-gpu-{{ gpu_id }}:
     <<: *gpu_worker_base
     container_name: presto-native-worker-gpu-{{ gpu_id }}
+{% if ucx_efa %}
+    network_mode: host
+{% endif %}
     environment:
       WORKER_ID: {{ gpu_id }}
       NVIDIA_VISIBLE_DEVICES: all
       PROFILE: ${PROFILE:-}
       PROFILE_ARGS: ${PROFILE_ARGS:-}
       SERVER_START_TIMESTAMP: ${SERVER_START_TIMESTAMP:-}
-      UCX_LOG_LEVEL: info #debug
+{% if ucx_efa %}
+      UCX_RNDV_FRAG_SIZE: ${UCX_RNDV_FRAG_SIZE:-cuda:32M}
+      UCX_RNDV_FRAG_MEM_TYPES: ${UCX_RNDV_FRAG_MEM_TYPES:-cuda}
+      UCX_MAX_RNDV_RAILS: ${UCX_MAX_RNDV_RAILS:-2}
+      UCX_TLS: ${UCX_TLS:-tcp,srd,sm,cuda_copy,cuda_ipc}
+      UCX_NET_DEVICES: ${UCX_NET_DEVICES_GPU_{{ gpu_id }}:-}
+      UCX_SOCKADDR_TLS_PRIORITY: ${UCX_SOCKADDR_TLS_PRIORITY:-tcp}
+{% else %}
       UCX_RNDV_PIPELINE_ERROR_HANDLING: y
       UCX_TLS: tcp,cuda_copy,cuda_ipc
+{% endif %}
+      UCX_LOG_LEVEL: info #debug
       UCX_TCP_CM_REUSEADDR: y
       UCX_PROTO_INFO: y
-      # don't know, leave it for now
       UCX_TCP_KEEPINTVL: 1ms
       UCX_KEEPALIVE_INTERVAL: 1ms
       KVIKIO_NTHREADS: {{ kvikio_threads }}
@@ -97,19 +103,32 @@ services:
   presto-native-worker-gpu:
     <<: *gpu_worker_base
     container_name: presto-native-worker-gpu
+{% if ucx_efa %}
+    network_mode: host
+{% endif %}
 {%- if workers %}
     command: ["bash", "/opt/presto_profiling_wrapper.sh"{% for gpu_id in workers %}, "{{ gpu_id }}"{% endfor %}]
     environment:
       NVIDIA_VISIBLE_DEVICES: all
       PROFILE: ${PROFILE:-}
       PROFILE_ARGS: ${PROFILE_ARGS:-}
       SERVER_START_TIMESTAMP: ${SERVER_START_TIMESTAMP:-}
-      UCX_LOG_LEVEL: info #debug
+{% if ucx_efa %}
+      UCX_RNDV_FRAG_SIZE: ${UCX_RNDV_FRAG_SIZE:-cuda:32M}
+      UCX_RNDV_FRAG_MEM_TYPES: ${UCX_RNDV_FRAG_MEM_TYPES:-cuda}
+      UCX_MAX_RNDV_RAILS: ${UCX_MAX_RNDV_RAILS:-2}
+      UCX_TLS: ${UCX_TLS:-tcp,srd,sm,cuda_copy,cuda_ipc}
+      UCX_SOCKADDR_TLS_PRIORITY: ${UCX_SOCKADDR_TLS_PRIORITY:-tcp}
+{% for gpu_id in workers %}
+      UCX_NET_DEVICES_GPU_{{ gpu_id }}: ${UCX_NET_DEVICES_GPU_{{ gpu_id }}:-}
+{% endfor %}
+{% else %}
       UCX_RNDV_PIPELINE_ERROR_HANDLING: y
       UCX_TLS: tcp,cuda_copy,cuda_ipc
+{% endif %}
+      UCX_LOG_LEVEL: info #debug
       UCX_TCP_CM_REUSEADDR: y
       UCX_PROTO_INFO: y
-      # don't know, leave it for now
       UCX_TCP_KEEPINTVL: 1ms
       UCX_KEEPALIVE_INTERVAL: 1ms
       KVIKIO_NTHREADS: {{ kvikio_threads }}
@@ -120,12 +139,20 @@ services:
       PROFILE: ${PROFILE:-}
       PROFILE_ARGS: ${PROFILE_ARGS:-}
       SERVER_START_TIMESTAMP: ${SERVER_START_TIMESTAMP:-}
-      UCX_LOG_LEVEL: info #debug
+{% if ucx_efa %}
+      UCX_RNDV_FRAG_SIZE: ${UCX_RNDV_FRAG_SIZE:-cuda:32M}
+      UCX_RNDV_FRAG_MEM_TYPES: ${UCX_RNDV_FRAG_MEM_TYPES:-cuda}
+      UCX_MAX_RNDV_RAILS: ${UCX_MAX_RNDV_RAILS:-2}
+      UCX_TLS: ${UCX_TLS:-tcp,srd,sm,cuda_copy,cuda_ipc}
+      UCX_NET_DEVICES: ${UCX_NET_DEVICES:-}
+      UCX_SOCKADDR_TLS_PRIORITY: ${UCX_SOCKADDR_TLS_PRIORITY:-tcp}
+{% else %}
       UCX_RNDV_PIPELINE_ERROR_HANDLING: y
       UCX_TLS: tcp,cuda_copy,cuda_ipc
+{% endif %}
+      UCX_LOG_LEVEL: info #debug
       UCX_TCP_CM_REUSEADDR: y
       UCX_PROTO_INFO: y
-      # don't know, leave it for now
       UCX_TCP_KEEPINTVL: 1ms
       UCX_KEEPALIVE_INTERVAL: 1ms
       KVIKIO_NTHREADS: {{ kvikio_threads }}

presto/docker/launch_presto_servers.sh

@@ -49,6 +49,15 @@ launch_worker() {
     fi
 
     cuda_env=("CUDA_VISIBLE_DEVICES=$worker_id")
+    local ucx_device_env_name="UCX_NET_DEVICES_GPU_${worker_id}"
+    local ucx_device="${UCX_NET_DEVICES:-}"
+    if [[ -n "${!ucx_device_env_name:-}" ]]; then
+      ucx_device="${!ucx_device_env_name}"
+    fi
+    if [[ -n "$ucx_device" ]]; then
+      cuda_env+=("UCX_NET_DEVICES=$ucx_device")
+      echo "UCX network device: $ucx_device"
+    fi
     gpu_name="$(nvidia-smi --query-gpu=name --format=csv,noheader 2>/dev/null -i "$worker_id")"
   # No GPU: fall back to NUMA interleaving across all nodes for CPU workers.
   # Requires SYS_NICE capability in the container (set via cap_add in docker-compose).

presto/docker/native_build.dockerfile

@@ -7,6 +7,33 @@ RUN rpm --import https://developer.download.nvidia.com/compute/cuda/repos/ubuntu
     dnf install -y nsight-systems-cli-2026.3.1 numactl
 
 ARG GPU=ON
+ARG UCX_VERSION=1.22.0
+COPY velox-testing/presto/docker/ucx-11865-rtx-ipc-bandwidth.patch /tmp/ucx-11865-rtx-ipc-bandwidth.patch
+RUN if [ "$GPU" = "ON" ]; then \
+    dnf install -y patch rdma-core-devel && \
+    mkdir -p /tmp/ucx-src /tmp/ucx-build && \
+    wget -qO- "https://github.com/openucx/ucx/releases/download/v${UCX_VERSION}/ucx-${UCX_VERSION}.tar.gz" \
+      | tar -C /tmp/ucx-src --strip-components=1 -xz && \
+    patch -d /tmp/ucx-src -p1 < /tmp/ucx-11865-rtx-ipc-bandwidth.patch && \
+    cd /tmp/ucx-build && \
+    /tmp/ucx-src/contrib/configure-release \
+      --prefix=/usr/local \
+      --with-sysroot \
+      --enable-cma \
+      --enable-mt \
+      --with-gnu-ld \
+      --with-rdmacm \
+      --with-verbs \
+      --without-gda \
+      --without-go \
+      --without-java \
+      --with-cuda=/usr/local/cuda && \
+    make -j"$(nproc)" && \
+    make install && \
+    ldconfig && \
+    rm -rf /tmp/ucx-src /tmp/ucx-build /tmp/ucx-11865-rtx-ipc-bandwidth.patch; \
+    fi
+
 ARG BUILD_TYPE=release
 ARG BUILD_BASE_DIR=/presto_native_${BUILD_TYPE}_gpu_${GPU}_build
 ARG NUM_THREADS=12

presto/docker/ucx-11865-rtx-ipc-bandwidth.patch

@@ -0,0 +1,25 @@
+diff --git a/src/uct/cuda/base/cuda_iface.h b/src/uct/cuda/base/cuda_iface.h
+index a1cb28a4e14..a7f9a3688f1 100644
+--- a/src/uct/cuda/base/cuda_iface.h
++++ b/src/uct/cuda/base/cuda_iface.h
+@@ -22,5 +22,6 @@ typedef enum uct_cuda_base_gen {
+     UCT_CUDA_BASE_GEN_V100 = 7,
+     UCT_CUDA_BASE_GEN_A100 = 8,
+     UCT_CUDA_BASE_GEN_H100 = 9,
+-    UCT_CUDA_BASE_GEN_B100 = 10
++    UCT_CUDA_BASE_GEN_B100 = 10,
++    UCT_CUDA_BASE_GEN_BRTX = 12  /* Blackwell RTX */
+ } uct_cuda_base_gen_t;
+diff --git a/src/uct/cuda/cuda_ipc/cuda_ipc_iface.c b/src/uct/cuda/cuda_ipc/cuda_ipc_iface.c
+index ed09c8fa127..a7c169cf07a 100644
+--- a/src/uct/cuda/cuda_ipc/cuda_ipc_iface.c
++++ b/src/uct/cuda/cuda_ipc/cuda_ipc_iface.c
+@@ -197,5 +197,7 @@ static double uct_cuda_ipc_iface_get_bw()
+     case UCT_CUDA_BASE_GEN_B100:
+         return 800000.0 * UCS_MBYTE;
++    case UCT_CUDA_BASE_GEN_BRTX:
++        return 50000.0 * UCS_MBYTE;
+     default:
+         return 6911.0  * UCS_MBYTE;
+     }
+ }

presto/scripts/generate_presto_config.sh

@@ -60,6 +60,21 @@ function duplicate_worker_configs() {
   sed -i "s+cudf.exchange.server.port=.*+cudf.exchange.server.port=${exch_port}+g" ${worker_native_config}
   # Give each worker a unique id.
   sed -i "s+node\.id.*+node\.id=worker_${worker_id}+g" ${worker_config}/node.properties
+
+  # EFA workers use host networking. They share the host address, use unique
+  # HTTP/UCX ports, and reach the coordinator through its published host port.
+  if [[ "${PRESTO_WORKER_HOST_NETWORK:-false}" == "true" ]]; then
+    local internal_address="${PRESTO_WORKER_INTERNAL_ADDRESS:?PRESTO_WORKER_INTERNAL_ADDRESS must be set for host-network workers}"
+    sed -i "s+discovery\.uri=.*+discovery.uri=http://127.0.0.1:8080+g" \
+      "${worker_config}/config_native.properties" \
+      "${worker_config}/config_java.properties"
+    if grep -q '^node\.internal-address=' "${worker_config}/node.properties"; then
+      sed -i "s+node\.internal-address=.*+node.internal-address=${internal_address}+g" \
+        "${worker_config}/node.properties"
+    else
+      echo "node.internal-address=${internal_address}" >> "${worker_config}/node.properties"
+    fi
+  fi
 }
 
 # get host values

presto/scripts/run_g7e48_local_nvme_q18.sh

@@ -0,0 +1,82 @@
+#!/usr/bin/env bash
+# Reproduce the compression-free, local-NVMe SF3K Q18 EFA/SRD benchmark on
+# the audited g7e.48xlarge topology. Iteration 1 warms caches; iteration 2 is
+# the reported hot result.
+set -Eeuo pipefail
+
+SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)
+
+: "${PRESTO_IMAGE_TAG:?set PRESTO_IMAGE_TAG to the coordinator/worker image tag}"
+: "${PRESTO_DATA_DIR:?set PRESTO_DATA_DIR to the parent of sf3k_v2_float on NVMe}"
+SCHEMA=${SCHEMA:-tianyu_nvme_sf3k_v2_float}
+RESULTS_DIR=${RESULTS_DIR:-$HOME/Development/results}
+RESULT_TAG=${RESULT_TAG:-g7e48_local_nvme_sf3k_q18_ucx_srd_no_compression_i2}
+
+if [[ -z ${PRESTO_WORKER_INTERNAL_ADDRESS:-} ]]; then
+  PRESTO_WORKER_INTERNAL_ADDRESS=$(ip -4 -o addr show dev enp135s0 | awk '{split($4, a, "/"); print a[1]; exit}')
+  export PRESTO_WORKER_INTERNAL_ADDRESS
+fi
+: "${PRESTO_WORKER_INTERNAL_ADDRESS:?could not determine the host private IPv4 address}"
+
+export UCX_NET_DEVICES_GPU_0='rdmap145s0:1,enp135s0'
+export UCX_NET_DEVICES_GPU_1='rdmap145s0:1,enp135s0'
+export UCX_NET_DEVICES_GPU_2='rdmap162s0:1,enp135s0'
+export UCX_NET_DEVICES_GPU_3='rdmap162s0:1,enp135s0'
+export UCX_NET_DEVICES_GPU_4='rdmap179s0:1,enp135s0'
+export UCX_NET_DEVICES_GPU_5='rdmap179s0:1,enp135s0'
+export UCX_NET_DEVICES_GPU_6='rdmap196s0:1,enp135s0'
+export UCX_NET_DEVICES_GPU_7='rdmap196s0:1,enp135s0'
+
+export UCX_TLS='tcp,srd,cuda_copy'
+export UCX_SOCKADDR_TLS_PRIORITY=tcp
+export UCX_RNDV_FRAG_SIZE='cuda:32M'
+export UCX_RNDV_FRAG_MEM_TYPES=cuda
+export UCX_MAX_RNDV_RAILS=1
+
+for device in rdma_cm uverbs0 uverbs1 uverbs2 uverbs3; do
+  [[ -c "/dev/infiniband/$device" ]] || {
+    echo "Missing /dev/infiniband/$device" >&2
+    exit 1
+  }
+done
+
+for nic in enp135s0 enp153s0 enp170s0 enp187s0; do
+  sudo -n ethtool -L "$nic" combined 16
+  [[ $(ethtool -l "$nic" | awk '/Current hardware settings:/{current=1; next} current && /Combined:/{print $2; exit}') == 16 ]] || {
+    echo "$nic does not have 16 combined queues" >&2
+    exit 1
+  }
+done
+
+start_args=(
+  --ucx-efa
+  --overwrite-config
+  --single-container
+  --kvikio-threads 16
+  --num-drivers 2
+  --num-workers 8
+  --gpu-ids 0,1,2,3,4,5,6,7
+)
+if [[ ${BUILD_WORKER:-false} == true ]]; then
+  start_args+=(--build worker --num-threads "$(nproc)")
+fi
+if [[ ${NO_CACHE_BUILD:-false} == true ]]; then
+  start_args+=(--no-cache)
+fi
+
+"$SCRIPT_DIR/start_native_gpu_presto.sh" "${start_args[@]}"
+
+if grep -Rqs '^cudf\.exchange_compression=' \
+    "$SCRIPT_DIR/../docker/config/generated/gpu"/etc_worker_*/config_native.properties; then
+  echo "Compression configuration unexpectedly present in generated workers" >&2
+  exit 1
+fi
+
+"$SCRIPT_DIR/run_benchmark.sh" \
+  --benchmark-type tpch \
+  --queries 18 \
+  --schema-name "$SCHEMA" \
+  --iterations 2 \
+  --output-dir "$RESULTS_DIR" \
+  --tag "$RESULT_TAG" \
+  --skip-analyze-check

presto/scripts/start_presto_helper.sh

@@ -180,7 +180,7 @@ if [[ "$VARIANT_TYPE" == "gpu" ]]; then
   LOCAL_NUM_WORKERS="${NUM_WORKERS:-0}"
 
   RENDER_SCRIPT_PATH=$(readlink -f "${SCRIPT_DIR}/../../template_rendering/render_docker_compose_template.py")
-  RENDER_ARGS="--template-path $TEMPLATE_PATH --output-path $RENDERED_PATH --num-workers $NUM_WORKERS --single-container $SINGLE_CONTAINER --kvikio-threads $KVIKIO_THREADS --sccache $ENABLE_SCCACHE"
+  RENDER_ARGS="--template-path $TEMPLATE_PATH --output-path $RENDERED_PATH --num-workers $NUM_WORKERS --single-container $SINGLE_CONTAINER --kvikio-threads $KVIKIO_THREADS --sccache $ENABLE_SCCACHE --ucx-efa $UCX_EFA"
   if [[ -n $GPU_IDS ]]; then
     RENDER_ARGS="$RENDER_ARGS --gpu-ids $GPU_IDS"
   fi