我可以更好地优化这种并发性吗？

Question

我最近开始了我的第一个多线程代码，我很感激一些 cmets。

它从缓冲区提供视频样本，该缓冲区由流解析器在后台填充（不在本问题的范围内）。如果缓冲区为空，则需要等到缓冲区级别变得可以接受后再继续。

代码适用于 Silverlight 4，删除了一些错误检查：

// External class requests samples - can happen multiple times concurrently
protected override void GetSampleAsync()
{
    Interlocked.Add(ref getVideoSampleRequestsOutstanding, 1);                
}

// Runs on a background thread
void DoVideoPumping()
{
    do
    {
        if (getVideoSampleRequestsOutstanding > 0)
        {
            PumpNextVideoSample();

            // Decrement the counter
            Interlocked.Add(ref getVideoSampleRequestsOutstanding, -1);
        }
        else Thread.Sleep(0);  

    } while (!this.StopAllBackgroundThreads);
}       

void PumpNextVideoSample()
{
    // If the video sample buffer is empty, tell stream parser to give us more samples 
    bool MyVidBufferIsEmpty = false; bool hlsClientIsExhausted = false;
    ParseMoreSamplesIfMyVideoBufferIsLow(ref MyVidBufferIsEmpty, ref parserAtEndOfStream);

    if (parserAtEndOfStream)  // No more data, start running down buffers
        this.RunningDownVideoBuffer = true;
    else if (MyVidBufferIsEmpty)  
    {
        // Buffer is empty, wait for samples
        WaitingOnEmptyVideoBuffer = true;
        WaitOnEmptyVideoBuffer.WaitOne();
    }

    // Buffer is OK
    nextSample = DeQueueVideoSample(); // thread-safe, returns NULL if a problem

    // Send the sample to the external renderer
    ReportGetSampleCompleted(nextSample);

}

代码似乎运行良好。但是，有人告诉我使用 Thread.Wait(...) 是“邪恶的”：当没有请求样本时，我的代码会不必要地循环，占用 CPU 时间。

我的代码可以进一步优化吗？由于我的课程是为需要样本的环境设计的，潜在的“无意义循环”场景是否超过了当前设计的简单性？

非常感谢评论。

Answer 1

这看起来像经典的生产者/消费者模式。解决这个问题的常规方法是使用所谓的阻塞队列。

.net 4.0 版针对此类问题引入了一组高效、精心设计的concurrent collection classes。我认为BlockingCollection<T> 将满足您当前的需求。

如果您无法访问 .net 4.0，那么有许多网站包含阻塞队列的实现。就我个人而言，我的标准参考是 Joe Duffy 的书，Concurrent Programming on Windows。 Marc Gravell's blocking queue presented here in Stack Overflow 是一个好的开始。

使用阻塞队列的第一个优点是您不再使用繁忙的等待循环、对Sleep() 的恶意调用等。使用阻塞队列来避免此类代码总是一个好主意。

但是，我认为使用阻塞队列有一个更重要的好处。目前，您生成工作项、使用它们和处理队列的代码都是混合在一起的。如果你正确使用阻塞队列，那么你最终会得到更好的分解代码，它将算法的各个组件分开：队列、生产者和消费者。

Answer 2

你有一个主要问题：Thread.Sleep()

它的粒度约为 20 毫秒，这对于视频来说有点粗糙。此外，Sleep(0) 可能会出现低优先级线程 [] 的饥饿问题。

更好的方法是等待 Waithandle，最好内置在队列中。

Answer 3

Blocking queue 是一个很好且简单的阻塞队列示例。
主要的关键是线程需要与信号协调，而不是通过检查计数器的值或数据结构的状态。任何检查都需要资源 (CPU)，因此您需要信号（Monitor.Wait 和 Monitor.Pulse）。

Answer 4

您可以使用AutoResetEvent 而不是手动thread.sleep。这样做相当简单：

AutoResetEvent e;
void RequestSample()
{
    Interlocked.Increment(ref requestsOutstanding);
    e.Set(); //also set this when StopAllBackgroundThreads=true!
}

void Pump()
{
    while (!this.StopAllBackgroundThreads) {
        e.WaitOne();
        int leftOver = Interlocked.Decrement(ref requestsOutstanding);
        while(leftOver >= 0) {
            PumpNextVideoSample();
            leftOver = Interlocked.Decrement(ref requestsOutstanding);
        }
        Interlocked.Increment(ref requestsOutstanding);
    }
}

请注意，实现信号量可能更有吸引力。基本上;无论如何，同步开销在您的场景中可能几乎为零，并且更简单的编程模型是值得的。使用信号量，你会得到这样的东西：

MySemaphore sem;
void RequestSample()
{
    sem.Release();
}

void Pump()
{
    while (true) {
        sem.Acquire();
        if(this.StopAllBackgroundThreads) break;
        PumpNextVideoSample();
    }
}

...我会说简单是值得的！

例如信号量的简单实现：

public sealed class SimpleSemaphore
{
    readonly object sync = new object();
    public int val;

    public void WaitOne()
    {
        lock(sync) {
            while(true) {
                if(val > 0) {
                    val--;
                    return;
                }
                Monitor.Wait(sync);
            }
        }
    }

    public void Release()
    {
        lock(sync) {
            if(val==int.MaxValue)
                throw new Exception("Too many releases without waits.");
            val++;
            Monitor.Pulse(sync);
        }
    }
}

在一个简单的基准测试中，这个简单的实现需要大约 1.7 秒，而 Semaphore 需要 7.5 而 SemaphoreSlim 需要 1.1；换句话说，非常合理。