Criando uma fila de bloqueio <T> no .NET?

163

Eu tenho um cenário em que vários threads são adicionados a uma fila e vários threads são lidos na mesma fila. Se a fila atingir um tamanho específico, todos os segmentos que estão preenchendo a fila serão bloqueados na adição até que um item seja removido da fila.

A solução abaixo é o que estou usando agora e minha pergunta é: como isso pode ser melhorado? Existe um objeto que já habilite esse comportamento na BCL que eu deveria estar usando?

internal class BlockingCollection<T> : CollectionBase, IEnumerable
{
    //todo: might be worth changing this into a proper QUEUE

    private AutoResetEvent _FullEvent = new AutoResetEvent(false);

    internal T this[int i]
    {
        get { return (T) List[i]; }
    }

    private int _MaxSize;
    internal int MaxSize
    {
        get { return _MaxSize; }
        set
        {
            _MaxSize = value;
            checkSize();
        }
    }

    internal BlockingCollection(int maxSize)
    {
        MaxSize = maxSize;
    }

    internal void Add(T item)
    {
        Trace.WriteLine(string.Format("BlockingCollection add waiting: {0}", Thread.CurrentThread.ManagedThreadId));

        _FullEvent.WaitOne();

        List.Add(item);

        Trace.WriteLine(string.Format("BlockingCollection item added: {0}", Thread.CurrentThread.ManagedThreadId));

        checkSize();
    }

    internal void Remove(T item)
    {
        lock (List)
        {
            List.Remove(item);
        }

        Trace.WriteLine(string.Format("BlockingCollection item removed: {0}", Thread.CurrentThread.ManagedThreadId));
    }

    protected override void OnRemoveComplete(int index, object value)
    {
        checkSize();
        base.OnRemoveComplete(index, value);
    }

    internal new IEnumerator GetEnumerator()
    {
        return List.GetEnumerator();
    }

    private void checkSize()
    {
        if (Count < MaxSize)
        {
            Trace.WriteLine(string.Format("BlockingCollection FullEvent set: {0}", Thread.CurrentThread.ManagedThreadId));
            _FullEvent.Set();
        }
        else
        {
            Trace.WriteLine(string.Format("BlockingCollection FullEvent reset: {0}", Thread.CurrentThread.ManagedThreadId));
            _FullEvent.Reset();
        }
    }
}
Eric Schoonover
fonte
5
.Net como classes internas para ajudar com este cenário. A maioria das respostas listadas aqui são obsoletas. Veja as respostas mais recentes na parte inferior. Examine as coleções de bloqueio seguras para threads. As respostas podem ser obsoletas, mas ainda é uma boa pergunta!
Tom A
Acho que ainda é uma boa idéia aprender sobre o Monitor.Wait / Pulse / PulseAll, mesmo se tivermos novas classes simultâneas no .NET.
thewpfguy
1
Concorde com @thewpfguy. Você desejará compreender os mecanismos básicos de bloqueio nos bastidores. Também é importante notar que o Systems.Collections.Concurrent não existia até abril de 2010 e somente no Visual Studio 2010 e versões posteriores. Definitivamente não é uma opção para os VS2008 saídas de espera ...
Vic
Se você está lendo isso agora, consulte System.Threading.Channels para uma implementação multi-writer / multi-reader, limitada e opcionalmente bloqueada, para o .NET Core e o .NET Standard.
precisa saber é o seguinte

Respostas:

200

Isso parece muito inseguro (muito pouca sincronização); que tal algo como:

class SizeQueue<T>
{
    private readonly Queue<T> queue = new Queue<T>();
    private readonly int maxSize;
    public SizeQueue(int maxSize) { this.maxSize = maxSize; }

    public void Enqueue(T item)
    {
        lock (queue)
        {
            while (queue.Count >= maxSize)
            {
                Monitor.Wait(queue);
            }
            queue.Enqueue(item);
            if (queue.Count == 1)
            {
                // wake up any blocked dequeue
                Monitor.PulseAll(queue);
            }
        }
    }
    public T Dequeue()
    {
        lock (queue)
        {
            while (queue.Count == 0)
            {
                Monitor.Wait(queue);
            }
            T item = queue.Dequeue();
            if (queue.Count == maxSize - 1)
            {
                // wake up any blocked enqueue
                Monitor.PulseAll(queue);
            }
            return item;
        }
    }
}

(editar)

Na realidade, você desejaria uma maneira de fechar a fila para que os leitores comecem a sair de maneira limpa - talvez algo como um sinalizador bool - se configurada, uma fila vazia retornará (em vez de bloquear):

