SNOW-300951

Introduction

Back when we started on this project it was mainly to decrease the number of our dependencies and the overall size of snowflake-connector-python, as we can disregard all other capabilities other than blob storage.

Since the start we ran into a few other issues with the cloud SDKs, like some more exotic proxies not being supported by the cloud SDKs and not supporting asyncio.

Design

Initial design

Different blob storage services work mostly the same, but there are subtle differences. For example consider that on S3 we need to keep track of the block numbers and the returned etags for all the pieces for multipart upload, while on Azure we specify the certain block IDs when putting the pieces and then we send the list of those ids to an end-point, because of this we quickly arrived to a class design like this:

While performing PUT and GET operation most of the time is spent on the network calls, so we knew from the start that we needed to perform the network call in parallel.
Previously our method to parallelize execution was to use ThreadPoolExecutors (I’ll be referring to them as tpes from now on), which already handles some concurrency hurdles for us, but during this project their limitations became obvious.

Notice that the idea was that the main thread would be doing most of the work and then it would let the network_tpe take care of network calls.

Once the implementation was semi complete we ran benchmark and found that performance suffered greatly at least on machines with limited computing power when compared to their network speed (such as cheaper EC2 machines).

As it turned out when only the network calls were parallelized compression became the most expensive operation. So after some back-and-forth we arrived to the final design.

Final design

The new idea in this iteration is that we have separate thread pools for CPU and network bound tasks.

When designing the flow of files we wanted to make sure that if a chunk needed to be retried the queue to the network_tpe wasn’t too long so we introduced a conditional variable where each file chunk will wait to submit it’s chunk to the network tpe until the queue’s length is over 2 * max_concurrency. Once all of the chunks of a file have been processed by the network_tpe during a download gets submitted to the postprocess_tpe to be decrypted. In case of an upload, the last chunk to finish in the network_tpe makes a second call to finalize the file on blob storage. Note that the only difference between preprocess_tpe and postprocess_tpe is that we did not want to pollute the queue to the network_tpe with tasks that we can basically guaranteed to work.

Issues with ThreadPoolExecutors

1. Callbacks might be executed synchronously: This is a gotcha of the tpes, if a thread submits a job and if that job is done by the time the thread is attaching a callback onto the future, then the callback will be executed by the thread that is attaching the callback to the future. For this reason we introduced the fn function_and_callback_wrapper that makes sure that the callback is called concurrently from the tpe.

2. Error handling in callbacks is non-existent in tpes: We need to be careful how we handle errors inside of tpes. While calling .result() on a tpe raises the same exception that was raised in the callable that was executed there’s no such logic for Exceptions raised inside of callback function. For this reason we introduced manually track errors and handle them inside of the SnowflakeFileTransferAgent.result.
   Closes SNOW-141545: Request - Reduce package dependencies chaos #284