将具有大量“锁”的代码重构为更多无锁代码答案

【问题标题】：Refactoring code with a lot of "locks" to more lock-free code将具有大量“锁”的代码重构为更多无锁代码
【发布时间】：2013-02-17 09:34:13
【问题描述】：

更新感谢 Matthew Watson 注意到并注意到我计划将我的代码移植到 c++-linux，因此我更喜欢“平台无关”的代码

我的交易应用程序几乎是无锁的。下面的代码是我使用锁的唯一地方。让我从代码开始，它很长，但不要担心有很多重复的部分，所以它很简单。我更喜欢添加所有“重复”部分以更好地展示我的工作方式：

Task.Factory.StartNew(() =>
{
    while (true)
    {
        Iterate();
    }
}, TaskCreationOptions.LongRunning);

private void Iterate()
{
    bool marketDataUpdated = false;

    lock (ordersToRegisterLock)
    {
        if (ordersToRegister.Count > 0)
        {
            marketDataUpdated = true;
            while (ordersToRegister.Count > 0)
            {
                Order order = ordersToRegister.Dequeue();
                // Stage1, Process
            }
        }
    }

    lock (aggrUpdatesLock)
    {
        if (aggrUpdates.Count > 0)
        {
            marketDataUpdated = true;
            while (!aggrUpdates.IsNullOrEmpty())
            {
                var entry = aggrUpdates.Dequeue();
                // Stage1, Process
            }
        }
    }

    lock (commonUpdatesLock)
    {
        if (commonUpdates.Count > 0)
        {
            marketDataUpdated = true;
            while (!commonUpdates.IsNullOrEmpty())
            {
                var entry = commonUpdates.Dequeue();
                // Stage1, Process
            }
        }
    }

    lock (infoUpdatesLock)
    {
        if (infoUpdates.Count > 0)
        {
            marketDataUpdated = true;
            while (!infoUpdates.IsNullOrEmpty())
            {
                var entry = infoUpdates.Dequeue();
                // Stage1, Process
            }
        }
    }

    lock (tradeUpdatesLock)
    {
        if (tradeUpdates.Count > 0)
        {
            marketDataUpdated = true;
            while (!tradeUpdates.IsNullOrEmpty())
            {
                var entry = tradeUpdates.Dequeue();
                // Stage1, Process
            }    

        }
    }

    if (marketDataUpdated)
    {
        // Stage2 !
        // make a lot of work. expensive operation. recalculate strategies, place orders etc.
    }
}

private readonly Queue<Order> ordersToRegister = new Queue<Order>();
private readonly object ordersToRegisterLock = new object();

private readonly Queue<AggrEntry> aggrUpdates = new Queue<AggrEntry>();
private readonly object aggrUpdatesLock = new object();

private readonly Queue<CommonEntry> commonUpdates = new Queue<CommonEntry>();
private readonly object commonUpdatesLock = new object();

private readonly Queue<InfoEntry> infoUpdates = new Queue<InfoEntry>();
private readonly object infoUpdatesLock = new object();

private readonly Queue<TradeEntry> tradeUpdates = new Queue<TradeEntry>();
private readonly object tradeUpdatesLock = new object();


    public void RegistorOrder(object sender, Gate.RegisterOrderArgs e)
    {
        lock (ordersToRegisterLock)
        {
            ordersToRegister.Enqueue(e.order);
        }
    }

    public void TradeUpdated(object sender, Gate.TradeArgs e)
    {
        lock (tradeUpdatesLock)
        {
            foreach (var entry in e.entries)
            {
                tradeUpdates.Enqueue(entry);
            }
        }
    }

    public void InfoUpdated(object sender, Gate.InfoArgs e)
    {
        lock (infoUpdatesLock)
        {
            foreach (var entry in e.entries)
            {
                infoUpdates.Enqueue(entry);
            }
        }
    }

    public void CommonUpdated(object sender, Gate.CommonArgs e)
    {
        lock (commonUpdatesLock)
        {
            foreach (var entry in e.entries)
            {
                commonUpdates.Enqueue(entry);
            }
        }
    }

    public void AggrUpdated(object sender, Gate.AggrArgs e)
    {
        lock (aggrUpdatesLock)
        {
            foreach (var entry in e.entries)
            {
                aggrUpdates.Enqueue(entry);
            }
        }
    }

