DYAD Service API

The DYAD service runs on the producer side, serving file data to consumers via RPC for coordination and a communication channel optimizated for data transport.

Flux module

DYAD Flux module runs as a broker plugin of flux-core. It is loaded using flux module load dyad.so [options] [producer_path].

DYAD Flux broker module implementation.

Implements the DYAD service as a Flux broker module. Flux services are implemented as dynamically loaded broker plugins (“broker modules”) that register message handlers for their methods and run the Flux reactor to remain responsive while handling requests from multiple clients concurrently using event-driven (reactive) programming techniques.

This module can be loaded using:

flux module load dyad.so [options] [producer_path]

Available options:

-h, help Show help and exit without loading the module.
-d, debug Enable debug logging.
-m, mode DTL mode (FLUX_RPC or UCX).
-i, info_log Redirect info logging to a file (requires -DDYAD_LOGGER=PRINTF at build time).
-e, error_log Redirect error logging to a file (requires -DDYAD_LOGGER=PRINTF at build time).

Defines

TIME_DIFF(Tstart, Tend)

Typedefs

typedef struct dyad_mod_ctx dyad_mod_ctx_t

Context structure for the DYAD Flux module.

Holds the Flux message handler table and the DYAD context for a running module instance. Allocated per broker handle via get_mod_ctx() and freed at module finalization time via freectx().

typedef struct opt_parse_out opt_parse_out_t

Functions

static void dyad_mod_fini(void)

Finalizes the DYAD Flux module at library unload time.

Registered as a destructor via ((destructor)). Called automatically when the module shared library is unloaded by the broker.

Note

If DYAD_PROFILER_DFTRACER is defined, finalizes the DFTracer profiler. Flux handle operations are intentionally omitted here as calling flux_open() at finalization time is known to cause errors.

static void freectx(void *arg)

Frees the DYAD module context at Flux module finalization time.

Registered as the destructor callback for the "dyad" auxiliary data on the Flux handle via flux_aux_set(). Called by the Flux broker when the module is unloaded. Releases the message handler table, finalizes the DYAD context via dyad_ctx_fini(), and frees the context struct.

Parameters:: arg – [in] Pointer to the dyad_mod_ctx_t to free. Cast from void* as required by the flux_free_f signature.

static dyad_mod_ctx_t *get_mod_ctx(void **h)

Retrieves or allocates the DYAD module context for a Flux handle.

Looks up the dyad_mod_ctx_t associated with h via flux_aux_get(). If none exists, allocates a new one, initializes it to NULL, and registers it with flux_aux_set() so that freectx() is called automatically at module finalization time.

Parameters:: h – [in] Flux handle to look up or register the context on.
Returns:: Pointer to the dyad_mod_ctx_t, or NULL if allocation or registration failed.

static void dyad_fetch_request_cb(void **h, flux_msg_handler_t *w, const void **msg, void *arg)

Flux message handler callback that serves file data to a consumer via RPC.

Registered as the handler for DYAD_DTL_RPC_NAME requests in htab. Invoked by the Flux reactor when a consumer dispatches an RPC to the producer’s broker requesting file data. Performs the following steps:

Validates that the incoming message is a streaming RPC.
Unpacks the relative file path (upath) from the RPC payload via dtl_handle->rpc_unpack().
Sends an initial RPC response to acknowledge the request via dtl_handle->rpc_respond().
Resolves the full file path by combining prod_managed_path and upath.
Opens the file and acquires a shared lock via dyad_shared_flock() to allow concurrent reads while blocking exclusive (producer) locks.
Reads the file contents into a DTL buffer. For large files (at or above DYAD_POSIX_TRANSFER_GRANULARITY bytes), reads in chunks.
Releases the shared lock, establishes a DTL connection to the consumer via dtl_handle->establish_connection(), and sends the data via dtl_handle->send().
Closes the DTL connection and signals end-of-stream to the consumer by responding with ENODATA via flux_respond_error().

On any error, responds to the consumer with the current errno via flux_respond_error() and returns. The shared lock and file descriptor are released before returning in all error paths.

When built with UCX DTL support (DYAD_ENABLE_UCX_DTL), the file size is prepended to the DTL buffer so the consumer can locate the data boundary without an additional RMA call.

When built with DYAD_SPIN_WAIT, spins on get_stat() before opening the file to wait for it to become accessible.

Note

This function is an internal Flux message handler and is not intended to be called directly. It is registered via htab in mod_main().

Note

The shared lock acquired in step 5 coordinates with the exclusive lock held by the producer during a write. Because POSIX fcntl locks are cooperative, this only provides guarantees between processes that also participate in locking.

Parameters:

h – [in] Flux handle for the broker.
w – [in] Flux message handler (unused directly).
msg – [in] Incoming Flux RPC message containing the file path packed by the consumer.
arg – [in] Auxiliary argument (the Flux handle, passed as void* from flux_msg_handler_addvec()).

static void show_help(void)

int opt_parse(opt_parse_out_t *opt, const unsigned broker_rank, int argc, char **argv)

Parses command-line arguments passed to the DYAD Flux module.

Parses argc / argv using getopt_long(). Because Flux module argument vectors do not include the executable name in argv[0] (unlike standard main()), a synthetic _argv is constructed with a NULL dummy first element so that getopt() works correctly. optind is reset to 1 before each call to handle repeated invocations.

Recognized options:

-h / help Sets opt->showed_help and returns.
-d / debug Sets opt->debug.
-m / mode Sets opt->dtl_mode.
-i / info_log Redirects info log output to a per-rank file.
-e / error_log Redirects error log output to a per-rank file.

