Stream.CopyToAsync с отчетностью о прогрессе - прогресс сообщается даже после завершения копирования

Question

Я создал простые консольные приложения, которые загружают файлы из Интернета.
Потому что I had problems with WebClient Я решил написать свое приложение, используя HttpClient.

В основном я делаю запрос читать заголовки, а затем используя ReadAsStreamAsync Я получаю поток, который я копирую в локальный файл, используя CopyToAsync.

Я нашел метод расширения для потока, который поддерживает IProgress:

public static class StreamExtensions 
{ 
    public static async Task CopyToAsync(this Stream source, Stream destination, IProgress<long> progress, CancellationToken cancellationToken = default(CancellationToken), int bufferSize = 0x1000) 
    { 
     var buffer = new byte[bufferSize]; 
     int bytesRead; 
     long totalRead = 0; 
     while ((bytesRead = await source.ReadAsync(buffer, 0, buffer.Length, cancellationToken)) > 0) 
     { 
      await destination.WriteAsync(buffer, 0, bytesRead, cancellationToken); 
      cancellationToken.ThrowIfCancellationRequested(); 
      totalRead += bytesRead; 
      //Thread.Sleep(10); 
      progress.Report(totalRead); 
     } 
    } 
} 

Мои приложения работает, но я получаю неправильную информацию о ходе.
Например, при загрузке 2 файла я вижу это в окне вывода:

file1.tmp 60.95% 
file2.tmp 98.09% 
file1.tmp 60.98% 
file2.tmp 98.21% 
file2.tmp 98.17% 
file2.tmp 98.25% 
file1.tmp 61.02% 
file2.tmp 98.41% 
file2.tmp downloaded 
file2.tmp 98.29% 
file2.tmp 98.37% 
file1.tmp 61.06% 
file2.tmp 89.27% 
file2.tmp 89.31% 
file2.tmp 98.33% 
file2.tmp 98.45% 
file2.tmp 98.48% 
file1.tmp 61.10% 
file1.tmp 61.14% 
file2.tmp 98.52% 
file1.tmp 61.22% 
file2.tmp 98.60% 
file2.tmp 98.56% 
file1.tmp 61.30% 
file2.tmp 98.88% 
file2.tmp 90.44% 
file1.tmp 61.53% 
file2.tmp 98.72% 
file1.tmp 61.41% 
file1.tmp 61.73% 
file2.tmp 98.80% 
file1.tmp 61.26% 
file1.tmp 61.49% 
file1.tmp 61.57% 
file1.tmp 61.69% 
... 
file1.tmp 99.31% 
file1.tmp 98.84% 
file1.tmp 98.80% 
file1.tmp 99.04% 
file1.tmp 99.43% 
file1.tmp 99.12% 
file1.tmp 99.00% 
file1.tmp downloaded 
file1.tmp 100.00% 
file1.tmp 98.73% 
file1.tmp 98.88% 
file1.tmp 99.47% 
file1.tmp 99.98% 
file1.tmp 99.90% 
file1.tmp 98.96% 
file1.tmp 99.78% 
file1.tmp 99.99% 
file1.tmp 99.74% 
file1.tmp 99.59% 
file1.tmp 99.94% 
file1.tmp 98.49% 
file1.tmp 98.53% 
ALL FILES DOWNLOADED 
file1.tmp 99.55% 
file1.tmp 98.41% 
file1.tmp 99.62% 
file1.tmp 98.34% 
file1.tmp 99.66% 
file1.tmp 98.69% 
file1.tmp 98.37% 

Как Вы можете видеть, что я получил информацию, что file2 загружается, но я все еще получаю доклад от CopyToAsync, то же самое с file1.

из-за того, что я иногда получаю этот странный консольный вывод:

В идеале я хотел бы, чтобы быть уверенным, чем когда я звоню:

await streamToReadFrom.CopyToAsync(streamToWriteTo, progress, source.Token,0x2000); 
Debug.WriteLine(filename+" downloaded"); 

после я получаю эту информацию отладки не сообщается о достигнутом прогрессе (файл загружен). Я думал, что await решит мою проблему, но это не так.

Как это исправить? В качестве временного решения я добавляю Thread.Sleep к CopyToAsync как раз перед сообщением о прогрессе.

Ниже мой текущий код:

using System; 
using System.Collections.Generic; 
using System.Diagnostics; 
using System.IO; 
using System.Linq; 
using System.Net.Http; 
using System.Threading; 
using System.Threading.Tasks; 

namespace AsyncDownloadTest 
{ 
    class Program 
    { 
     private const string LocalPath = @"D:\TEMP"; 

