Làm thế nào để tránh reentrancy với bộ xử lý sự kiện không đồng bộ void?

Question

Trong một ứng dụng WPF, tôi có một lớp nhận tin nhắn qua mạng. Bất cứ khi nào một đối tượng của lớp học đã nhận được một tin nhắn đầy đủ, một sự kiện được nêu ra. Trong MainWindow của ứng dụng, tôi có một trình xử lý sự kiện đã đăng ký với sự kiện đó. Trình xử lý sự kiện được đảm bảo được gọi trên luồng GUI của ứng dụng.

Bất cứ khi nào trình xử lý sự kiện được gọi, nội dung của tin nhắn cần được áp dụng cho mô hình. Làm như vậy có thể khá tốn kém (> 200ms trên phần cứng hiện tại). Đó là lý do tại sao áp dụng tin nhắn được tải lên hồ bơi thread với Task.Run.

Bây giờ, các tin nhắn có thể được nhận liên tiếp rất gần, do đó, trình xử lý sự kiện có thể được gọi trong khi thay đổi trước đó vẫn đang được xử lý. Cách đơn giản nhất để đảm bảo rằng thư chỉ được áp dụng một lần vào thời điểm nào? Cho đến nay, tôi đã đi lên với những điều sau đây:

using System; 
using System.Threading.Tasks; 
using System.Windows; 

public partial class MainWindow : Window 
{ 
    private Model model = new Model(); 
    private Task pending = Task.FromResult<bool>(false); 

    // Assume e carries a message received over the network. 
    private void OnMessageReceived(object sender, EventArgs e) 
    { 
     this.pending = ApplyToModel(e); 
    } 

    private async Task ApplyToModel(EventArgs e) 
    { 
     await this.pending; 
     await Task.Run(() => this.model.Apply(e)); // Assume this is an expensive call. 
    } 
} 

Điều này dường như làm việc như mong đợi, tuy nhiên nó cũng xuất hiện này chắc chắn sẽ tạo ra một "rò rỉ bộ nhớ", bởi vì nhiệm vụ để áp dụng một thông báo sẽ luôn luôn đầu tiên đợi nhiệm vụ đã áp dụng tin nhắn trước đó. Nếu có, thì thay đổi sau đây nên tránh sự rò rỉ:

private async Task ApplyToModel(EventArgs e) 
{ 
    if (!this.pending.IsCompleted) 
    { 
     await this.pending; 
    } 

    await Task.Run(() => this.model.Apply(e)); 
} 

Đây có phải là cách hợp lý để tránh sự reentrancy với trình xử lý sự kiện không đồng bộ không?

EDIT: Đã xóa câu hỏi không cần thiết await this.pending; trong OnMessageReceived.

CHỈNH SỬA 2: Thư phải được áp dụng cho mô hình theo cùng thứ tự mà chúng đã nhận được.

Answer 1

Chúng tôi cần cảm ơn Stephen Toub ở đây, vì ông có một số cấu trúc khóa không đồng bộ rất hữu ích được trình bày trong một loạt blog, bao gồm khối async lock.

Đây là mã từ bài viết đó (bao gồm một số mã từ phần trước của loạt bài này):

public class AsyncLock 
{ 
    private readonly AsyncSemaphore m_semaphore; 
    private readonly Task<Releaser> m_releaser; 

    public AsyncLock() 
    { 
     m_semaphore = new AsyncSemaphore(1); 
     m_releaser = Task.FromResult(new Releaser(this)); 
    } 

    public Task<Releaser> LockAsync() 
    { 
     var wait = m_semaphore.WaitAsync(); 
     return wait.IsCompleted ? 
      m_releaser : 
      wait.ContinueWith((_, state) => new Releaser((AsyncLock)state), 
       this, CancellationToken.None, 
       TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); 
    } 

    public struct Releaser : IDisposable 
    { 
     private readonly AsyncLock m_toRelease; 

     internal Releaser(AsyncLock toRelease) { m_toRelease = toRelease; } 

     public void Dispose() 
     { 
      if (m_toRelease != null) 
       m_toRelease.m_semaphore.Release(); 
     } 
    } 
} 

public class AsyncSemaphore 
{ 
    private readonly static Task s_completed = Task.FromResult(true); 
    private readonly Queue<TaskCompletionSource<bool>> m_waiters = new Queue<TaskCompletionSource<bool>>(); 
    private int m_currentCount; 

    public AsyncSemaphore(int initialCount) 
    { 
     if (initialCount < 0) throw new ArgumentOutOfRangeException("initialCount"); 
     m_currentCount = initialCount; 
    } 
    public Task WaitAsync() 
    { 
     lock (m_waiters) 
     { 
      if (m_currentCount > 0) 
      { 
       --m_currentCount; 
       return s_completed; 
      } 
      else 
      { 
       var waiter = new TaskCompletionSource<bool>(); 
       m_waiters.Enqueue(waiter); 
       return waiter.Task; 
      } 
     } 
    } 
    public void Release() 
    { 
     TaskCompletionSource<bool> toRelease = null; 
     lock (m_waiters) 
     { 
      if (m_waiters.Count > 0) 
       toRelease = m_waiters.Dequeue(); 
      else 
       ++m_currentCount; 
     } 
     if (toRelease != null) 
      toRelease.SetResult(true); 
    } 
} 

Bây giờ áp dụng nó vào trường hợp của bạn:

private readonly AsyncLock m_lock = new AsyncLock(); 

private async void OnMessageReceived(object sender, EventArgs e) 
{ 
    using(var releaser = await m_lock.LockAsync()) 
    { 
     await Task.Run(() => this.model.Apply(e)); 
    } 
} 

Answer 2

Với một eventhandler trong đó sử dụng async đang chờ chúng tôi không thể sử dụng khóa bên ngoài Task vì chuỗi cuộc gọi giống nhau cho mọi cuộc gọi sự kiện để khóa luôn cho phép vượt qua.

var object m_LockObject = new Object(); 

private async void OnMessageReceived(object sender, EventArgs e) 
{ 
    // Does not work 
    Monitor.Enter(m_LockObject); 

    await Task.Run(() => this.model.Apply(e)); 

    Monitor.Exit(m_LockObject); 
} 

Nhưng chúng ta có thể khóa bên trong công tác vì Task.Run luôn tạo ra một nhiệm vụ mới mà không được chạy song song trên cùng một sợi

var object m_LockObject = new Object(); 

private async void OnMessageReceived(object sender, EventArgs e) 
{ 
    await Task.Run(() => 
    { 
     // Does work 
     lock(m_LockObject) 
     { 
      this.model.Apply(e); 
     } 
    }); 
} 

Vì vậy, khi một cuộc gọi sự kiện OnMessageReceived nó trả immidiatly và model.Apply chỉ được nhập cái khác.