justinstenning / sharedmemory Goto Github PK

I tried creating one large array of ints, basically 100 million ints. And I compared it to 10 arrays of 10 million ints. I don't see a see a measurable difference in the processes memory utilization or how long it takes.

Does that mean that if my application requires hundreds, if not thousands of arrays, i should create each of the as a separate array?

If I need to create one big array, I'd probably create an ArraySegment or ArraySlice that works kinda like this: http://referencesource.microsoft.com/#q=ArraySegment

but i may not need to do that!

Hi
Is there an option to add a method to be called as event on second inter process when the inter process one is writing a buffer? in other words, how to continuously listen to the buffer and find when to read it?
watch this video for an example of what i have in mind, take note, the second application does not know when the message is being sent and will be notified by an event
https://www.youtube.com/watch?v=0LsaHiqK12A

In C#, you can create jagged array is an array of objects, where each object is another array. A buffer cannot contain objects. Also these individual arrays can be variable length. In order to put a jagged array into a buffer (i.e. one _flat array), we need to flatted out all the data into a memory buffer, but then have a mechanism for application code to view it as a jagged array.

I have a possible implementation that I'm working with: https://github.com/kasajian/SharedMemory/tree/JaggedArrayDev
There's unit-tests there and comments on how to use it.

I haven't created a pull request yet because I want to try it in a real-world application first, and see how it goes. There may be a few things I might want to change.

Do SharedBuffers work in Linux? Anyone successfully used it? As far as I understand it's based on named MemoryMaps, which are not supported in. NET Core under Linux.

Is there any alternative for shared memory with. NET Core under Linux?

Hello,

If possible, I would suggest changing the license to MIT or Apache 2.0.

A number of classes match existing framework class names, these should be changed.

Array -> SharedArray
Buffer -> SharedBuffer

i am sorry, i do miss something

There is no way to interrupt a method like CircularBuffer.Read before the timeout limit is exceeded. This is problem is you want to gracefully wait for exit of a worker thread, you must wait the timeout of Read method. This applies to all methods that accept timeouts. To support cancellation for Read method, GetNodeForReading method should be modified like this

protected unsafe virtual Node* GetNodeForReading(int timeout, CancellationToken cancellationToken)
{
    Node* ptr;
    while (true)
    {
        int readStart = _nodeHeader->ReadStart;
        ptr = this[readStart];
        if (readStart == _nodeHeader->WriteEnd)
        {
            int result = WaitHandle.WaitAny(new WaitHandle[] { cancellationToken.WaitHandle, DataExists }, timeout, false);

            if(result == WaitHandle.WaitTimeout || result == 0)
            {
                return null;
            }
        }
        else if (Interlocked.CompareExchange(ref _nodeHeader->ReadStart, ptr->Next, readStart) == readStart)
        {
            break;
        }
    }

same goes for other methods that are having wait handles.
    return ptr;
}

Allow all consumers of a circular buffer to retrieve all the data, i.e. the nodes are not writable again until all consumers have read the data.

One of the ideas around how to do this was to use an array of ReadStart pointers in the nodeheader rather than a single pointer. But probably also using a counter for the DoneReading would be needed...

In BufferWithLocks, any Write/Read memthod just call WriteWait/ReadWait to check that WriteWaitEvent/ReadWaitEvent is set, but not acqire locks, why?

Should I call AcquireWriteLock/AcquireReadLock in my userland code?

any plan to support MacOS？

Can I use the NuGet package in a closed source application without violating the open source MIT License terms?

thank you for providing this library. When trying to use, however, I run into a serious problem.
I want to use SharedMemory package to share data between two C# apps written in C# .Net 3.5 (unfortunately cannot upgrade). I wrote a simple test to see how it works

    private void runTest()
    {
        string testText = "someTest";   //"some", "someTest12"
        CreateOrOpenMappedFile(testText);
        string result = ReadMemoryMappedFile();
    }

    protected void CreateOrOpenMappedFile(string data)
    {
        char[] dataBuffer = data.ToCharArray();
        SharedMemory.BufferReadWrite buff = new SharedMemory.BufferReadWrite("sharedMemoryName", 4096);
        buff.Write(dataBuffer);
    }
    protected string ReadMemoryMappedFile()
    {
        char[] data = new char[10];         
        SharedMemory.BufferReadWrite buff = new SharedMemory.BufferReadWrite("sharedMemoryName");
        buff.Read(data);
        buff.Close();
        return new string(data);
    }

