cloudtoid / interprocess Goto Github PK

You're using GlobalSetup/Cleanup instead of IterationSetup/Cleanup to create the Publisher, so after the first few runs TryEnqueue just starts returning false without doing any work.

Once you fix that, you'll start getting warnings that the op count is too low to get legitimately valid results.

Once everything is fixed, you'll probably get something like this, if you use a message size of 128, queue capacity of 33554432 bytes, and 246723 operations per iteration. (e.g. [Benchmark(Description = "Enqueue messages", OperationsPerInvoke = 246723)] and for (var i = 0; i < 246723; ++i))

Of course, per machine, YMMV...

|                                                    Method |     Mean |    Error |   StdDev |  Gen 0 | Allocated |
|---------------------------------------------------------- |---------:|---------:|---------:|-------:|----------:|
|                                        'Enqueue messages' | 543.8 ns |  4.19 ns |  3.50 ns |      - |         - |
|                            'Enqueue messages (zero-copy)' | 547.2 ns |  2.78 ns |  2.32 ns |      - |         - |
|             'Enqueue messages (zero-copy, func pointers)' | 557.0 ns |  3.71 ns |  3.29 ns |      - |         - |
|      'Enqueue and dequeue messages (buffered allocating)' | 724.7 ns |  5.08 ns |  4.50 ns | 0.0284 |     152 B |
|             'Enqueue and dequeue messages (buffer reuse)' | 712.3 ns |  6.32 ns |  5.27 ns |      - |         - |
|                'Enqueue and dequeue messages (zero-copy)' | 709.4 ns | 13.67 ns | 15.20 ns |      - |         - |
| 'Enqueue and dequeue messages (zero-copy, func pointers)' | 718.9 ns |  7.67 ns |  6.80 ns |      - |         - |

You don't have the zero-copy ops yet, but you can review a future PR at https://github.com/StirlingLabs/interprocess

It's still quite fast, but 7ns/op is 28 CPU cycles/op at 4ghz (between 7 and 36 GP instructions), which is a dead giveaway that no work was being done; it's literally impossibly fast, it has to just be checking a value and returning false. Putting the benchmark into debug mode (BenchmarkRunner.Run(typeof(Program).Assembly, new DebugInProcessConfig());) and stepping into it confirmed it.

If the 2 processes are running as the same user, there is no problem with security. But if you want 2 processes with different users to communicate, you will get "access denied".

In .NET Framework this was possible using MemoryMappedFileSecurity - but, bizarelly, Microsoft have chosen not to add support for this in .NET Core/.NET 5/6.

The request here is to add the interop plumbing to allow setting an ACL for MMFs (I did something similar a while back, when Microsoft made the same bizare decision for named pipes; not sure if they ever did implement ACLs there).

Hi, cloudtoid

in publisher code, message length needs to be small than capacity, if tailoffset is great than long.MaxLength, it will SafeIncrement to a negative value, so this line here should check if TailOffset < HeadOffset

great work by the work

Hi,

Can you please help me with that error. I just created a sample on TFM .NET 5.0 and get that exception on Ubuntu WSL (WIndows Subsystem for Linux) on Window 10.

Sample: https://[email protected]/zone-eric/NET%20Core%20Crossplattform%20IPC/_git/NETCoreXplatIPC

Description: Ubuntu 20.04.1 LTS
Release: 20.04
Codename: focal

System.DllNotFoundException: Unable to load shared library 'librt' or one of its dependencies. In order to help diagnose loading problems, consider setting the LD_DEBUG environment variable: liblibrt: cannot open shared object file: No such file or directory
at Cloudtoid.Interprocess.Semaphore.Linux.Interop.sem_open(String name, Int32 oflag, UInt32 mode, UInt32 value)
at Cloudtoid.Interprocess.Semaphore.Linux.Interop.CreateOrOpenSemaphore(String name, UInt32 initialCount)
at Cloudtoid.Interprocess.Semaphore.Linux.SemaphoreLinux..ctor(String name, Boolean deleteOnDispose)
at Cloudtoid.Interprocess.InterprocessSemaphore.CreateWaiter(String name)
at Cloudtoid.Interprocess.Subscriber..ctor(QueueOptions options, ILoggerFactory loggerFactory)
at Cloudtoid.Interprocess.QueueFactory.CreateSubscriber(QueueOptions options)
at Sunify.WorkerService.Worker.ExecuteAsync(CancellationToken stoppingToken) in /home/eric/source/repos/NETCoreXplatIPC/xPlat-IPC/Sunify.WorkerService/Worker.cs:line 26
at Microsoft.Extensions.Hosting.Internal.Host.StartAsync(CancellationToken cancellationToken)
at Microsoft.Extensions.Hosting.HostingAbstractionsHostExtensions.RunAsync(IHost host, CancellationToken token)
at Microsoft.Extensions.Hosting.HostingAbstractionsHostExtensions.RunAsync(IHost host, CancellationToken token)
at Microsoft.Extensions.Hosting.HostingAbstractionsHostExtensions.Run(IHost host)
at Sunify.WorkerService.Program.Main(String[] args) in /home/eric/source/repos/NETCoreXplatIPC/xPlat-IPC/Sunify.WorkerService/Program.cs:line 15

Full command line log is here: https://gist.github.com/ericbrunner/cd2122fad75920cdb758735cce0bb220

