Abilitazione coda <T> con concomitanza

Question

Ho un precedente question che ho fornito la mia soluzione; tuttavia, non ho accesso a ConcurrentQueue<T> da quando sono su .Net 3.5. Ho bisogno di Queue<T> per consentire la concorrenza. Ho letto questo question e sembra presentare un problema se un articolo è non nella coda e il metodo filettato tenta di desquare un elemento.

Il mio compito ora è determinare se posso derivare la mia classe di coda concorrente. Questo è ciò che mi si avvicinò con:

public sealed class ConcurrentQueue : Queue<DataTable> 
{ 
    public event EventHandler<TableQueuedEventArgs> TableQueued; 
    private ICollection que; 

    new public void Enqueue(DataTable Table) 
    { 
     lock (que.SyncRoot) 
     { 
      base.Enqueue(Table); 
     } 

     OnTableQueued(new TableQueuedEventArgs(Dequeue())); 
    } 

    // this is where I think I will have a problem... 
    new public DataTable Dequeue() 
    { 
     DataTable table; 

     lock (que.SyncRoot) 
     { 
      table = base.Dequeue(); 
     } 

     return table; 
    } 

    public void OnTableQueued(TableQueuedEventArgs table) 
    { 
     EventHandler<TableQueuedEventArgs> handler = TableQueued; 

     if (handler != null) 
     { 
      handler(this, table); 
     } 
    } 
} 

Così, quando un DataTable è in coda, i EventArgs passerà un tavolo accodamento al sottoscrittore dell'evento. Questa implementazione mi fornirà una coda thread-safe?

Answer 1

Stai rimuovendo i tuoi articoli mentre li accodai.
È necessario aumentare l'evento utilizzando il parametro.

Se è effettivamente thread-safe dipende da come lo si utilizza.
Se si controlla lo Count o si verifica la presenza di vuoto, non è protetto da thread e non può essere reso facilmente protetto da thread.
Se non lo fai, probabilmente puoi usare qualcosa di più semplice di una coda.

Answer 2

Un breve viaggio verso il mio motore di ricerca preferito ha rivelato che la mia memoria era corretta; you can get the Task Parallel Library even on .NET 3.5. Vedi anche The PFX team blog post on the subject e lo Reactive Extensions che scarichi per ottenere allo System.Threading.dll desiderato.

Answer 3

Nella linea OnTableQueued(new TableQueuedEventArgs(Dequeue())); nel metodo Enqueue

uso Peek invece di Dequeue

Dovrebbe essere

OnTableQueued(new TableQueuedEventArgs(base.Peek()));

Answer 4

Il fatto è necessario utilizzare new per nascondere i metodi dalla base la classe indica in genere che è necessario utilizzare la composizione anziché l'ereditarietà ...

Ecco un semplice coda di sincronizzato, che non usa l'ereditarietà, ma ancora si basa sul comportamento dello standard Queue<T>:

public class ConcurrentQueue<T> : ICollection, IEnumerable<T> 
{ 
    private readonly Queue<T> _queue; 

    public ConcurrentQueue() 
    { 
     _queue = new Queue<T>(); 
    } 

    public IEnumerator<T> GetEnumerator() 
    { 
     lock (SyncRoot) 
     { 
      foreach (var item in _queue) 
      { 
       yield return item; 
      } 
     } 
    } 

    IEnumerator IEnumerable.GetEnumerator() 
    { 
     return GetEnumerator(); 
    } 

    public void CopyTo(Array array, int index) 
    { 
     lock (SyncRoot) 
     { 
      ((ICollection)_queue).CopyTo(array, index); 
     } 
    } 

    public int Count 
    { 
     get 
     { 
      // Assumed to be atomic, so locking is unnecessary 
      return _queue.Count; 
     } 
    } 

    public object SyncRoot 
    { 
     get { return ((ICollection)_queue).SyncRoot; } 
    } 

    public bool IsSynchronized 
    { 
     get { return true; } 
    } 

    public void Enqueue(T item) 
    { 
     lock (SyncRoot) 
     { 
      _queue.Enqueue(item); 
     } 
    } 

    public T Dequeue() 
    { 
     lock(SyncRoot) 
     { 
      return _queue.Dequeue(); 
     } 
    } 

    public T Peek() 
    { 
     lock (SyncRoot) 
     { 
      return _queue.Peek(); 
     } 
    } 

    public void Clear() 
    { 
     lock (SyncRoot) 
     { 
      _queue.Clear(); 
     } 
    } 
} 

Answer 5

Qualche tempo dopo la domanda iniziale, lo so (questo è venuto in su come "correlate" a destra di un'altra domanda), ma sono andato con il seguente in casi simili. Non è buono per l'uso della cache della CPU come potrebbe essere, ma l'uso semplice della cache, privo di lock, thread-safe e spesso CPU-CPU non è così importante se ci sono spesso grandi lacune tra le operazioni e quando non la vicinanza dell'allocazione potrebbe ridurre l'impatto:

internal sealed class LockFreeQueue<T> 
{ 
    private sealed class Node 
    { 
    public readonly T Item; 
    public Node Next; 
    public Node(T item) 
    { 
     Item = item; 
    } 
    } 
    private volatile Node _head; 
    private volatile Node _tail; 
    public LockFreeQueue() 
    { 
    _head = _tail = new Node(default(T)); 
    } 
#pragma warning disable 420 // volatile semantics not lost as only by-ref calls are interlocked 
    public void Enqueue(T item) 
    { 
    Node newNode = new Node(item); 
    for(;;) 
    { 
     Node curTail = _tail; 
     if (Interlocked.CompareExchange(ref curTail.Next, newNode, null) == null) //append to the tail if it is indeed the tail. 
     { 
     Interlocked.CompareExchange(ref _tail, newNode, curTail); //CAS in case we were assisted by an obstructed thread. 
     return; 
     } 
     else 
     { 
     Interlocked.CompareExchange(ref _tail, curTail.Next, curTail); //assist obstructing thread. 
     } 
    } 
    }  
    public bool TryDequeue(out T item) 
    { 
    for(;;) 
    { 
     Node curHead = _head; 
     Node curTail = _tail; 
     Node curHeadNext = curHead.Next; 
     if (curHead == curTail) 
     { 
     if (curHeadNext == null) 
     { 
      item = default(T); 
      return false; 
     } 
     else 
      Interlocked.CompareExchange(ref _tail, curHeadNext, curTail); // assist obstructing thread 
     } 
     else 
     { 
     item = curHeadNext.Item; 
     if (Interlocked.CompareExchange(ref _head, curHeadNext, curHead) == curHead) 
     { 
      return true; 
     } 
     } 
    } 
    } 
#pragma warning restore 420 
}