     static void Main() 
     { 
      try 
      { 
       var filesToDownlad = new List<Tuple<string, string>> 
       { 
        new Tuple<string, string>("file1.tmp", "http://ipv4.download.thinkbroadband.com/10MB.zip"), 
        new Tuple<string, string>("file2.tmp", "http://ipv4.download.thinkbroadband.com/10MB.zip") 
       }; 
       _consolePosition = -1; 
       Console.CursorVisible = false; 

       Parallel.ForEach(filesToDownlad, new ParallelOptions { MaxDegreeOfParallelism = 4 }, doc => 
       { 
        DownloadFile(doc.Item2,doc.Item1).Wait(); 
       }); 
       Debug.WriteLine("ALL FILES DOWNLOADED"); 
       Console.CursorVisible = true;  
      } 
      catch (Exception e) 
      { 
       Console.WriteLine(e); 
       Console.ReadLine(); 
      } 
     } 

     private static readonly object ConsoleLock = new object(); 
     private static int _consolePosition; 

     static readonly CancellationTokenSource source = new CancellationTokenSource(); 

     private static async Task DownloadFile(string url, string filename) 
     { 
      int currenctLineNumber = 0; 
      int currectProgress = 0; 

      try 
      { 
       lock (ConsoleLock) 
       { 
        _consolePosition++; 
        currenctLineNumber = _consolePosition; 
       } 

       long fileSize = -1; 

       IProgress<long> progress = new Progress<long>(value => 
       { 
        decimal tmp = (decimal)(value * 100)/fileSize; 

        if (tmp != currectProgress && tmp > currectProgress) 
        { 
         lock (ConsoleLock) 
         { 
          currectProgress = (int)tmp; 
          Console.CursorTop = currenctLineNumber; 
          Console.CursorLeft = 0; 
          Console.Write("{0,10} - {2,11} - {1,6:N2}%", filename, tmp, "DOWNLOADING"); 
         } 
         Debug.WriteLine("{1} {0:N2}%", tmp, filename); 
        } 
       }); 

       using (HttpClient client = new HttpClient()) 
       { 
        using (HttpResponseMessage response = await client.GetAsync(url, HttpCompletionOption.ResponseHeadersRead, source.Token)) 
        { 
         response.EnsureSuccessStatusCode(); 
         if (response.Content.Headers.ContentLength.HasValue) fileSize = response.Content.Headers.ContentLength.Value; 

         if (response.Content.Headers.ContentDisposition != null) 
         { 
          var tmp = response.Content.Headers.ContentDisposition.FileName.Replace("\"", ""); 
          Debug.WriteLine("Real name: {0}",tmp); 
         } 

         using (Stream streamToReadFrom = await response.Content.ReadAsStreamAsync()) 
         { 
          using (Stream streamToWriteTo = File.Open(Path.Combine(LocalPath, filename), FileMode.Create, FileAccess.Write)) 
          { 
           await streamToReadFrom.CopyToAsync(streamToWriteTo, progress, source.Token,0x2000); 

           Debug.WriteLine(filename+" downloaded"); 

           lock (ConsoleLock) 
           { 
            Console.CursorTop = currenctLineNumber; 
            Console.CursorLeft = 0; 
            var oldColor = Console.ForegroundColor; 
            Console.ForegroundColor = ConsoleColor.Green; 
            Console.Write("{0,10} - {2,11} - {1,6:N2}%", filename, 100, "SUCCESS"); 
            Console.ForegroundColor = oldColor; 
           } 
          } 
         } 
        } 
       } 
      } 
      catch (Exception e) 
      { 
       Debug.WriteLine(e.Message); 
       lock (ConsoleLock) 
       { 
        Console.CursorTop = currenctLineNumber; 
        Console.CursorLeft = 0; 
        var oldColor = Console.ForegroundColor; 
        Console.ForegroundColor = ConsoleColor.Red; 
        Console.Write("{0,10} - {2,11} - {1,6:N2}%", filename, currectProgress, "ERROR"); 
        Console.ForegroundColor = oldColor; 
       } 
      } 
     } 
    } 

    public static class StreamExtensions 
    { 
     public static async Task CopyToAsync(this Stream source, Stream destination, IProgress<long> progress, CancellationToken cancellationToken = default(CancellationToken), int bufferSize = 0x1000) 
     { 
      var buffer = new byte[bufferSize]; 
      int bytesRead; 
      long totalRead = 0; 
      while ((bytesRead = await source.ReadAsync(buffer, 0, buffer.Length, cancellationToken)) > 0) 
      { 
       await destination.WriteAsync(buffer, 0, bytesRead, cancellationToken); 
       cancellationToken.ThrowIfCancellationRequested(); 
       totalRead += bytesRead; 
       Thread.Sleep(10); 
       progress.Report(totalRead); 
      } 
     } 
    } 
} 

