Data transfers

This page describes the main modules used for data transfers, the Data Control and the Data API. It also outlines the alternative stage-out mechanism.

Data Control: Control interface to the Data API.

This module manages all data transfer operations for pilot jobs, including stage-in of input files and stage-out of output files and logs. It operates as a control layer, orchestrating data transfers through a system of "copytools."

The core of this module is a set of threads that continuously monitor queues for jobs that require data transfers:

• copytool_in: This thread handles the stage-in of input files for jobs. It retrieves jobs from the data_in queue, determines the appropriate copytool to use, and initiates the transfer.
• copytool_out: This thread manages the stage-out of output files. Once a job has finished execution, it is placed in the data_out queue, and this thread handles the upload of output files to the appropriate storage elements.
• queue_monitoring: This thread monitors the status of the data transfer queues and handles jobs that have either completed or failed the transfer process.

The actual data transfer logic is abstracted away into "copytools," which are individual modules located in the pilot/copytool/ directory. Each copytool is responsible for a specific transfer protocol or technology, such as Rucio, xrdcp, etc. This module interacts with the copytools through the StageInClient and StageOutClient classes from the pilot.api.data module, which provide a unified interface for initiating transfers regardless of the underlying copytool used.

The selection of which copytool to use is determined by the site configuration and the protocols supported by the storage elements involved in the transfer. This allows for a flexible and extensible data transfer framework that can be adapted to different computing environments.

Data API: API for data transfers.

This module provides a high-level API for managing data transfers (stage-in and stage-out) within the Pilot framework. It serves as an abstraction layer over various underlying transfer protocols and tools, collectively known as "copytools." The primary goal is to provide a unified interface for staging data, regardless of the specific technology used for the transfer.

Core Classes:

• StagingClient: This is the base class that provides the common framework for data staging. It handles the dynamic selection of copytools based on site configuration and the type of activity (e.g., 'read_lan', 'write_wan'). It includes methods for resolving file replicas from catalogs like Rucio, sorting them based on priority (e.g., LAN vs. WAN), and orchestrating the transfer process through the transfer method. It also manages tracing and logging for transfers.

• StageInClient: This class inherits from StagingClient and specializes in handling the stage-in of input files. It contains logic to resolve the best replica for input files, considering factors like direct access modes (LAN/WAN), allowed schemas (e.g., 'root', 'https'), and site-specific storage configurations. It is also responsible for checking available disk space and verifying that input file sizes are within configured limits.

• StageOutClient: This class, also inheriting from StagingClient, is responsible for staging out output files. Its key functionality includes preparing destinations by resolving the correct output storage element (RSE) based on the activity. It constructs the final destination SURL (Storage URL) for the output files, calculates checksums for verification, and ensures that output files exist and have a non-zero size before initiating the transfer.

Key Concepts:

• Copytools: The actual file transfers are delegated to specific "copytool" modules, which are located in the pilot/copytool/ directory (e.g., rucio, xrdcp, gfal). The StagingClient dynamically imports and uses the appropriate copytool based on the acopytools configuration for a given activity. This design makes the system extensible to new transfer protocols.

• Replica Resolution: For stage-in, the client queries a catalog (like Rucio) to find all available replicas (copies) of a file. These replicas are then sorted by priority (e.g., network proximity, site preference) to select the most efficient source for the transfer.

• Protocol and Destination Resolution: For stage-out, the client determines the correct destination storage and the protocol to use for the transfer based on site and experiment configurations stored in ddmconf and astorages.

• Direct Access: The StageInClient supports a "direct access" or "remote I/O" mode, where files are not physically copied to the worker node but are accessed remotely by the payload. The client identifies when this mode is applicable and sets the file status accordingly, providing the payload with the correct remote TURL (Transport URL).

Workflow:

1. A StageInClient or StageOutClient is instantiated with site-specific information.
2. The transfer method is called with a list of FileSpec objects that represent the files to be transferred.
3. The client determines the appropriate copytool(s) for the given activity.
4. For each copytool, the client prepares the files:
   • Stage-in: Resolves replicas, selects the best source URL, and checks for direct access possibilities.
   • Stage-out: Resolves the destination URL and prepares the source file by checking for its existence and calculating its checksum.
5. The client calls the copy_in or copy_out function of the selected copytool module, passing the list of files to be transferred.
6. The copytool executes the transfer.
7. The client updates the status of the FileSpec objects and handles any errors. If one copytool fails, it can try the next one in the configured list.

Alternative Stage-Out

In case the StageOutClient fails to stage-out a file, the pilot has the ability to retry at a secondary RSE. The mechanism is implemented in the pilot/control/data.py module and function _do_stageout().

This function orchestrates the stage-out process, including a fallback mechanism known as "alternative stage-out." This allows the pilot to attempt transferring output files to a secondary storage element (RSE) if the primary one fails.

The alternative stage-out logic is as follows:

1. Initial Transfer Attempt: The function first calls client.transfer() to attempt uploading all files to their primary destination RSEs. Crucially, if alternative stage-out is enabled for the job (job.allow_altstageout() is True), the transfer method is called with raise_exception=False. This ensures that if a transfer fails, the function does not immediately exit but continues execution, allowing for a retry.

2. Failure Detection: After the initial attempt, the code checks for any files that were not successfully transferred by creating a remain_files list.

3. Conditions for Retry: An alternative stage-out is attempted only if all of the following conditions are met: a. The job is configured to allow alternative stage-out (altstageout is True). b. There are files remaining that failed the first transfer attempt (remain_files is not empty). c. Every file that failed has an alternative destination defined (has_altstorage is True). The alternative destination is stored in the ddmendpoint_alt attribute of the file spec.

4. Executing the Alternative Transfer: If the conditions are met, the function iterates through the list of failed files. For each file, it swaps the primary destination (entry.ddmendpoint) with the alternative one (entry.ddmendpoint_alt).

5. Second Transfer Attempt: The client.transfer() method is called a second time. The StageOutClient will now attempt to transfer the previously failed files to their newly assigned alternative destinations. Files that were successfully transferred in the first attempt are ignored.