bool closing;
public void Close()
{
    lock(queue)
    {
        closing = true;
        Monitor.PulseAll(queue);
    }
}
public bool TryDequeue(out T value)
{
    lock (queue)
    {
        while (queue.Count == 0)
        {
            if (closing)
            {
                value = default(T);
                return false;
            }
            Monitor.Wait(queue);
        }
        value = queue.Dequeue();
        if (queue.Count == maxSize - 1)
        {
            // wake up any blocked enqueue
            Monitor.PulseAll(queue);
        }
        return true;
    }
}
Marc Gravell
fonte
1
Que tal mudar o tempo de espera para um WaitAny e passando em um cessar WaitHandle em construção ...
Sam Saffron
1
@ A otimização de Marc, se você esperava que a fila sempre atingisse a capacidade, seria passar o valor maxSize para o construtor da Fila <T>. Você pode adicionar outro construtor à sua classe para acomodar isso.
precisa saber é o seguinte
3
Por que SizeQueue, por que não FixedSizeQueue?
mindless.panda
4
@Lasse - libera o (s) bloqueio (s) durante Wait, para que outros threads possam adquiri-lo. Recupera o (s) bloqueio (s) quando acorda.
Marc Gravell
1
Bom, como eu disse, havia algo que eu não estava recebendo :) Isso com certeza me faz querer revisitar parte do meu código de thread ....
Lasse V. Karlsen
14

"Como isso pode ser melhorado?"

Bem, você precisa examinar todos os métodos da sua classe e considerar o que aconteceria se outro thread estivesse chamando esse método ou qualquer outro método simultaneamente. Por exemplo, você coloca um bloqueio no método Remove, mas não no método Add. O que acontece se um thread é adicionado ao mesmo tempo que outro thread removido?Coisas ruins.

Considere também que um método pode retornar um segundo objeto que fornece acesso aos dados internos do primeiro objeto - por exemplo, GetEnumerator. Imagine que um segmento está passando por esse enumerador, outro segmento está modificando a lista ao mesmo tempo. Não é bom.

Uma boa regra é simplificar o processo, reduzindo o número de métodos da classe ao mínimo absoluto.

Em particular, não herde outra classe de contêiner, porque você exporá todos os métodos dessa classe, fornecendo uma maneira de o chamador corromper os dados internos ou ver alterações parcialmente completas nos dados (tão ruins quanto os dados). aparece corrompido naquele momento). Esconda todos os detalhes e seja completamente implacável sobre como você permite o acesso a eles.

Eu recomendo fortemente que você use soluções disponíveis no mercado - obtenha um livro sobre encadeamento ou use uma biblioteca de terceiros. Caso contrário, dado o que você está tentando, você estará depurando seu código por um longo tempo.

Além disso, não faria mais sentido remover remover um item (por exemplo, o que foi adicionado primeiro, por ser uma fila), em vez de o chamador escolher um item específico? E quando a fila estiver vazia, talvez o Remove também deva bloquear.

Atualização: a resposta de Marc realmente implementa todas essas sugestões! :) Mas vou deixar isso aqui, pois pode ser útil entender por que a versão dele é uma melhoria.

Daniel Earwicker
fonte
12

Você pode usar o BlockingCollection e ConcurrentQueue no espaço para nome System.Collections.Concurrent

 public class ProducerConsumerQueue<T> : BlockingCollection<T>
{
    /// <summary>
    /// Initializes a new instance of the ProducerConsumerQueue, Use Add and TryAdd for Enqueue and TryEnqueue and Take and TryTake for Dequeue and TryDequeue functionality
    /// </summary>
    public ProducerConsumerQueue()  
        : base(new ConcurrentQueue<T>())
    {
    }

  /// <summary>
  /// Initializes a new instance of the ProducerConsumerQueue, Use Add and TryAdd for Enqueue and TryEnqueue and Take and TryTake for Dequeue and TryDequeue functionality
  /// </summary>
  /// <param name="maxSize"></param>
    public ProducerConsumerQueue(int maxSize)
        : base(new ConcurrentQueue<T>(), maxSize)
    {
    }



}
Andreas
fonte
3
O padrão BlockingCollection é Fila. Então, não acho que isso seja necessário.
Curtis White
BlockingCollection preserva a ordem como uma fila?
joelc
Sim, quando é inicializado com um ConcurrentQueue
Andreas
6

Acabei de terminar isso usando as extensões reativas e lembrei-me desta pergunta:

public class BlockingQueue<T>
{
    private readonly Subject<T> _queue;
    private readonly IEnumerator<T> _enumerator;
    private readonly object _sync = new object();

    public BlockingQueue()
    {
        _queue = new Subject<T>();
        _enumerator = _queue.GetEnumerator();
    }

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

    public T Dequeue()
    {
        _enumerator.MoveNext();
        return _enumerator.Current;
    }
}

Não necessariamente totalmente seguro, mas muito simples.

