Приложение WPF, выполняющее параллельные операции

проблема в том, что я знаю, что Task.Run может вызвать задачу, которая никогда не будет завершена, и я беспокоюсь, что это может вызвать ошибки.

Очевидная проблема с этим кодом заключается в отсутствии обработки исключений. С этим довольно легко справиться, поместив все внутри UdpMessageReceivedHandler в try-catch и обработав любые исключения. Как с ними справиться, зависит от вас, опубликуйте сообщение в ветке пользовательского интерфейса? Зафиксировать неудачу? Просто позволить приложению вылететь?

Вы также можете просто вызвать обработчик сообщений непосредственно из потока приема, например:

string receivedString = Encoding.UTF8.GetString(result.Buffer);
await UdpMessageReceivedHandler(receivedString);

...
private async Task UdpMessageReceivedHandler(string receivedString){
...

Но учтите, что это приведет к сериализации всей работы, поэтому, если это займет больше времени, чем отправка сообщений о скорости, вы столкнетесь с проблемами. В этой модели любые исключения будут передаваться тому, что вызывается StartListeningAsync, поэтому оно может их обработать.

Другой вариант — поместить все полученные сообщения в очередь (например, BlockingCollection) и использовать отдельную задачу для обработки сообщений из очереди. Это должно позволить вам проверить, завершается ли задача процесса очереди преждевременно с ошибкой, и обработать ее. Поток данных — это один из способов настроить конвейер очередей для обработки, но есть много других вариантов. Преимущество DataFlow в том, что он очень гибок, поэтому выполнять параллельную обработку сообщений довольно просто.

Также подумайте, действительно ли UDP является подходящим протоколом для вашего приложения. Встроенной надежности нет, поэтому следует ожидать потери пакетов. Вместо этого я бы рассмотрел какую-нибудь библиотеку, которая может передавать сообщения, например http, gRPC, MQTT или любой из многих других протоколов.

Судя по тому, что вы показали, я бы использовал своего рода модель производитель-потребитель. На стороне потребителя вы можете решить, хотите ли вы обрабатывать очередь асинхронно или параллельно. Асинхронная обработка очереди позволит вашему пользовательскому интерфейсу оставаться отзывчивым, а параллельная обработка в параллельном потоке позволяет ускорить обработку (для очистки очереди используется больше ядер ЦП). Обратите внимание: не все задачи выигрывают от параллелизма. Параллелизм может даже усугубить проблему с точки зрения производительности и целостности.

Поскольку вы недостаточно описали исходную проблему, неясно, возможен ли параллелизм. Учитывая жестко закодированные пути к файлам в контексте операции копирования файлов, похоже, что это не вариант. В этом случае вы можете легко преобразовать приведенный ниже пример в чисто асинхронное решение, удалив Parallel.ForEachAsync из метода ProcessQueueAsync.

То, как вы читаете сетевой поток, также выглядит странно. В общем, нельзя полагаться на то, что каждый полученный пакет содержит разумный контент. Обычно вам приходится объединять несколько пакетов. Это означает, что вам нужна буферизация.
Вам также понадобится протокол (например, JSON-RPC), который позволит вам безопасно идентифицировать символ, описывающий действие. Вы не можете случайным образом выбирать символы из пакета или серии пакетов. В общем. На тот случай, если у вас этого не было на радаре.

Простое решение с использованием Channel<T> и Parallel.ForEachAsync может выглядеть следующим образом:

Параллельная версия

public class ChannelExample
{
  private Channel<byte[]>? channel;

  public async Task StartListeningAsync(CancellationToken cancellationToken = default)
  {
    Debug.WriteLine("UDP LISTENER IS ACTIVE");

    // Use using statement or expression over try-finally
    using var udpClient = new UdpClient();

    InitializeChannel();

    // Process queue in the background
    var backgroundTask = Task.Run(() => ProcessQueueAsync(cancellationToken), cancellationToken);

    try
    {
      while (true)
      {
        UdpReceiveResult result = await udpClient.ReceiveAsync(cancellationToken).ConfigureAwait(false);
        await EnqueueItemAsync(result.Buffer, cancellationToken).ConfigureAwait(false);
      }

      // Simply await the single background task to unwrap 
      // its exceptions and to catch them as usual...

      // ...use this line to wait for multiple tasks
      //await Task.WhenAll(backgroundTask).ConfigureAwait(false);

      // ...or simply await the single background task to unwrap 
      // its exceptions and to catch them as usual
      await backgroundTask.ConfigureAwait(false);
    }
    catch (/* TODO::Catch exceptions thrown by the backgroundTask Task if necessary */)
    {
    }
    catch (OperationCanceledException)
    {
      // TODO::Handle cancellation then recover or rethrow.
      throw;
    }
    catch (SocketException ex)
    {
      Debug.WriteLine($"Error in UDP listener: {ex.Message}");

      // We must rethrow the unhandled exception
      throw;
    }
    finally
    {
      FinalizeChannel();
    }
  }

  private async Task EnqueueItemAsync(byte[] data, CancellationToken cancellationToken)
  {
    ChannelWriter<byte[]> channelWriter = this.channel!.Writer;
    await channelWriter.WriteAsync(data, cancellationToken).ConfigureAwait(false);
  }

  private async Task ProcessQueueAsync(CancellationToken cancellationToken)
  {
    ChannelReader<byte[]> channelReader = this.channel!.Reader;
    var parallelOptions = new ParallelOptions()
    {
      CancellationToken = cancellationToken,
      MaxDegreeOfParallelism = Environment.ProcessorCount, // Adjust to optimize parallelism        
    };
    IAsyncEnumerable<byte[]> queue = channelReader.ReadAllAsync(cancellationToken);

    // Process queue in parallel
    await Parallel.ForEachAsync(queue, parallelOptions, ProcessItemAsync).ConfigureAwait(false);
  }

  private async ValueTask ProcessItemAsync(byte[] buffer, CancellationToken cancellationToken)
  {
    string bufferTextContent = Encoding.UTF8.GetString(buffer);
    switch (bufferTextContent[0])
    {
      // Trigger Action A
      case 'A':
        {
          Debug.WriteLine("Action A triggered!");
          // Implement your action logic here
          break;
        }
      // Trigger Action B
      case 'B':
        {
          Debug.WriteLine("Action B triggered!");
          // Implement your action logic here}
          break;
        }
      default:
        throw new NotSupportedException();
    }
  }

  private void InitializeChannel()
  {
    var channelOptions = new UnboundedChannelOptions()
    {
      AllowSynchronousContinuations = false,
      SingleReader = true,
      SingleWriter = true
    };

    this.channel = Channel.CreateUnbounded<byte[]>(channelOptions);
  }

  private void FinalizeChannel()
    => this.channel!.Writer.Complete();
}

Асинхронная версия

public class ChannelExample
{
  private Channel<byte[]>? channel;

  public async Task StartListeningAsync(CancellationToken cancellationToken = default)
  {
    Debug.WriteLine("UDP LISTENER IS ACTIVE");

    // Use using statement or expression over try-finally
    using var udpClient = new UdpClient();

    InitializeChannel();

    // Process queue in the background
    var backgroundTask = Task.Run(() => ProcessQueueAsync(cancellationToken), cancellationToken);

    try
    {
      while (true)
      {
        UdpReceiveResult result = await udpClient.ReceiveAsync(cancellationToken).ConfigureAwait(false);
        await EnqueueItemAsync(result.Buffer, cancellationToken).ConfigureAwait(false);
      }

      // Simply await the single background task to unwrap 
      // its exceptions and to catch them as usual...

      // ...use this line to wait for multiple tasks
      //await Task.WhenAll(backgroundTask).ConfigureAwait(false);

      // ...or simply await the single background task to unwrap 
      // its exceptions and to catch them as usual
      await backgroundTask.ConfigureAwait(false);
    }
    catch (/* TODO::Catch exceptions thrown by the backgroundTask Task if necessary */)
    {
    }
    catch (OperationCanceledException)
    {
      // TODO::Handle cancellation then recover or rethrow.
      throw;
    }
    catch (SocketException ex)
    {
      Debug.WriteLine($"Error in UDP listener: {ex.Message}");

      // We must rethrow the unhandled exception
      throw;
    }
    finally
    {
      FinalizeChannel();
    }
  }

  private async Task EnqueueItemAsync(byte[] data, CancellationToken cancellationToken)
  {
    ChannelWriter<byte[]> channelWriter = this.channel!.Writer;
    await channelWriter.WriteAsync(data, cancellationToken).ConfigureAwait(false);
  }

  private async Task ProcessQueueAsync(CancellationToken cancellationToken)
  {
    ChannelReader<byte[]> channelReader = this.channel!.Reader;
    
    // Process queue asynchronously
    while (await channelReader.WaitToReadAsync(cancellationToken).ConfigureAwait(false))
    {
      while (channelReader.TryRead(out byte[] buffer))
      {
        await ProcessItemAsync(buffer, cancellationToken);
      } 
    }
  }

  private async Task ProcessItemAsync(byte[] buffer, CancellationToken cancellationToken)
  {
    string bufferTextContent = Encoding.UTF8.GetString(buffer);
    switch (bufferTextContent[0])
    {
      // Trigger Action A
      case 'A':
        {
          Debug.WriteLine("Action A triggered!");
          // Implement your action logic here
          break;
        }
      // Trigger Action B
      case 'B':
        {
          Debug.WriteLine("Action B triggered!");
          // Implement your action logic here}
          break;
        }
      default:
        throw new NotSupportedException();
    }
  }

  private void InitializeChannel()
  {
    var channelOptions = new UnboundedChannelOptions()
    {
      AllowSynchronousContinuations = false,
      SingleReader = true,
      SingleWriter = true
    };

    this.channel = Channel.CreateUnbounded<byte[]>(channelOptions);
  }

  private void FinalizeChannel()
    => this.channel!.Writer.Complete();
}