Updating items in a list from a blocking source asynchronously

Question

This is a Windows forms application.

this.Server.GetLogMessages() will block if there are no messages to get, so I want this in a separate thread. I'm putting the messages into a ConcurrentStack and then pulling them out on the main thread to output them.

Here is the respective code:

private async Task UpdateLogs()
{
 var logMessages = new ConcurrentStack<string>();
 // assign to `_` to avoid warning
 var _ = Task.Run(async () =>
 {
 await foreach (var message in this.Server.GetLogMessages())
 {
 logMessages.Push(message);
 }
 }, this.CancellationTokenSource.Token);
 await foreach (var message in GetMessages())
 {
 // don't update the UI if task is canceled
 if (this.CancellationTokenSource.Token.IsCancellationRequested)
 {
 break;
 }
 this.textLogs.AppendText($"{message}{Environment.NewLine}");
 }
 async IAsyncEnumerable<string> GetMessages()
 {
 var message = String.Empty;
 while (!this.CancellationTokenSource.IsCancellationRequested)
 {
 while (!logMessages.TryPop(out message))
 {
 await Task.Delay(100);
 }
 yield return message;
 }
 }
}

It works fine, but seems as if it could be better/cleaner somehow.

• aepot: this.Server.GetLogMessages() will block if there are no messages to get means that the method is not properly implemented. Can it be fixed?
• Peter Csala: Have you considered to use DataFlow for this problem?
• rhughes: @aepot It may be able to be fixed. That is a different issue though.
• rhughes: @PeterCsala I have not heard of that. What is it?
• rhughes: @aepot It seems so. I haven't tested it yet sorry. I will let you know.

Answer 1

My try to simplify and prettify the method.

Avoid using globals in async code, pass Token as argument.

// optimized out async State Machine
private Task UpdateLogsAsync(IProgress<string> status, CancellationToken token) => Task.Run(async () =>
{
 await foreach (var message in this.Server.GetLogMessages().WithCancellation(token))
 {
 status.Report(message);
 }
}, token);

Usage

// synchronized callback. Create new in UI Thread and its body will be always executed there.
IProgress<string> status = new Progess<string>(message =>
{
 this.textLogs.AppendText($"{message}{Environment.NewLine}");
});
try
{
 await UpdateLogsAsync(status, this.CancellationTokenSource.Token);
}
catch (OperationCanceledException ex)
{
 status.Report(ex.Message);
}

• Peter Csala: You can take advantage of the WithCancellation method.
• aepot: @PeterCsala this.Server.GetLogMessages().WithCancellation(token)? Sorry, I'm not familiar with it.
• Peter Csala: Yes, as long as the GetLogMessages returns with an IAsyncEnumerable<T> then you can use the WithCancellation extension method to pass the ct to the async enumerator. For further details please check the following SO answer.
• aepot: @PeterCsala got it, thank you! Updated the answer.
• Peter Csala: In this case there is no need to explicitly call this:token.ThrowIfCancellationRequested();