在我的代码中，我有两个阶段。 Stage1 是更新阶段，Stage2 是工作阶段。我需要尽快在这两个阶段之间切换，就像这样：

有更新吗？没有
有更新吗？没有
有更新吗？是的，订单已更新！申请更新，做Stage2
有更新吗？没有
有更新吗？是的，订单需要注册！申请更新，做Stage2
有更新吗？是的，交易发生了，申请更新，做Stage2

在Stage2 中，我不应该更新，但应该继续“收集”更新，以便以后应用它们。

重要的是 - 这是对延迟非常关键的代码，所以我同意“花费”一个内核来获得 minimal 延迟！所以当任何更新发生时，我需要尽快处理它并执行Stage2。

所以我希望现在很清楚我需要实现什么，并且很清楚我是如何实现的。现在是时候讨论我的代码有多好了。我确实看到了几个潜在的问题：

很多锁！可以用一些“无锁”代码代替吗？用 CAS 或什么的自旋锁？
占用 100% 的 CPU 内核，是否可以节省一些 CPU 资源不影响延迟？
我可以/应该告诉 .NET 使用“专用”核心（设置任务关联？）以避免额外的“切换”吗？
我从一个线程添加到队列，我从另一个线程读取队列。会不会是个问题？如果添加和读取队列是易变的？我的 reading 线程是否有可能因为缓存更新问题而看不到来自队列的更新？

欢迎提出改进我所写内容的任何建议，谢谢！

upd 部分解决了 - 据我了解，我最好将查询替换为无锁（可能基于环形缓冲区？）查询。我想我稍后会使用 c++ 版本的中断器。我也使用了这篇文章http://www.umbraworks.net/bl0g/rebuildall/2010/03/08/Running_NET_threads_on_selected_processor_cores 并用在“固定”核心上运行的线程替换了任务，但是我仍在使用“忙转”，也许我应该使用更智能的东西？

【问题讨论】：

如果您要锁定线程安全队列访问，请尝试使用ConcurrentQueue<T> (.NET 4.0+)。如需进一步阅读，请参阅：-ConcurrentQueue in a nutshell（简要说明）-ConcurrentQueue vs. Queue（表明 CQ 比 Q+Lock 更快）
是的，这正是为这种事情而设计的。另外，请查看 BlockingCollection：msdn.microsoft.com/en-us/library/dd267312.aspx
无锁代码会增加延迟。您正在寻找免等待代码。
在我这个非常简单的案例中不需要使用ConcurrentQueue。只需Queue 就足够了，而且很可能会更快。至少因为我获得了一次锁定并添加了很多元素，ConcurrentQueue是不可能的
小心人们。他显然想用 C++ 重写这一切，但他不想告诉我们。似乎他宁愿我们浪费时间给出错误的答案。

标签： c# locking task-parallel-library lock-free low-latency

【解决方案1】：

使用下面的代码，您在“阶段 1”处理期间不再被锁定：

Task.Factory.StartNew(() =>
{
    while (true)
    {
        Iterate();
    }
}, TaskCreationOptions.LongRunning);


private void Iterate()
{
    bool marketDataUpdated = false;

    foreach (Order order in ordersToRegister)
    {
        marketDataUpdated = true;
        // Stage1, Process
    }

    foreach (var entry in aggrUpdates)
    {
        marketDataUpdated = true;
        // Stage1, Process
    }

    foreach (var entry in commonUpdates)
    {
        marketDataUpdated = true;
        // Stage1, Process
    }

    foreach (var entry in infoUpdates)
    {
        marketDataUpdated = true;
        // Stage1, Process
    }

    foreach (var entry in tradeUpdates)
    {
        marketDataUpdated = true;
        // Stage1, Process
    }

    if (marketDataUpdated)
    {
        // Stage2 !
        // make a lot of work. expensive operation. recalculate strategies, place orders etc.
    }
}

private readonly ConcurrentQueue<Order> ordersToRegister = new ConcurrentQueue<Order>();

private readonly ConcurrentQueue<AggrEntry> aggrUpdates = new ConcurrentQueue<AggrEntry>();

private readonly ConcurrentQueue<CommonEntry> commonUpdates = new ConcurrentQueue<CommonEntry>();