In runTest method I expected testText string to be the same as result, but what I was getting was a string that was clipped. For testText == "someTest" result was "some", for "some", result was "so", for "someTest12" it was "someT".
What could be happening here? Marek

The following field will not work correctly in FastStructure:

[MarshalAs(UnmanagedType.ByValArray, SizeConst = 8, ArraySubType = UnmanagedType.U1)]
public byte[] Contents;

Instead it needs to be in a LayoutKind.Explicit struct with a fixed byte array, e.g.

[FieldOffset(48)]
public fixed byte Contents[8];

However this will then not work correctly with Marshal.PtrToStructure either. Instead only the first byte will be copied into the structure.

Need to either implement checks in FastStructure static constructor so that users of the library are notified of issues ASAP, or find a work around.

If a consumer app attempts to open a Shared Memory File before the producer has created it, a FileNotFound Exception is thrown. A little digging didn't turn up any way to gracefully test/check for a MemoryMappedFile before trying to open it (or even better get some kind of file watcher or WaitEvent for it to show up).

So I used this function, but it feels like it belongs in the library somewhere:

    static Boolean TryOpenExisting(out SharedMemory.CircularBuffer buf, String name, TimeSpan ts)
    {
        System.Diagnostics.Stopwatch sw = new System.Diagnostics.Stopwatch();
        if (ts.TotalMilliseconds > 0) { sw.Start();  }

        do
        {
            try
            {
                buf = new SharedMemory.CircularBuffer(name);
                return true;
            }
            catch (System.IO.FileNotFoundException e)
            {
                buf = null;
            }
            catch (Exception e)
            {
                throw e;
            }
        } while (sw.Elapsed < ts);

        return (buf != null);
    }

Here's how I used it in the consumer code:

                if (consumer == null) {
                    TryOpenExisting(out consumer, name: "MySharedMemory", ts: TimeSpan.FromSeconds(5f));
                    if (consumer != null) { cbuf = new byte[consumer.NodeBufferSize]; }
                }
                if (consumer != null)
                { 
                    while ((consumer.Read<byte>(cbuf) > 0)) {
                        ........
                    }
                }

We've a potential case for changing the type of Length in SharedArray (in a custom implementation) from int to long. Do you forsee any potential issues with this apart from the obvious impact on memory use/availability?

The proper dispose pattern requires a GC.SupressFinalize to prevent the dispose to be called two times:

 public void Dispose()
 {
            Dispose(true);
            GC.SuppressFinalize(this);
 }

 protected virtual void Dispose(bool disposeManagedResources)
 {
            if (disposeManagedResources)
            {
                this.Close();
            }
 }

Modify Project files to include .NET Core 3.0 targets

Hello, I am getting a null reference exception within SharedMemory.RpcBuffer.ReadThreadV1(), when I call dispose on a MasterBuffer.

I am using SharedMemory to retrieve a port from a server process so that a client can connect to a service.

I would show you the call to dispose, but that would require pasting in a big blob of my code for it to make sense. If you need anything else let me know. What I can confirm though it that the master buffer is disposed first.

 public class SharedMemoryPortGetter : IPortGetter
    {
        RpcBuffer _rpcBuffer;
        public async Task<int> GetPort(string id)
        {
            if(_rpcBuffer == null)
            {
                _rpcBuffer = new RpcBuffer(id);
            }

            return (await _rpcBuffer.RemoteRequestAsync()).Data.FromBinary<int>();
        }

        private bool disposedValue = false; // To detect redundant calls

        protected virtual void Dispose(bool disposing)
        {
            if (!disposedValue)
            {
                if (disposing)
                {
                    _rpcBuffer?.Dispose();
                }

                disposedValue = true;
            }
        }

        public void Dispose()
        {
            Dispose(true);
        }
    }
public class SharedMemoryPortSender : IPortSender
    {
        RpcBuffer _rpcBuffer;
        
        public void Init(GetPort getPort, string id)
        {
            _rpcBuffer = new RpcBuffer(id, (msgId, payLoad) =>
            {
                return getPort().ToBinary();
            });
        }

        private bool disposedValue = false; 

        protected virtual void Dispose(bool disposing)
        {
            if (!disposedValue)
            {
                if (disposing)
                {
                    _rpcBuffer?.Dispose();
                }

                disposedValue = true;
            }
        }

        public void Dispose()
        {
            Dispose(true);
        }
    }

