Skip to main content

Handling Asynchronous Cache Operations

Flyte uses an asynchronous catalog client to perform cache lookups and uploads without blocking the main execution loop of a task. Since catalog operations often involve network calls to external services or data stores, executing them synchronously would significantly degrade the performance of the Flyte propeller or plugin executors.

The AsyncClient Interface​

The AsyncClient interface, defined in flyteplugins/go/tasks/pluginmachinery/catalog/async_client.go, provides two primary methods for interacting with the catalog:

type AsyncClient interface {
	// Returns if an entry exists for the given task and input.
	Download(ctx context.Context, requests ...DownloadRequest) (outputFuture DownloadFuture, err error)

	// Adds a new entry to catalog for the given task execution context and the generated output
	Upload(ctx context.Context, requests ...UploadRequest) (putFuture UploadFuture, err error)
}

When you call Download or Upload, the client immediately returns a Future (either DownloadFuture or UploadFuture). These methods do not wait for the operation to complete; instead, they queue the requests in internal workqueues.

Working with Futures​

A Future represents the eventual result of an asynchronous operation. You can check the status of the operation using GetResponseStatus():

type Future interface {
	// Gets the response status for the future. 
	// If the future represents multiple operations, the status will only be 
	// ready if all of them are.
	GetResponseStatus() ResponseStatus

	// Sets a callback handler to be called when the future status changes to ready.
	OnReady(handler ReadyHandler)

	GetResponseError() error
}

Handling Downloads​

When performing a cache lookup, you use DownloadFuture. Once the status is ResponseStatusReady, you can retrieve the DownloadResponse:

outputFuture, err := catalogClient.Download(ctx, downloadRequest)
if err != nil {
    return err
}

if outputFuture.GetResponseStatus() == catalog.ResponseStatusReady {
    resp, err := outputFuture.GetResponse()
    if err != nil {
        return err
    }
    
    if resp.GetCachedCount() > 0 {
        // Handle cache hit
        cachedResults := resp.GetCachedResults()
        // ...
    }
}

The DownloadResponse interface (implemented by downloadFuture in flyteplugins/go/tasks/pluginmachinery/catalog/response.go) provides a bitarray.BitSet to identify which specific requests in a batch resulted in a cache hit.

Internal Architecture​

The AsyncClientImpl (found in flyteplugins/go/tasks/pluginmachinery/catalog/async_client_impl.go) manages two separate workqueue.IndexedWorkQueue instances: a Reader for downloads and a Writer for uploads.

Request Coalescing​

To prevent redundant work, Flyte coalesces identical requests using unique IDs generated from the request data:

  • Downloads: The ID is a consistent hash of the target output prefix path.
  • Uploads: The ID is a hash of the task inputs (via hashInputs).

If multiple plugins request the same cache lookup simultaneously, the IndexedWorkQueue ensures that only one actual network call is made, and all callers receive the same result.

Background Processing​

The actual work is performed by ReaderProcessor and WriterProcessor. These processors wrap a synchronous catalog.Client and execute the Get or Put operations.

For the AsyncClient to begin processing these queues, you must explicitly call Start(ctx):

func (c AsyncClientImpl) Start(ctx context.Context) error {
	if err := c.Reader.Start(ctx); err != nil {
		return errors.Wrapf(ErrSystemError, err, "Failed to start reader queue.")
	}

	if err := c.Writer.Start(ctx); err != nil {
		return errors.Wrapf(ErrSystemError, err, "Failed to start writer queue.")
	}

	return nil
}

Configuration​

You can tune the performance of the asynchronous client via the catalogCache configuration section in flyteplugins/go/tasks/pluginmachinery/catalog/config.go. This allows you to control the number of parallel workers and retry logic for both reading and writing.

type Config struct {
	ReaderWorkqueueConfig workqueue.Config `json:"reader"`
	WriterWorkqueueConfig workqueue.Config `json:"writer"`
	// ...
}

Default settings in Flyte include:

  • Workers: 10 (for both reader and writer)
  • MaxRetries: 3
  • IndexCacheMaxItems: 10,000

These settings ensure that even during high-concurrency scenarios, the catalog service is not overwhelmed while maintaining a high throughput for cache operations.