Answer 1

Ваша проблема на самом деле здесь:

new Progress<long> 

Progress<T> класс always invokes its callbacks in a SynchronizationContext - который в данном случае является пул потоков SynchronizationContext. Это означает, что когда код отчета о проделанных вызовах вызывает Report, он просто ставит очередь на обратный вызов пула потоков. Таким образом, их можно увидеть не по порядку (или по-прежнему немного после завершения загрузки).

Чтобы исправить это, вы можете создать свои собственные пользовательские реализации IProgress<T>:

//C#6.0 
public sealed class SynchronousProgress<T> : IProgress<T> 
{ 
    private readonly Action<T> _callback; 
    public SynchronousProgress(Action<T> callback) { _callback = callback; } 
    void IProgress<T>.Report(T data) => _callback(data); 
} 
//older version 
public sealed class SynchronousProgress<T> : IProgress<T> 
{ 
    private readonly Action<T> _callback; 

    public SynchronousProgress(Action<T> callback) 
    { 
     _callback = callback; 
    } 

    void IProgress<T>.Report(T data) 
    { 
     _callback(data); 
    } 
} 

Затем замените строку

IProgress<long> progress = new Progress<long>(value => 

с

IProgress<long> progress = new SynchronousProgress<long>(value => 

Answer 2

ОП просил меня показать, как для выполнения своей программы с использованием потока данных TPL в комментариях. На самом деле это довольно простое преобразование. Сначала добавьте ссылку на пакет NuGet System.Threading.Tasks.Dataflow.Тогда просто изменить свою основную функцию

static void Main() 
{ 
    try 
    { 
     var filesToDownlad = new List<Tuple<string, string>> 
     { 
      new Tuple<string, string>("file1.tmp", "http://ipv4.download.thinkbroadband.com/10MB.zip"), 
      new Tuple<string, string>("file2.tmp", "http://ipv4.download.thinkbroadband.com/10MB.zip") 
     }; 
     _consolePosition = -1; 
     Console.CursorVisible = false; 

     var downloadBlock = new ActionBlock<Tuple<string, string>>(doc => DownloadFile(doc.Item2, doc.Item1), 
                    new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = 4}); 

     foreach (var file in filesToDownlad) 
     { 
      downloadBlock.Post(file); 
     } 
     downloadBlock.Complete(); 
     downloadBlock.Completion.Wait(); 


     Debug.WriteLine("ALL FILES DOWNLOADED"); 
     Console.CursorVisible = true; 
    } 
    catch (Exception e) 
    { 
     Console.WriteLine(e); 
     Console.ReadLine(); 
    } 
} 

Если вы делаете, где это с программой с контекстом синхронизации и вы хотели бы дождаться завершения и проводки а не делать синхронные операции, которые вы могли бы сделать

static async Task Example() 
{ 
    try 
    { 
     var filesToDownlad = new List<Tuple<string, string>> 
     { 
      new Tuple<string, string>("file1.tmp", "http://ipv4.download.thinkbroadband.com/10MB.zip"), 
      new Tuple<string, string>("file2.tmp", "http://ipv4.download.thinkbroadband.com/10MB.zip") 
     }; 
     _consolePosition = -1; 
     Console.CursorVisible = false; 

     var downloadBlock = new ActionBlock<Tuple<string, string>>(doc => DownloadFile(doc.Item2, doc.Item1), 
                    new ExecutionDataflowBlockOptions {MaxDegreeOfParallelism = 4}); 

     foreach (var file in filesToDownlad) 
     { 
      await downloadBlock.SendAsync(file); 
     } 
     downloadBlock.Complete(); 
     await downloadBlock.Completion; 


     Debug.WriteLine("ALL FILES DOWNLOADED"); 
     Console.CursorVisible = true; 
    } 
    catch (Exception e) 
    { 
     Console.WriteLine(e); 
     Console.ReadLine(); 
    } 
} 

Обратите внимание, что это не устраняет вашу проблему «ВСЕ ФАЙЛЫ, ЗАГРУЖЕННЫЕ», запущенные на ранней стадии. Вам нужно использовать Stephen's solution to fix that. Все это исправляет потенциальный тупик, если этот код работал в ситуации, когда в вызывающем потоке может быть SynchronizationContext.