Along with the support for NetStandard support, proposed in another issue...

Could it be possible to have Span and ArraySegment semantics for reading/writing ?

I have a very simple implementation using RpcBuffer in F#.
But it keeps missing messages.

If I send messages between host and client in a loop, about a tenth of the time responses time-out.
I'm not sure exactly how to replicate the failures, they seem like heisenbugs.

This isn't an accurate repro, but it has the ratio of timeouts I'm seeing.

    static class Program
    {
        private static async Task Main(string[] args)
        {
            // Ensure a unique channel name
            var rpcName = $"rpc-{Guid.NewGuid()}";
            var rpcMaster = new RpcBuffer(rpcName, bufferNodeCount: 2);
            var input = Encoding.Unicode.GetBytes(new String('I', 1024));
            var output = Encoding.Unicode.GetBytes(new String('O', 1024));
            var rpcSlave = new RpcBuffer(rpcName,
                (_, payload) => Task.Run(async () =>
                {
                    await Task.Delay(100);
                    return output;
                })
            );

            async Task Run()
            {
                var result = await rpcMaster.RemoteRequestAsync(input, timeoutMs: 1000);
                if (!result.Success)
                {
                    Console.WriteLine("Timedout");
                }
                else
                {
                    var sresult = Encoding.Unicode.GetString(result.Data);
                    Console.WriteLine($"Received: {sresult.Substring(1, 5)}...");
                }
            }


            void LogIfFailed(Task task)
            {
                if(task.IsFaulted)
                    Console.WriteLine("Failed");
            }

            var eval = 
            Parallel.For(1, 20, 
                new ParallelOptions { MaxDegreeOfParallelism = 2 }, _ => LogIfFailed(Run()));
            
            Console.ReadLine();
        }

    }

I am new to shared memory, what example do I need to use to communicate with a driver that uses shared memory? Thanks in Advance!

Unity version: 5.5.2f1 (free version)
SharedMemory version: 2.0.16 (.NET 3.5 version)

Code: https://pastebin.com/qPtWjKPn

Problem: On launch gives an error:

FileNotFoundException: Unable to find the specified file.
System.IO.MemoryMappedFiles.MemoryMappedFile.OpenExisting (System.String mapName)
SharedMemory.SharedBuffer.Open ()

I don't know if it's a bug or if I'm missing something

Seems to work in this project which seems to have very similar SharedMemory base init code: https://bitbucket.org/vitaly_chashin/simpleunitybrowser/src/ad994940800930c03bafa14ca1f4b159789c4c77/UnityClient/Assets/SimpleWebBrowser/Scripts/BrowserEngine.cs?at=default&fileviewer=file-view-default#BrowserEngine.cs-434

Hey,

I'm trying to use the RPC buffer. It's injected in a process via easyhook with bootstrapped CLR.

This line

new RpcBuffer("Test");

Results in:

System.IO.IOException: Nicht ausreichend Speicher für die Zuordnung der Ansicht (Not enough memory to map the view) bei System.IO.MemoryMappedFiles.MemoryMappedView.CreateView(SafeMemoryMappedFileHandle memMappedFileHandle, MemoryMappedFileAccess access, Int64 offset, Int64 size) bei System.IO.MemoryMappedFiles.MemoryMappedFile.CreateViewAccessor(Int64 offset, Int64 size, MemoryMappedFileAccess access) bei SharedMemory.SharedBuffer.Open() bei SharedMemory.CircularBuffer..ctor(String name, Int32 nodeCount, Int32 nodeBufferSize) bei SharedMemory.RpcBuffer..ctor(String name, Int32 bufferCapacity, RpcProtocol protocolVersion, Int32 bufferNodeCount)

Any idea what could cause that problem?

I want to implement a Shared Memory Server which can always wait to new stuff to be written and print that, (To hold the server always Open and Write Whenever there is new Write in Mem)
and a client process which will open write and close.(It is basic).

I come to know that this Library has implemented it via CodeProject But I don't know how can i use it as there are no such examples.

