C#•15mo ago

✅ How to avoid blocking until all subscribed handlers complete handling an event

I have the following code:

public sealed class MessageBroker
{
  private readonly ConcurrentDictionary<string, byte> m_sentMessages = new();
  private readonly Action<MessageMetadata, string> m_onNewMessage;
  private readonly FileSystemWatcher m_watcher = new();

  public MessageBroker(string exchangePath, Action<MessageMetadata, string> onNewMessage)
  {
    m_onNewMessage = onNewMessage ?? throw new ArgumentNullException(nameof(onNewMessage));

    m_watcher.Path = exchangePath;
    m_watcher.Filter = "[*]_[*]_[*]_[*]_[*]";
    m_watcher.EnableRaisingEvents = true;
    m_watcher.IncludeSubdirectories = false;
    m_watcher.InternalBufferSize = 16_384;
    m_watcher.Created += WatcherOnCreated;
  }

  private static async ValueTask WaitForFileAsync(string filePath)
  {
    // Some Async method
  }

  private async ValueTask ProcessNewMessageAsync(string fullPath)
  {
    if (m_sentMessages.TryRemove(fullPath, out var _))
      return;
    if (!MessageHelper.TryParseToMetadata(fullPath, out var metadata) || metadata is null)
      return;

    await WaitForFileAsync(fullPath).ConfigureAwait(false);

    m_onNewMessage(metadata.Value, fullPath);
  }

  private async void WatcherOnCreated(object sender, FileSystemEventArgs e)
    => await ProcessNewMessageAsync(e.FullPath).ConfigureAwait(false);
}

public sealed class MessageBroker
{
  private readonly ConcurrentDictionary<string, byte> m_sentMessages = new();
  private readonly Action<MessageMetadata, string> m_onNewMessage;
  private readonly FileSystemWatcher m_watcher = new();

  public MessageBroker(string exchangePath, Action<MessageMetadata, string> onNewMessage)
  {
    m_onNewMessage = onNewMessage ?? throw new ArgumentNullException(nameof(onNewMessage));

    m_watcher.Path = exchangePath;
    m_watcher.Filter = "[*]_[*]_[*]_[*]_[*]";
    m_watcher.EnableRaisingEvents = true;
    m_watcher.IncludeSubdirectories = false;
    m_watcher.InternalBufferSize = 16_384;
    m_watcher.Created += WatcherOnCreated;
  }

  private static async ValueTask WaitForFileAsync(string filePath)
  {
    // Some Async method
  }

  private async ValueTask ProcessNewMessageAsync(string fullPath)
  {
    if (m_sentMessages.TryRemove(fullPath, out var _))
      return;
    if (!MessageHelper.TryParseToMetadata(fullPath, out var metadata) || metadata is null)
      return;

    await WaitForFileAsync(fullPath).ConfigureAwait(false);

    m_onNewMessage(metadata.Value, fullPath);
  }

  private async void WatcherOnCreated(object sender, FileSystemEventArgs e)
    => await ProcessNewMessageAsync(e.FullPath).ConfigureAwait(false);
}

The event handler WatcherOnCreated should have very short execution time to avoid overflowing the FileSystemWatcher buffer. I'd like to ask what is the most optimal way of avoiding blocking the FSW when handling the event? I suppose it is running the handler using async void, but from what I understand the execution is blocked until the WaitForFileAsync is awaited, right? Can I await Task.Yield() in the beginning of ProcessNewMessageAsync to avoid blocking sooner?

7 Replies

phaseshift•15mo ago

delegates are not great for that because you can only get, at best, one return value from all subscribers. and async void is... :monkas: don't do it unless forced

DenisOP•15mo ago

why would I want to get a return value from an event handler? and isn't async void perfectly reasonable for event handlers?

phaseshift•15mo ago

You cant await async void iirc

DenisOP•15mo ago

Yes the internals of FileSystemWatcher just Invoke the event whenever a new file is created and it does not care whether the handler is async or not

phaseshift•15mo ago

right, but it's not blocking anything if its async void Or does it still need the first await within it to really go 'true async'?

DenisOP•15mo ago

from my understanding yes, but that is why I'm asking this in the first place

phaseshift•15mo ago

ok then yeah I think that's what task.Yield is for

Gaming

Programming

✅ How to avoid blocking until all subscribed handlers complete handling an event