Асинхронная блокировка на основе ключа

Question

Я пытаюсь выяснить проблему, возникшую в библиотеке ImageProcessor here, где я получаю прерывистые ошибки доступа к файлу при добавлении элементов в кеш.

System.IO.IOException: Процесс не может получить доступ к файлу 'D: \ главная \ сайт \ Wwwroot \ App_Data \ кэш \ 0 \ 6 \ 5 \ е \ 2 \ 7 \ 065f27fc2c8e843443d210a1e84d1ea28bbab6c4.webp', потому что используется другим процессом.

Я написал класс, предназначенный для выполнения асинхронной блокировки на основе ключа, созданного хэшированным URL-адресом, но, похоже, я пропустил что-то в реализации.

Мой замок класс

public sealed class AsyncDuplicateLock 
{ 
    /// <summary> 
    /// The collection of semaphore slims. 
    /// </summary> 
    private static readonly ConcurrentDictionary<object, SemaphoreSlim> SemaphoreSlims 
          = new ConcurrentDictionary<object, SemaphoreSlim>(); 

    /// <summary> 
    /// Locks against the given key. 
    /// </summary> 
    /// <param name="key"> 
    /// The key that identifies the current object. 
    /// </param> 
    /// <returns> 
    /// The disposable <see cref="Task"/>. 
    /// </returns> 
    public IDisposable Lock(object key) 
    { 
     DisposableScope releaser = new DisposableScope(
     key, 
     s => 
     { 
      SemaphoreSlim locker; 
      if (SemaphoreSlims.TryRemove(s, out locker)) 
      { 
       locker.Release(); 
       locker.Dispose(); 
      } 
     }); 

     SemaphoreSlim semaphore = SemaphoreSlims.GetOrAdd(key, new SemaphoreSlim(1, 1)); 
     semaphore.Wait(); 
     return releaser; 
    } 

    /// <summary> 
    /// Asynchronously locks against the given key. 
    /// </summary> 
    /// <param name="key"> 
    /// The key that identifies the current object. 
    /// </param> 
    /// <returns> 
    /// The disposable <see cref="Task"/>. 
    /// </returns> 
    public Task<IDisposable> LockAsync(object key) 
    { 
     DisposableScope releaser = new DisposableScope(
     key, 
     s => 
     { 
      SemaphoreSlim locker; 
      if (SemaphoreSlims.TryRemove(s, out locker)) 
      { 
       locker.Release(); 
       locker.Dispose(); 
      } 
     }); 

     Task<IDisposable> releaserTask = Task.FromResult(releaser as IDisposable); 
     SemaphoreSlim semaphore = SemaphoreSlims.GetOrAdd(key, new SemaphoreSlim(1, 1)); 

     Task waitTask = semaphore.WaitAsync(); 

     return waitTask.IsCompleted 
        ? releaserTask 
        : waitTask.ContinueWith(
         (_, r) => (IDisposable)r, 
         releaser, 
         CancellationToken.None, 
         TaskContinuationOptions.ExecuteSynchronously, 
         TaskScheduler.Default); 
    } 

    /// <summary> 
    /// The disposable scope. 
    /// </summary> 
    private sealed class DisposableScope : IDisposable 
    { 
     /// <summary> 
     /// The key 
     /// </summary> 
     private readonly object key; 

     /// <summary> 
     /// The close scope action. 
     /// </summary> 
     private readonly Action<object> closeScopeAction; 

     /// <summary> 
     /// Initializes a new instance of the <see cref="DisposableScope"/> class. 
     /// </summary> 
     /// <param name="key"> 
     /// The key. 
     /// </param> 
     /// <param name="closeScopeAction"> 
     /// The close scope action. 
     /// </param> 
     public DisposableScope(object key, Action<object> closeScopeAction) 
     { 
      this.key = key; 
      this.closeScopeAction = closeScopeAction; 
     } 

     /// <summary> 
     /// Disposes the scope. 
     /// </summary> 
     public void Dispose() 
     { 
      this.closeScopeAction(this.key); 
     } 
    } 
} 

Использование - в HttpModule

private readonly AsyncDuplicateLock locker = new AsyncDuplicateLock(); 

using (await this.locker.LockAsync(cachedPath)) 
{ 
    // Process and save a cached image. 
} 

Может кто-нибудь месте, где я пошло не так? Я волнуюсь, что я недопонимаю что-то фундаментальное.

Полный источник для библиотеки хранятся на Github here

Answer 1

Как other answerer noted, исходный код удаления SemaphoreSlim из ConcurrentDictionary, прежде чем он освобождает семафор. Таким образом, у вас слишком много семафоров - они удаляются из словаря, когда они все еще могут быть использованы (не получены, но уже получены из словаря).

Проблема с этим видом «блокировки отображения» заключается в том, что трудно узнать, когда семафор больше не нужен. Один из вариантов - никогда не использовать семафоры вообще; это простое решение, но может быть неприемлемым в вашем сценарии.Другая опция - если семафоры фактически связаны с объектами объекта , а не значениями (например, строки) - это их привязка с использованием эфемеронов; однако, я считаю, что этот вариант также не будет приемлемым в вашем сценарии.