Any remaining non-option argument is treated as the producer-managed directory path and stored in opt->prod_managed_path.

Log redirection options have no effect unless DYAD was built with -DDYAD_LOGGER=PRINTF.

Parameters:

opt – [out] Output structure populated with parsed options. Must not be NULL.
broker_rank – [in] Broker rank, used to name per-rank log files.
argc – [in] Number of module arguments.
argv – [in] Module argument strings.

Return values:

DYAD_RC_OK – Parsing succeeded.
DYAD_RC_SYSFAIL – An unrecognized option was encountered.

Returns:

dyad_rc_t return code indicating the outcome:

dyad_rc_t dyad_module_ctx_init(const opt_parse_out_t *opt, void **h)

Initializes the DYAD context for the Flux module.

Configures the DYAD context for use as a Flux module (producer side), bridging command-line arguments and environment variables before delegating to dyad_ctx_init().

Configuration is layered in the following order of precedence:

Environment variables provide the baseline configuration.
Command-line arguments in opt override environment variables by calling setenv() before dyad_ctx_init() is invoked.

Specifically:

If opt->prod_managed_path is set, it is written to DYAD_PATH_PRODUCER_ENV and the directory is created if it does not already exist.
If opt->dtl_mode is set, it is written to DYAD_DTL_MODE_ENV.
If DYAD_KVS_NAMESPACE is not set in the environment, a dummy value is written to allow dyad_ctx_init() to proceed. This is a known limitation (see TODO in source).

After environment setup, calls dyad_ctx_init() with DYAD_COMM_SEND and the provided Flux handle h, then retrieves the initialized context and stores it in mod_ctx->ctx. The Flux handle and debug flag are also applied directly to the context after initialization.

Note

Unlike the C GOTCHA wrapper and C++ stream paths, the Flux module initializes with DYAD_COMM_SEND (producer) rather than DYAD_COMM_RECV (consumer), and adopts the broker-provided Flux handle rather than opening its own.

Parameters:

opt – [in] Parsed command-line options. Must not be NULL. Contains optional overrides for the producer path and DTL mode.
h – [in] Flux handle provided by the broker to mod_main(). Must not be NULL. Stored directly in the context after initialization.

Return values:

DYAD_RC_OK – The context was successfully initialized.
DYAD_RC_NOCTX – opt, h, or the module context is NULL, or dyad_ctx_init() failed to initialize the context or its DTL handle.

Returns:

dyad_rc_t return code indicating the outcome:

int mod_main(void **h, int argc, char **argv)

Entry point for the DYAD Flux module, invoked in a new broker thread when the module is loaded.

Called by the Flux broker when the DYAD module is loaded. The h handle provides direct communication with the broker over shared memory. When mod_main() returns, the thread is terminated and the module is unloaded.

Performs the following steps in order:

Validates the Flux handle h.
Retrieves the module context via get_mod_ctx().
Parses command-line arguments via opt_parse(). If -h was passed, prints help and returns immediately.
Initializes the DYAD context via dyad_module_ctx_init(), which applies command-line overrides to environment variables before calling dyad_ctx_init().
Registers Flux message handlers from htab via flux_msg_handler_addvec().
Runs the Flux reactor loop via flux_reactor_run(), blocking until the module is unloaded.

On any error, jumps to mod_error and returns EXIT_FAILURE. On success or after printing help, jumps to mod_done and returns EXIT_SUCCESS.

Note

If DYAD_PROFILER_DFTRACER is defined, initializes the DFTracer profiler using the broker rank as the process ID before any other setup.

Parameters:

h – [in] Flux handle provided by the broker over shared memory. Must not be NULL.
argc – [in] Number of command-line arguments passed to the module by the broker.
argv – [in] Command-line argument strings derived from the free arguments on the flux module load command line. Parsed by opt_parse() to extract the producer path, DTL mode, and debug flag.

Return values:

EXIT_SUCCESS – The module ran and exited cleanly, or -h was passed and help was displayed.
EXIT_FAILURE – Any step in the initialization or reactor loop failed.

Returns:

int

Variables

const struct dyad_mod_ctx dyad_mod_ctx_default = {NULL, NULL}

static const struct flux_msg_handler_spec htab[] = {{FLUX_MSGTYPE_REQUEST, DYAD_DTL_RPC_NAME, dyad_fetch_request_cb, 0}, FLUX_MSGHANDLER_TABLE_END}

Flux message handler table for the DYAD module.

Registers dyad_fetch_request_cb as the handler for all incoming FLUX_MSGTYPE_REQUEST messages addressed to DYAD_DTL_RPC_NAME. This is the single RPC endpoint exposed by the DYAD module — consumers send file fetch requests to this name on the producer’s broker, and the reactor dispatches them to dyad_fetch_request_cb. DYAD_DTL_RPC_NAME is defined as “dyad.fetch”

Passed to flux_msg_handler_addvec() in mod_main() and terminated by FLUX_MSGHANDLER_TABLE_END as required by the Flux API.

struct dyad_mod_ctx

Context structure for the DYAD Flux module.

Holds the Flux message handler table and the DYAD context for a running module instance. Allocated per broker handle via get_mod_ctx() and freed at module finalization time via freectx().

Public Members

flux_msg_handler_t **handlers: Flux message handler table.

dyad_ctx_t *ctx: DYAD context for this module instance.

struct opt_parse_out

Parsed command-line options for the DYAD Flux module.

Public Members

const char *prod_managed_path: Producer-managed directory path, or NULL.

const char *dtl_mode: DTL mode string, or NULL for default.

bool debug: Whether debug logging is enabled.

bool showed_help: Whether -h was passed and help was shown.