Mark Rendle
fonte
O que é o Assunto <t>? Não tenho nenhum resolvedor para seu espaço de nome.
theJerm
Faz parte das extensões reativas.
Mark Rendle
Não é uma resposta. Isso não responde à pergunta.
Makhdumi:
5

Foi isso que optei por uma fila de bloqueio limitada por thread.

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

public class BlockingBuffer<T>
{
    private Object t_lock;
    private Semaphore sema_NotEmpty;
    private Semaphore sema_NotFull;
    private T[] buf;

    private int getFromIndex;
    private int putToIndex;
    private int size;
    private int numItems;

    public BlockingBuffer(int Capacity)
    {
        if (Capacity <= 0)
            throw new ArgumentOutOfRangeException("Capacity must be larger than 0");

        t_lock = new Object();
        buf = new T[Capacity];
        sema_NotEmpty = new Semaphore(0, Capacity);
        sema_NotFull = new Semaphore(Capacity, Capacity);
        getFromIndex = 0;
        putToIndex = 0;
        size = Capacity;
        numItems = 0;
    }

    public void put(T item)
    {
        sema_NotFull.WaitOne();
        lock (t_lock)
        {
            while (numItems == size)
            {
                Monitor.Pulse(t_lock);
                Monitor.Wait(t_lock);
            }

            buf[putToIndex++] = item;

            if (putToIndex == size)
                putToIndex = 0;

            numItems++;

            Monitor.Pulse(t_lock);

        }
        sema_NotEmpty.Release();


    }

    public T take()
    {
        T item;

        sema_NotEmpty.WaitOne();
        lock (t_lock)
        {

            while (numItems == 0)
            {
                Monitor.Pulse(t_lock);
                Monitor.Wait(t_lock);
            }

            item = buf[getFromIndex++];

            if (getFromIndex == size)
                getFromIndex = 0;

            numItems--;

            Monitor.Pulse(t_lock);

        }
        sema_NotFull.Release();

        return item;
    }
}
Kevin
fonte
Você poderia fornecer alguns exemplos de código de como eu enfileiraria algumas funções de encadeamento usando esta biblioteca, incluindo como eu instanciaria essa classe?
theJerm
Esta pergunta / resposta é um pouco datada. Você deve consultar o espaço de nome System.Collections.Concurrent para bloquear o suporte da fila.
Kevin
2

Eu não explorei completamente o TPL, mas eles podem ter algo que atenda às suas necessidades, ou pelo menos alguma forragem do Reflector para obter alguma inspiração.

Espero que ajude.

TheMissingLINQ
fonte
Estou ciente de que isso é antigo, mas meu comentário é para os novatos no SO, pois o OP já sabe disso hoje. Esta não é uma resposta, deveria ter sido um comentário.
John Demetriou
0

Bem, você pode olhar para a System.Threading.Semaphoreaula. Fora isso - não, você tem que fazer isso sozinho. AFAIK não existe essa coleção interna.

Vilx-
fonte
Eu olhei para isso limitando o número de threads que estão acessando um recurso, mas ele não permite que você bloqueie todo o acesso a um recurso com base em alguma condição (como Collection.Count). AFAIK enfim
Eric Schoonover
Bem, você faz essa parte, como faz agora. Simplesmente, em vez de MaxSize e _FullEvent, você tem o Semáforo, que você inicializa com a contagem correta no construtor. Em seguida, em cada Adicionar / Remover, você chama WaitForOne () ou Release ().
Vilx-
Não é muito diferente do que você tem agora. IMHO apenas mais simples.
Vilx-
Você pode me dar um exemplo mostrando isso funcionando? Não vi como ajustar dinamicamente o tamanho de um semáforo, que esse cenário exige. Como você deve poder bloquear todos os recursos apenas se a fila estiver cheia.
9119 Eric Schoonover
Ahh, mudando de tamanho! Por que você não disse isso imediatamente? OK, então um semáforo não é para você. Boa sorte com esta abordagem!
Vilx-
-1

Se você deseja um rendimento máximo, permitindo que vários leitores leiam e apenas um escritor escreva, o BCL possui algo chamado ReaderWriterLockSlim que deve ajudar a diminuir o seu código ...

DavidN
fonte
Eu quero que ninguém seja capaz de escrever se a fila estiver cheia.
9119 Eric Schoonover
Então você combina com uma fechadura. Aqui estão alguns exemplos muito bons: albahari.com/threading/part2.aspx#_ProducerConsumerQWaitHandle albahari.com/threading/part4.aspx
DavidN
3
Com fila / desenfileiramento, todos são escritores ... um bloqueio exclusivo talvez seja mais pragmático
Marc Gravell
Estou ciente de que isso é antigo, mas meu comentário é para os novatos no SO, pois o OP já sabe disso hoje. Esta não é uma resposta, deveria ter sido um comentário.
John Demetriou