使用 SharedMemory MMF 实现快速 .NET 无锁进程间答案

【问题标题】：implementing a Fast .NET Lock-Free Inter-Process using SharedMemory MMF使用 SharedMemory MMF 实现快速 .NET 无锁进程间
【发布时间】：2016-06-07 00:40:36
【问题描述】：

我是 multitasking 和 IPC 的新手，我正在尝试构建一种使用共享内存进行快速进程间通信的方法（起初我正在研究 IPC 术语，考虑到 wcf sockets 和named pipes 只是为了最终发现 MMF）。

现在我已经通过使用Lock 和EventWaitHandle 信号成功地在两个进程之间使用共享内存实现了一个小型测试，我正在尝试一种实现非阻塞/无等待模式的方法。现在，我正在尝试结合Thread.MemoryBarrier() 和从MemoryMapedFile 读取signalling Sector。

问题不明！第一轮通过，第二轮最后一次位于百慕大三角......超出了调试器的范围......

说进程a正在向进程b发送一连串showMsg()请求。

                                         //offset positions in mmf
MemoryMappedViewAccessor MmfAcc; const int opReady= 0, opCompleteRead = 4, .....



ReadTrd()
{
  //[0,3] - Reader is stationed
  //[4,7] - Read Complete successfully
  //[8,11] - Data-size 
  //[12,15] - Reader-exiting
  "format" the signals Section (write zeroes). 

   for(;;){if (WrTrd-StepMMF1  Confimed) break;}
  MmfAcc- read DataSize val @offset[8]
  MmfAcc- read Data val @offset[50]
  MmfAcc Write exit to offset....
  ....heavy use of  Thread.MemoryBarrier(); sets !!! (all over the place, on every shared variable...)

}

writeTrd()
{
  heavy use of  Thread.MemoryBarrier() !!!
  //[15-19] - wr is stationed
  //[20-23] - wr Complete successfully
  //[24-27] - wrExiting
  "format" the signals Section . 
   for(;;){if Reader-StepMMF1 is Confim break;}
   MmfAcc- DataSize to offset[8]
   write Data To offset[50] using the method below
   for(;;){if Read StepMMF2 is Confim break;}
 }

当我第一次使用命名管道解决方案时，与命名管道方法相比，Mmf 方法（尽管有 Lock 和 EventWaitHandle）在性能上有很大的提升，但我可以进一步使用上述方法以某种方式..？

我可以像条带 Raid 一样克隆这种模式 ...

Reader1 + Reader2 & WriteThred1  + WriteThread2

所以我尝试了它并卡在了那个点。

这种使用全内存围栏和共享内存进行信号传输的有效方法吗？

如果是这样，剩下的就是看第二次迭代失败的原因，性能差异。

EDIT - 添加额外线程测试背后的逻辑

这是我用来操作编写器线程的“桥梁”（读者使用相同的方法。

public void Write(byte[] parCurData)
{
    if (ReadPosition < 0 || WritePosition < 0)
        throw new ArgumentException();
    this.statusSet.Add("ReadWrite:-> " + ReadPosition + "-" + WritePosition);
    // var s = (FsMomitorIPCCrier)data;

    ////////lock (this.dataToSend)
    ////////{
    Thread.MemoryBarrier();
        LiveDataCount_CurIndex = dataQue.Where(i => i != null).Count();
    this.dataQue[LiveDataCount_CurIndex] = parCurData;

    Console.WriteLine("^^^^^" + Thread.CurrentThread.Name + " has Entered WritingThreads BRIDGE");
    Console.WriteLine("^^^^^[transactionsQue] = {1}{0}^^^^^[dataQue.LiveDataASIndex = {2}{0}^^^^^[Current Requests Count = {3}{0}", "\r\n", Wtransactions, LiveDataCount_CurIndex, ++dataDelReqCount);

    //this.itsTimeForWTrd2 = false;

    if (Wtransactions != 0 && Wtransactions > ThrededSafeQ_Initial_Capcity - 1)
    if (this.dataQueISFluded) this.DataQXpand();


    if (itsTimeForWTrd2)
    {

        bool firstWt = true;
        while (writerThread2Running)
        {
            if (!firstWt) continue;
            Console.WriteLine("SECOND WRITERThread [2] is In The CoffeeCorner");
                    firstWt=false;
        }

        this.dataDelivery2 = this.dataQue[LiveDataCount_CurIndex];
        Console.WriteLine("Activating SECOND WRITERThread [2]");
        itsTimeForWTrd2 = false;

        writerThread2Running = true;
        //writerThread1Running = true;
        writerThread2 = new System.Threading.Thread(WriterThread2);
        writerThread2.IsBackground = true;
        writerThread2.Name = this.DepoThreadName + "=[WRITER2]";
        writerThread2.Start();

    }
    else
    {
        bool firstWt = true;
        while (writerThread1Running)
        {
            if (!firstWt)continue;
                Console.WriteLine("WRITERThread [1] is In The CoffeeCorner");
            firstWt=false;
        }
        Console.WriteLine("Activating WRITERThread [1]");
        this.dataDelivery1 = this.dataQue[LiveDataCount_CurIndex]; 

        writerThread1Running = true;
        writerThread1 = new System.Threading.Thread(WriterThread1);
        writerThread1.IsBackground = true;
        writerThread1.Name = this.DepoThreadName+"=[WRITER1]";
        writerThread1.Start();
        itsTimeForWTrd2 = true;

    }
    Thread.MemoryBarrier();
}

