Параллельный пакетный файл с Amazon S3 с использованием AWS S3 SDK для .NET.

Проблема: Я хотел бы загрузить 100 файлов параллельно с AWS S3, используя их .NET SDK. Загруженный контент должен храниться в 100 потоках памяти (файлы достаточно малы, и я могу взять их оттуда). Я запутался между Task, IAsyncResult, Parallel. * И другими различными подходами в .NET 4.0.

Если я попытаюсь решить проблему самостоятельно, с головы до ног я представляю себе что-то вроде этого псевдокода: (отредактировано для добавления типов к некоторым переменным)

using Amazon;
using Amazon.S3;
using Amazon.S3.Model;

AmazonS3 _s3 = ...;
IEnumerable<GetObjectRequest> requestObjects = ...;


// Prepare to launch requests
var asyncRequests = from rq in requestObjects 
    select _s3.BeginGetObject(rq,null,null);

// Launch requests
var asyncRequestsLaunched = asyncRequests.ToList();

// Prepare to finish requests
var responses = from rq in asyncRequestsLaunched 
    select _s3.EndGetRequest(rq);

// Finish requests
var actualResponses = responses.ToList();

// Fetch data
var data = actualResponses.Select(rp => {
    var ms = new MemoryStream(); 
    rp.ResponseStream.CopyTo(ms); 
    return ms;
});

Этот код запускает 100 запросов параллельно, что хорошо. Однако есть две проблемы:

Последний оператор будет загружать файлы поочередно, а не параллельно. В потоке не существует метода BeginCopyTo()/EndCopyTo()...
Предыдущий оператор не отпустит, пока все запросы не ответят. Другими словами, ни один из файлов не начнет загружаться до тех пор, пока все они не начнутся.

Итак, я начинаю думать, что я иду по неправильному пути...

Справка

Ответ 1

Вероятно, проще, если вы перевернете операцию в метод, который будет обрабатывать один запрос асинхронно, а затем вызовите его 100 раз.

Чтобы начать, укажите нужный конечный результат. Поскольку вы работаете с MemoryStream, это означает, что вы захотите вернуть Task<MemoryStream> из вашего метода. Подпись будет выглядеть примерно так:

static Task<MemoryStream> GetMemoryStreamAsync(AmazonS3 s3, 
    GetObjectRequest request)

Поскольку ваш объект AmazonS3 реализует шаблон асинхронного проектирования, вы можете использовать FromAsync в TaskFactory class для создания Task<T> из класса, реализующего асинхронный дизайн Шаблон, например:

static Task<MemoryStream> GetMemoryStreamAsync(AmazonS3 s3, 
    GetObjectRequest request)
{
    Task<GetObjectResponse> response = 
        Task.Factory.FromAsync<GetObjectRequest,GetObjectResponse>(
            s3.BeginGetObject, s3.EndGetObject, request, null);

    // But what goes here?

Итак, вы уже в хорошем месте, у вас есть Task<T>, на который вы можете дождаться или получить обратный вызов при завершении вызова. Однако вам нужно каким-то образом перевести GetObjectResponse, возвращенный с вызова на Task<GetObjectResponse>, в MemoryStream.

С этой целью вы хотите использовать метод ContinueWith в классе Task<T>. Подумайте об этом как о асинхронной версии метода Select на Enumerable class, это просто проецирование в другое Task<T>, за исключением того, что каждый раз, когда вы вызываете ContinueWith, вы потенциально создаете новую задачу, которая запускает этот раздел кода.

При этом ваш метод выглядит следующим образом:

static Task<MemoryStream> GetMemoryStreamAsync(AmazonS3 s3, 
    GetObjectRequest request)
{
    // Start the task of downloading.
    Task<GetObjectResponse> response = 
        Task.Factory.FromAsync<GetObjectRequest,GetObjectResponse>(
            s3.BeginGetObject, s3.EndGetObject, request, null
        );

    // Translate.
    Task<MemoryStream> translation = response.ContinueWith(t => {
        using (Task<GetObjectResponse> resp = t ){
            var ms = new MemoryStream(); 
            t.Result.ResponseStream.CopyTo(ms); 
            return ms;
        } 
    });

    // Return the full task chain.
    return translation;
}

Обратите внимание, что в приведенном выше примере вы можете вызвать перегрузку ContinueWith, передав TaskContinuationOptions.ExecuteSynchronously, так как кажется, что вы выполняете минимальную работу (я не могу сказать, ответы могут быть огромными). В тех случаях, когда вы выполняете очень минимальную работу, когда было бы вредно запускать новую задачу для завершения работы, вы должны пройти TaskContinuationOptions.ExecuteSynchronously, чтобы не тратить время на создание новых задач для минимальных операций.

Теперь, когда у вас есть метод, который может преобразовать один запрос в Task<MemoryStream>, создание обертки, которая будет обрабатывать любое число из них, проста:

static Task<MemoryStream>[] GetMemoryStreamsAsync(AmazonS3 s3,
    IEnumerable<GetObjectRequest> requests)
{
    // Just call Select on the requests, passing our translation into
    // a Task<MemoryStream>.
    // Also, materialize here, so that the tasks are "hot" when
    // returned.
    return requests.Select(r => GetMemoryStreamAsync(s3, r)).
        ToArray();
}

В приведенном выше примере вы просто берете последовательность своих экземпляров GetObjectRequest и возвращаете массив из Task<MemoryStream>. Важен тот факт, что он возвращает материализованную последовательность. Если вы не материализуете его перед возвратом, то задачи не будут созданы до тех пор, пока последовательность не будет выполнена.

Конечно, если вы хотите этого поведения, то, во что бы то ни стало, просто удалите вызов .ToArray(), верните метод IEnumerable<Task<MemoryStream>>, а затем запросы будут сделаны, когда вы выполните итерацию по задачам.

Оттуда вы можете обрабатывать их по одному (используя метод Task.WaitAny в цикле) или дождаться их всех (путем вызова метода Task.WaitAll). Примером последнего может быть:

static IList<MemoryStream> GetMemoryStreams(AmazonS3 s3, 
    IEnumerable<GetObjectRequest> requests)
{
    Task<MemoryStream>[] tasks = GetMemoryStreamsAsync(s3, requests);
    Task.WaitAll(tasks);
    return tasks.Select(t => t.Result).ToList();
}

Кроме того, следует отметить, что это довольно хорошо подходит для Reactive Extensions framework, так как это очень хорошо подходит для IObservable<T>.