[idf@localhost Net40]$ ^C
[idf@localhost Net40]$ ^C
[idf@localhost Net40]$ mono ClientTest.exe
Press to start client

Open existing shared memory circular buffer

Unhandled Exception:
System.NotImplementedException: The method or operation is not implemented.
at SharedMemory.SharedBuffer.Open () [0x00099] in <548fbd8c964c4334bf10f7061eb7a57e>:0
[ERROR] FATAL UNHANDLED EXCEPTION: System.NotImplementedException: The method or operation is not implemented.
at SharedMemory.SharedBuffer.Open () [0x00099] in <548fbd8c964c4334bf10f7061eb7a57e>:0
[idf@localhost Net40]$

Hi Justin,

Could you please give an example of how to use FastStructure in client/server applications? The examples on your site and in the FastStructureTests.cs aren't clear about that (besides, this is a new subject for me). In fact, I don't get how to pass IntPtr between the applications.

And some more questions if you don't mind.

1. Why do you use unsafe keyword in the test method FastStructure_AllocHGlobalReadWrite()?

        unsafe
        {
            n.Compatible.Contents[0] = 4;
            n.Compatible.Contents[7] = 5;
        }
   

2. Don't you need to set StructLayoutAttribute.Pack field when instantiating StructLayoutAttribute, as shown here? Or the default value Pack=0 is ok in this case?

3. Is it possible to use DateTime/TimeSpan structures (and nested structures) in the non-generic version of FastStructure, or I need to convert them to longs?

Thank you in advance!

Alex

Is there a way for a new master to reuse the rpcName of a master that has previously disposed ?

Currently I have to restart the slave if the master restarts.
I have a use case where the slave continuously runs, but the master process will restart.

I have a workaround, by way of using a normal file to communicate the new master rpcName to the slave, but it is a hack.

To support .NET 3.5, need a wrapper around native MemoryMappedFile as this was not introduced into the BCL until .NET 4.

exe a code:
String ipcName = "ipcName";
RpcBuffer ipcSlave = new RpcBuffer(ipcName, (msgId, payload) =>
{
log.Info("ipcSlave receive");
//how to know master timeout?
Thread.Sleep(10*1000);
log.Info("ipcSlave finish work");
return new byte[1];
});

exe b code:
String ipcName = "ipcName";
RpcBuffer ipcMaster = new RpcBuffer(ipcName);
for (int i = 0; i < 1000; i++)
{
ipcMaster.RemoteRequest(timeoutMs: 10);
}
log.Info("all request timeout!");

When program b times out, how does program a know and stop work ?

Add a .NET Core 2.0 target

I am researching integrating this library as the IPC mechanism for the NetMQ project (DotNet ZeroMQ project - also here on gh) and am looking for how to manage multiple readers and writers for that scenario.

What would be ideal is having each file participant (when there are multiple readers) be able to track which nodes they've read from the buffer independently (which I think means something like they each have their own separate read/write pointers looking at the same shared buffer data); copy the data as few times as is possible/necessary (again, using separate node lists, but referencing the same shared memory data); bi-directional communication (one memory file per "writer"?); and some way to easily find out if/when a participant has written data to the channel.

Right behind that would be discovering when a new Participant has connected to the channel. Is there some kind of eventing that could be used for this? I'd thought about using a MessageOnly Window Class and sending the Node Address Pointer (which exists in the shared memory region) as the message payload. That way all the readers would get a separate message to add/append the node address to the end of their queue. Then it only requires a scheme to know when the node could be deleted/reclaimed while dealing with aborted/blocked processes that will never process the message and/or read the node block...

I'm still working through the details, but ideas are welcome.

Thanks

private SharedMemory.SharedArray _buffer = _buffer = new SharedArray(@"Global\MARKETTICK", 200000); from within window service.

Try creating
SharedMemory.SharedArray _reader = new SharedArray(@"Global\MARKETTICK");

I Get "Access to the path 'Global\MARKETTICK_evt_write' is denied"

When I create the file without Global Prefix i.e.
private SharedMemory.SharedArray _buffer = _buffer = new SharedArray("MARKETTICK", 200000); from within window service.

from Console
SharedMemory.SharedArray _reader = new SharedArray(@"MARKETTICK");

I get file not found error.