Problem description

At first, I import Cloudtoid.Interprocess to my ASP.NET WEB API backend application, every time it run using var publisher = factory.CreatePublisher(options); it got into the exception below:

Unhandled exception. System.ArgumentException: The initial count cannot be greater than 32767. (Parameter 'initialCount')
   at Cloudtoid.Interprocess.Semaphore.MacOS.Interop.CreateOrOpenSemaphore(String name, UInt32 initialCount)
   at Cloudtoid.Interprocess.Semaphore.MacOS.SemaphoreMacOS..ctor(String name, Boolean deleteOnDispose)
   at Cloudtoid.Interprocess.InterprocessSemaphore.CreateReleaser(String name)
   at Cloudtoid.Interprocess.Publisher..ctor(QueueOptions options, ILoggerFactory loggerFactory)
   at Cloudtoid.Interprocess.QueueFactory.CreatePublisher(QueueOptions options)
   ...

I move the code into a vary simple project which just contains a single Program.cs:

using Cloudtoid.Interprocess;
using Microsoft.Extensions.Logging;

var factory = new QueueFactory();
var options = new QueueOptions(
    queueName: "sample-queue", bytesCapacity: 24);// just 24
using var publisher = factory.CreatePublisher(options);

It got into the same exception.

Reproduction

I pushed that simple project to this GitHub repository

Additional information

Cloudtoid.Interprocess version 1.0.175

My OS information:

# uname -a
Darwin hostname.lan 22.4.0 Darwin Kernel Version 22.4.0: Mon Mar  6 20:59:28 PST 2023; root:xnu-8796.101.5~3/RELEASE_ARM64_T6000 arm64

My dotnet information:

# dotnet --info
.NET SDK:
 Version:   7.0.203
 Commit:    5b005c19f5

Runtime Environment:
 OS Name:     Mac OS X
 OS Version:  13.3
 OS Platform: Darwin
 RID:         osx.13-arm64
 Base Path:   /usr/local/share/dotnet/sdk/7.0.203/

Host:
  Version:      7.0.5
  Architecture: arm64
  Commit:       8042d61b17

.NET SDKs installed:
  7.0.203 [/usr/local/share/dotnet/sdk]

.NET runtimes installed:
  Microsoft.AspNetCore.App 7.0.5 [/usr/local/share/dotnet/shared/Microsoft.AspNetCore.App]
  Microsoft.NETCore.App 7.0.5 [/usr/local/share/dotnet/shared/Microsoft.NETCore.App]

Other architectures found:
  None

Environment variables:
  DOTNET_ROOT       [/opt/homebrew/opt/dotnet/libexec]

global.json file:
  Not found

Learn more:
  https://aka.ms/dotnet/info

Download .NET:
  https://aka.ms/dotnet/download

Hi! I have a working monitoring process using this great library. Today I decided that I was going to add a self-hosted API through Kestrel...something very simple and something I have done many times before. However, I am finding a couple of really odd issues: First, the processing works fine with these additions but Kestrel just returns no data. The controllers are getting called but no data is returned. So I pulled my hair figuring it all out and ended up commenting out everything but the WebHostBuilder configuration. When I remove this nuget package, and that is the only change, the controller behaves perfectly fine - returns data. Simply adding this package back into the project and the weird behavior begins.

Any ideas?

Update: I'm actually getting 504s in Fiddler. That's a gateway timeout - this is really weird to me. I appreciate any insight.

I was trying to figure out a way to make this even faster and found that I am leaking named semaphores on Linux. I'll work on a fix but one is not coming to mind immediately.

Hi
On windows, there is System.UnauthorizedAccessException: 'Access to the path is denied.' if the publisher and subscriber are created by different accounts in different processes.
for example, a windows service process create a subscriber, later, when an user application try to create the publisher, the exception will be thrown. Can we create the memory mapped file that can be accessed by all everyone?

Thanks

I test the sample publisher and subscriber on windows10 OS, then find that it doesn't support multiple subscribers. I start a publisher process and two same subscriber processes (called A and B), but subscriber A receive different data compared with subscriber B, so maybe this is a bug. Some screenshots are as follows.

under window 10 C# net6
version: 1.0.169

Created a simple publisher/subscriber test and working perfectly for 1P1C mode (P-producer, C- consumer)
however, when trying 1P2C mode, get the following exception after a short while in one of consumers

Unhandled exception. System.InvalidOperationException: This is unexpected and can be a serious bug. We take a lock on this message prior to this point which should ensure that the HeadOffset is left unchanged.
   at Cloudtoid.Interprocess.Subscriber.TryDequeueImpl(Nullable`1 resultBuffer, CancellationToken cancellation, ReadOnlyMemory`1& message)
   at Cloudtoid.Interprocess.Subscriber.TryDequeueCore(Nullable`1 resultBuffer, CancellationToken cancellation, ReadOnlyMemory`1& message)
   at Cloudtoid.Interprocess.Subscriber.TryDequeue(Memory`1 resultBuffer, CancellationToken cancellation, ReadOnlyMemory`1& message)

Any comments on that?

Hi, thanks for this nice library! 🙏

I was playing with the sample, and noticed that when the Subscriber starts, it reads back any existing queued messages first. What I'd like is for the Subscriber to only read new messages from that point onwards. Would that be possible?

Wanna use it inside Unity. Plz support netstandard compilation.