Как я могу убедиться, что ровно один поток будет что-то делать?

У меня есть несколько потоков, которые добавляют элементы в очередь без блокировки.
Затем элементы обрабатываются другим потоком.

В потоках производителей мне нужно запустить поток потребителей, но только если он еще не запущен или не стартовал.

В частности:

public void BeginInvoke(Action method)
{
    //This runs on multiple background threads
    pendingActions.Enqueue(method);
    if (ProcessQueue hasn't been posted)
        uiContext.Post(ProcessQueue, null);
}
private void ProcessQueue(object unused)
{
    //This runs on the UI thread.
    Action current;
    while (pendingActions.TryDequeue(out current))
        current();
}

Я использую .Net 3.5, а не 4.0.: (

Ответ 1

Для этого я создал следующий класс:

///<summary>Ensures that a block of code is only executed once at a time.</summary>
class Valve
{
    int isEntered;  //0 means false; 1 true

    ///<summary>Tries to enter the valve.</summary>
    ///<returns>True if no other thread is in the valve; false if the valve has already been entered.</returns>
    public bool TryEnter()
    {
        if (Interlocked.CompareExchange(ref isEntered, 1, 0) == 0)
            return true;
        return false;
    }

    ///<summary>Allows the valve to be entered again.</summary>
    public void Exit()
    {
        Debug.Assert(isEntered == 1);
        isEntered = 0;
    }
}

Я использую его следующим образом:

readonly Valve valve = new Valve();
public void BeginInvoke(Action method)
{
    pendingActions.Enqueue(method);
    if (valve.TryEnter())
        uiContext.Post(ProcessQueue, null);
}
private void ProcessQueue(object unused)
{
    //This runs on the UI thread.
    Action current;
    while (pendingActions.TryDequeue(out current))
        current();
    valve.Exit();
}

Является ли этот шаблон безопасным?
Есть ли лучший способ сделать это?
Есть ли более корректное имя для класса?

Ответ 2

Самый простой способ - использовать Semaphore. Он будет иметь счет размера очереди.

Ответ 3

Это работает для вас?

volatile int running;  //not a boolean to allow ProcessQueue to be reentrant.

private void ProcessQueue(object unused)
{
    do
    {
        ++running;
        Action current;
        while (pendingActions.TryDequeue(out current))
            current();

        --running;
    }
    while (pendingActions.Count != 0);
} 

public void BeginInvoke(Action method) 
{     
    pendingActions.Enqueue(method);
    if (running != 0)
        uiContext.Post(ProcessQueue, null); 
}

Ответ 4

Создайте второй Dispatcher для потребительского потока. Затем потоки производителей могут использовать этот метод BeginInvoke() диспетчера для отправки данных в потребительский поток. Очередь диспетчера занимает место вашей очереди pendingActions и гарантирует, что потребительский поток обрабатывает только один рабочий элемент за раз.

Вместо того, чтобы потоки производителей пытались координировать запуск и остановку потребительского потока, просто запустите потребительский поток до того, как будут запущены какие-либо производители, и пусть он сидит без дела. Диспетчер должен автоматически позаботиться о том, чтобы просыпаться, когда это необходимо.