Итак, мы делаем это с трудом. :)

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

public sealed class AsyncDuplicateLock 
{ 
    private sealed class RefCounted<T> 
    { 
    public RefCounted(T value) 
    { 
     RefCount = 1; 
     Value = value; 
    } 

    public int RefCount { get; set; } 
    public T Value { get; private set; } 
    } 

    private static readonly Dictionary<object, RefCounted<SemaphoreSlim>> SemaphoreSlims 
         = new Dictionary<object, RefCounted<SemaphoreSlim>>(); 

    private SemaphoreSlim GetOrCreate(object key) 
    { 
    RefCounted<SemaphoreSlim> item; 
    lock (SemaphoreSlims) 
    { 
     if (SemaphoreSlims.TryGetValue(key, out item)) 
     { 
     ++item.RefCount; 
     } 
     else 
     { 
     item = new RefCounted<SemaphoreSlim>(new SemaphoreSlim(1, 1)); 
     SemaphoreSlims[key] = item; 
     } 
    } 
    return item.Value; 
    } 

    public IDisposable Lock(object key) 
    { 
    GetOrCreate(key).Wait(); 
    return new Releaser { Key = key }; 
    } 

    public async Task<IDisposable> LockAsync(object key) 
    { 
    await GetOrCreate(key).WaitAsync().ConfigureAwait(false); 
    return new Releaser { Key = key }; 
    } 

    private sealed class Releaser : IDisposable 
    { 
    public object Key { get; set; } 

    public void Dispose() 
    { 
     RefCounted<SemaphoreSlim> item; 
     lock (SemaphoreSlims) 
     { 
     item = SemaphoreSlims[Key]; 
     --item.RefCount; 
     if (item.RefCount == 0) 
      SemaphoreSlims.Remove(Key); 
     } 
     item.Value.Release(); 
    } 
    } 
} 

Answer 2

Для данного ключа,

1. резьбы 1 вызовов GetOrAdd и добавляет новый семафор и получает его через Wait
2. Темы 2 звонков GetOrAdd и получает существующие семафоры и блоки на Wait
3. Тема 1 выпускает семафор, только после звонка TryRemove, который удалил семафор из словаря
4. Тема 2 теперь приобретает семафор.
5. Тема 3 вызывает GetOrAdd для того же ключа, что и нить 1 и 2. Резьба 2 все еще держит семафор, но семафор не находится в словаре, поэтому поток 3 создает новый семафор, и оба потока 2 и 3 получают доступ к тому же защищенный ресурс.

Вам необходимо настроить свою логику. Семафор следует удалять только из словаря, если у него нет официантов.

Вот один из возможных решений, минус асинхронной часть:

public sealed class AsyncDuplicateLock 
{ 
    private class LockInfo 
    { 
     private SemaphoreSlim sem; 
     private int waiterCount; 

     public LockInfo() 
     { 
      sem = null; 
      waiterCount = 1; 
     } 

     // Lazily create the semaphore 
     private SemaphoreSlim Semaphore 
     { 
      get 
      { 
       var s = sem; 
       if (s == null) 
       { 
        s = new SemaphoreSlim(0, 1); 
        var original = Interlocked.CompareExchange(ref sem, null, s); 
        // If someone else already created a semaphore, return that one 
        if (original != null) 
         return original; 
       } 
       return s; 
      } 
     } 

     // Returns true if successful 
     public bool Enter() 
     { 
      if (Interlocked.Increment(ref waiterCount) > 1) 
      { 
       Semaphore.Wait(); 
       return true; 
      } 
      return false; 
     } 

     // Returns true if this lock info is now ready for removal 
     public bool Exit() 
     { 
      if (Interlocked.Decrement(ref waiterCount) <= 0) 
       return true; 

      // There was another waiter 
      Semaphore.Release(); 
      return false; 
     } 
    } 

    private static readonly ConcurrentDictionary<object, LockInfo> activeLocks = new ConcurrentDictionary<object, LockInfo>(); 

    public static IDisposable Lock(object key) 
    { 
     // Get the current info or create a new one 
     var info = activeLocks.AddOrUpdate(key, 
      (k) => new LockInfo(), 
      (k, v) => v.Enter() ? v : new LockInfo()); 

     DisposableScope releaser = new DisposableScope(() => 
     { 
      if (info.Exit()) 
      { 
       // Only remove this exact info, in case another thread has 
       // already put its own info into the dictionary 
       ((ICollection<KeyValuePair<object, LockInfo>>)activeLocks) 
        .Remove(new KeyValuePair<object, LockInfo>(key, info)); 
      } 
     }); 

     return releaser; 
    } 

    private sealed class DisposableScope : IDisposable 
    { 
     private readonly Action closeScopeAction; 

     public DisposableScope(Action closeScopeAction) 
     { 
      this.closeScopeAction = closeScopeAction; 
     } 

     public void Dispose() 
     { 
      this.closeScopeAction(); 
     } 
    } 
}