private readonly ConcurrentQueue<InfoEntry> infoUpdates = new ConcurrentQueue<InfoEntry>();

private readonly ConcurrentQueue<TradeEntry> tradeUpdates = new ConcurrentQueue<TradeEntry>();

    public void RegistorOrder(object sender, Gate.RegisterOrderArgs e)
    {
        ordersToRegister.Enqueue(e.order);
    }

    public void TradeUpdated(object sender, Gate.TradeArgs e)
    {
        foreach (var entry in e.entries)
        {
            tradeUpdates.Enqueue(entry);
        }
    }

    public void InfoUpdated(object sender, Gate.InfoArgs e)
    {
        foreach (var entry in e.entries)
        {
            infoUpdates.Enqueue(entry);
        }
    }

    public void CommonUpdated(object sender, Gate.CommonArgs e)
    {
        foreach (var entry in e.entries)
        {
            commonUpdates.Enqueue(entry);
        }
    }

    public void AggrUpdated(object sender, Gate.AggrArgs e)
    {
        foreach (var entry in e.entries)
        {
            aggrUpdates.Enqueue(entry);
        }
    }

【讨论】：

谢谢，这就是我所做的。我仍在锁定，但在 ConcurrentQueue 中。当我决定移植到 Linux 时，这将是可移植性问题。

【解决方案2】：

这是一种可能更便携的方法。希望对您有所帮助。

public class SafeQueue<T> : Queue<T>
{
    public T SafeDequeue()
    {
        lock (this)
        {
            return (Count > 0) ? Dequeue() : null;
        }
    }

    public void SafeEnqueue(T entry)
    {
        lock (this)
        {
            Enqueue(entry);
        }
    }
}

Task.Factory.StartNew(() =>
{
    while (true)
    {
        Iterate();
    }
}, TaskCreationOptions.LongRunning);


private void Iterate()
{
    bool marketDataUpdated = false;

    while ((Order order = ordersToRegister.SafeDequeue()) != null)
    {
        marketDataUpdated = true;
        // Stage1, Process
    }

    while ((var entry = aggrUpdates.SafeDequeue()) != null)
    {
        marketDataUpdated = true;
        // Stage1, Process
    }

    while ((var entry = commonUpdates.SafeDequeue()) != null)
    {
        marketDataUpdated = true;
        // Stage1, Process
    }

    while ((var entry = infoUpdates.SafeDequeue()) != null)
    {
        marketDataUpdated = true;
        // Stage1, Process
    }

    while ((var entry = tradeUpdates.SafeDequeue()) != null)
    {
        marketDataUpdated = true;
        // Stage1, Process
    }

    if (marketDataUpdated)
    {
        // Stage2 !
        // make a lot of work. expensive operation. recalculate strategies, place orders etc.
    }
}

private readonly SafeQueue<Order> ordersToRegister = new SafeQueue<Order>();

private readonly SafeQueue<AggrEntry> aggrUpdates = new SafeQueue<AggrEntry>();

private readonly SafeQueue<CommonEntry> commonUpdates = new SafeQueue<CommonEntry>();

private readonly SafeQueue<InfoEntry> infoUpdates = new SafeQueue<InfoEntry>();

private readonly SafeQueue<TradeEntry> tradeUpdates = new SafeQueue<TradeEntry>();

    public void RegistorOrder(object sender, Gate.RegisterOrderArgs e)
    {
        ordersToRegister.SafeEnqueue(e.order);
    }

    public void TradeUpdated(object sender, Gate.TradeArgs e)
    {
        foreach (var entry in e.entries)
        {
            tradeUpdates.SafeEnqueue(entry);
        }
    }

    public void InfoUpdated(object sender, Gate.InfoArgs e)
    {
        foreach (var entry in e.entries)
        {
            infoUpdates.SafeEnqueue(entry);
        }
    }

    public void CommonUpdated(object sender, Gate.CommonArgs e)
    {
        foreach (var entry in e.entries)
        {
            commonUpdates.SafeEnqueue(entry);
        }
    }

    public void AggrUpdated(object sender, Gate.AggrArgs e)
    {
        foreach (var entry in e.entries)
        {
            aggrUpdates.SafeEnqueue(entry);
        }
    }

【讨论】：