使用写句柄读写实际数据（写类似代码）

public unsafe byte[] UsReadBytes(int offset, int num)
{
    byte[] arr = new byte[num];
    byte* ptr = (byte*)0;
    this.accessor.SafeMemoryMappedViewHandle.AcquirePointer(ref ptr);
    Marshal.Copy(IntPtr.Add(new IntPtr(ptr), offset), arr, 0, num);
    this.accessor.SafeMemoryMappedViewHandle.ReleasePointer();
    return arr;
}

正如我所说，我已经通过非阻塞/无等待等研究了数据和共享内存的同步问题。信号量锁，所以我试图在每个数据事务处理过程中消除任何类型的开销到共享内存映射文件中。我在这里想问一下，消除 Lock 和 EventWaitHandle 并用内存栅栏的逻辑和通过 mmf 发出信号的逻辑替换它可能是什么问题？

【问题讨论】：

我真的不明白为什么这会在一分钟内得到三个赞成票（很可能是因为“哦，很酷，内存映射文件”）；您是否可以花一些精力正确大写并在句子中使用标点符号，并尝试逐步解释此代码应该做什么以及它在哪里停止执行您期望的操作？我不完全确定，但我认为显示一些实际代码而不是这个伪代码可能也会有所帮助。
没问题，我认为你手上有一个很有趣的问题，我对线程和内存映射文件有点了解，只是它非常很难弄清楚实际的问题。就是这样。
这一切都是错误的，你根本不需要锁。你真正需要的是一个event，它告诉你另一个线程已经完成了数据的写入，这样你就可以可靠地读取它了。一个可等待的事件，这样您就不会不必要地等待其他线程完成工作而烧毁核心。在 .NET 中很容易获得，例如使用命名的 AutoResetEvent。在写入的线程中调用它的 Set() 方法，在读取的线程中调用它的 WaitOne() 方法。在没有比赛的情况下让它变得可靠会让你忙碌一段时间，很有可能你会重新发明一个命名管道。
@HansPassant blog.benoitblanchon.fr/csharp-concurrency-cheat-sheet 这是所有开销的一些图表
过早优化是根本...

标签： c# multithreading performance ipc wait-free

【解决方案1】：

如果您打算将其用于研发以外的特定目的，最简单的方法是使用已经提供该功能的库。考虑这一点的一种方法是，无锁 = 消息传递。两种可能的方法是：对于简约的消息传递实现，ZeroMQ IPC，它提供了出色的 .Net 支持和出色的 IPC 性能 (http://www.codeproject.com/Articles/488207/ZeroMQ-via-Csharp-Introduction)，对于更完整的 Actor-Model 实现（包括对 IPC 的支持），请查看 Akka .net (http://getakka.net/docs/#networking)。

如果目的是更多的研发性质（意思是，你想要编写自己的实现，这很酷），我仍然建议查看这些产品的来源（尤其是 Akka .net，因为它是用 C# 编写的），用于有关消息传递和基于 Actor 的 IPC 的实现思路。

【讨论】：