Hi @spazzarama , first of all thanks for this excellent library. It's made my life easier while developing numerous IPC applications.

One of my current applications has a high-frequency data producer coupled with a lower-frequency data consumer. I would like the new data to overwrite older data regardless of whether it has been consumed. In other words, I need the latest data to be available at any given time.

How would I achieve this? I feel it may contradict #8 in some ways.

As a suggestion, in a non-IPC variant of the CircularBuffer, I had used a Boolean property to allow/disallow the producer overwriting the data.

I just tried examples from here in Unity and I had an exception when using CircularBuffer.

EntryPointNotFoundException: mono_win32_compat_CopyMemory

CircularBuffer is somewhere using CopyMemoryPtr method ([DllImport("kernel32.dll", EntryPoint = "CopyMemory", SetLastError = false)])

What helped me to fix this error: change EntryPoint = "CopyMemory" for EntryPoint = "RtlCopyMemory"

If a class inherting from BufferWithLocks does not correctly release Read/Write locks, or does not correctly check that the AcquireReadLock/AcquireWriteLock has returned true, it is possible to end up with a deadlock situation where a subsequent Read/Write call will wait indefinitely.

As a safety measure, a ReadWriteTimeout property (defaulting to 100ms) will be added to BufferWithLocks, and all Read/Write operations will timeout with a TimedoutException if the WaitEvent is not set within that timeframe.

If using AcquireReadLock/ReleaseReadLock and AcquireWriteLock/ReleaseWriteLock correctly this timeout will never occur.

Hi, I'm trying create SharedArray inside Windows Service and open it in desktop app and I am getting the Unable to find the specified file exception. It is somehow possible to resolve this task and share data between Windows Service and desktop app?

Supporting SharedMemory.SharedDictionary<string, string> or even SharedMemory.SharedDictionary<string, object> could be great, it can be a flexible KV server.

Add a simple RPC implementation using the existing CircularBuffer class.

I start two processes, one writes data continuously, and another reads data, when i kill the write process and restart, it throws an exception "Cannot create a file when it already exists" , so i look into the library source code, change the "MemoryMappedFile.CreateNew" function in SharedBuffer class to "MemoryMappedFile.CreateOrOpen", the bug doesn't appear and worked very well.

Hi all,

I just posted the following Gist as an initial cut on making dynamic shared memory.
https://gist.github.com/MikeFair/6908afe26e0b84758ca4

The main idea is that (bufferPosition / blockSize) gives a unique segmentId.
"BufferName - [segmentId]" is then the file name of that segment which participants can test for the existence of and load them if they exist. if they do not exist, then participants can create them as they write data into those memory addressses.

There is no signalling happening about new segments; but attempts to read any segments that exist will be dynamically discovered.

Thanks!

What a wonderful tool!
My only suggestions would be to implement Performance statistics on the main (README.md) page,
they really help asses the effectiveness of this otherwise perfect library!

Hi,
I'm referencing the package from a netstandard2.0 project, and VS displays this warning:

warning NU1701: Package 'SharedMemory 2.1.0' was restored using '.NETFramework,Version=v4.6.1' instead of the project target framework '.NETStandard,Version=v2.0'. This package may not be fully compatible with your project.

Is this expected? Are there any risks in ignoring this warning?

Edit: I guess the library projects should target netstandard2.0 instead of netcoreapp2.0.

Brgds,
Robert

I am looking for advice how to implement the following functionality with SharedMemory

I have N number of processes and I would like any of those N processes to broadcast data to other N-1 processes.
I looked at the code and it seems this is not supported out of the box. I am looking how to implement that utilizing SharedMemory classes.
One possible solution is every process to have CircularBuffer with unique name. On start, every process will register the name of his Circular buffer on my own implementation of Memory Mapped file that will act as registry for the active processes. When a process need to broadcast data, he will first read the memory mapped file with the regirations and they will construct many circular buffers using the registration names and then will send the data to each buffer.

Is there easier way to do that using SharedMemory?

If the Shared Memory is running in a HighPerformance like it's called 50 times+ per second quite a lot allocations are going on to avoid this every allocation should be done once and then reused, if you want to i can take care of this and try to replace allocations to one allocation at as much places as possible and increase the speed of this for HighPerformance. Maybe it then makes sense to use Array Segments because they can give access to a part of a array and can already save a lot allocations