Add Unix socket support (#944)

* Add Unix socket support

* More cluster test adjustments

* Remove warnings

* dotnet format whitespace

* fix merge

* fix merge

* Add very basic test for the unix socket support

* Try fix long-path error & sync GarnetClientSession connection logic with GarnetClient

* Dispose the GarnetClient objects in tests

* Sync LightClient connection logic with GarnetClient(Session)

* fix

* Add exception message to StoreWrapper.GetIp

* Fix merge

* Add TLS test path

* Add more Unix socket tests with SE.Redis as client

* Add config parsing for --unixsocketperm and tests

* attempt fix import ordering

* Set unix socket permission after listener is bound

* attempt fix import ordering

* attempt fix import ordering

* attempt fix import ordering

* use ILogger in LightClient

---------

Co-authored-by: Vasileios Zois <[email protected]>

benchmark/BDN.benchmark/Cluster/ClusterContext.cs

@@ -1,6 +1,7 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
+using System.Net;
 using System.Runtime.InteropServices;
 using System.Text;
 using BDN.benchmark.CustomProcs;
@@ -34,7 +35,7 @@ public void SetupSingleInstance(bool disableSlotVerification = false)
             {
                 QuietMode = true,
                 EnableCluster = !disableSlotVerification,
-                Port = port,
+                EndPoint = new IPEndPoint(IPAddress.Loopback, port),
                 CleanClusterConfig = true,
             };
             if (RuntimeInformation.IsOSPlatform(OSPlatform.Linux))

benchmark/BDN.benchmark/Embedded/GarnetServerEmbedded.cs

@@ -2,6 +2,7 @@
 // Licensed under the MIT license.
 
 using System;
+using System.Net;
 using System.Net.Security;
 using System.Threading;
 using Garnet.common;
@@ -13,7 +14,7 @@ namespace Embedded.server
 {
     internal class GarnetServerEmbedded : GarnetServerBase, IServerHook
     {
-        public GarnetServerEmbedded() : base("0.0.0.0", 0, 1 << 10)
+        public GarnetServerEmbedded() : base(new IPEndPoint(IPAddress.Loopback, 0), 1 << 10)
         {
         }
 
@@ -36,7 +37,7 @@ public EmbeddedNetworkHandler CreateNetworkHandler(SslClientAuthenticationOption
                             throw new Exception("Unable to add handler to dictionary");
 
                         handler.Start(tlsOptions, remoteEndpointName);
-                        incr_conn_recv();
+                        IncrementConnectionsReceived();
                         return handler;
                     }
                     catch (Exception ex)
@@ -59,7 +60,7 @@ public void DisposeMessageConsumer(INetworkHandler session)
             if (activeHandlers.TryRemove(session, out _))
             {
                 Interlocked.Decrement(ref activeHandlerCount);
-                incr_conn_disp();
+                IncrementConnectionsDisposed();
                 try
                 {
                     session.Session?.Dispose();

benchmark/Resp.benchmark/Program.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Net;
 using System.Threading;
 using CommandLine;
 using Garnet.client;
@@ -195,7 +196,7 @@ static void Main(string[] args)
 
         static void WaitForServer(Options opts)
         {
-            using var client = new GarnetClientSession(opts.Address, opts.Port, new(), tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
+            using var client = new GarnetClientSession(new IPEndPoint(IPAddress.Parse(opts.Address), opts.Port), new(), tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
             while (true)
             {
                 try
@@ -228,7 +229,7 @@ static void RunBasicCommandsBenchmark(Options opts)
                 unsafe
                 {
                     var onResponseDelegate = new LightClient.OnResponseDelegateUnsafe(ReqGen.OnResponse);
-                    using var client = new LightClient(opts.Address, opts.Port, (int)OpType.GET, onResponseDelegate, opts.DbSize, opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
+                    using var client = new LightClient(new IPEndPoint(IPAddress.Parse(opts.Address), opts.Port), (int)OpType.GET, onResponseDelegate, opts.DbSize, opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
                     client.Connect();
                     client.Authenticate(opts.Auth);
                     BenchUtils.LoadSetGetScripts(client, out BenchUtils.sha1SetScript, out BenchUtils.sha1GetScript);

benchmark/Resp.benchmark/RespOnlineBench.cs

@@ -5,6 +5,7 @@
 using System.Buffers;
 using System.Diagnostics;
 using System.Linq;
+using System.Net;
 using System.Runtime.CompilerServices;
 using System.Threading;
 using System.Threading.Tasks;
@@ -33,8 +34,7 @@ internal class RespOnlineBench
         static bool IsValidRange(long value)
             => value < HISTOGRAM_UPPER_BOUND && value > HISTOGRAM_LOWER_BOUND;
 
-        readonly string address;
-        readonly int port;
+        readonly EndPoint endpoint;
         readonly int NumThreads;
         readonly OpType op;
         readonly Options opts;
@@ -71,8 +71,7 @@ public RespOnlineBench(Options opts, int resetInterval = 30, int runDuration = i
         {
             this.runDuration = runDuration;
             this.resetInterval = resetInterval;
-            this.address = opts.Address;
-            this.port = opts.Port;
+            this.endpoint = new IPEndPoint(IPAddress.Parse(opts.Address), opts.Port);
             this.op = opts.Op;
             this.opts = opts;
             this.auth = opts.Auth;
@@ -157,7 +156,7 @@ private void InitializeClients()
             {
                 gcsPool = new AsyncPool<GarnetClientSession>(opts.NumThreads.First(), () =>
                 {
-                    var c = new GarnetClientSession(address, port, new(), tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
+                    var c = new GarnetClientSession(endpoint, new(), tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
                     c.Connect();
                     if (auth != null)
                     {
@@ -173,7 +172,7 @@ private void InitializeClients()
                 {
                     gdbPool = new AsyncPool<GarnetClient>(opts.NumThreads.First(), () =>
                     {
-                        var gdb = new GarnetClient(address, port, opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null, recordLatency: opts.ClientHistogram);
+                        var gdb = new GarnetClient(endpoint, opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null, recordLatency: opts.ClientHistogram);
                         gdb.Connect();
                         if (auth != null)
                         {
@@ -184,7 +183,7 @@ private void InitializeClients()
                 }
                 else
                 {
-                    garnetClient = new GarnetClient(address, port, opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null, recordLatency: opts.ClientHistogram);
+                    garnetClient = new GarnetClient(endpoint, opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null, recordLatency: opts.ClientHistogram);
                     garnetClient.Connect();
                     if (auth != null)
                     {
@@ -427,7 +426,7 @@ public unsafe void OpRunnerLightClient(int thread_id)
 
             var onResponseDelegate = new LightClient.OnResponseDelegateUnsafe(ReqGen.OnResponse);
 
-            var client = new LightClient(address, port, (int)op, onResponseDelegate, size, opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
+            var client = new LightClient(endpoint, (int)op, onResponseDelegate, size, opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
             client.Connect();
             client.Authenticate(auth);
 
@@ -571,8 +570,7 @@ public async void OpRunnerGarnetClientSession(int thread_id)
             if (!opts.Pool)
             {
                 client = new GarnetClientSession(
-                    address,
-                    port,
+                    endpoint,
                     new(Math.Max(bufferSizeValue, opts.ValueLength * opts.IntraThreadParallelism)),
                     tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
                 client.Connect();
@@ -671,8 +669,7 @@ public async void OpRunnerGarnetClientSessionParallel(int thread_id, int paralle
             if (!opts.Pool)
             {
                 client = new GarnetClientSession(
-                    address,
-                    port,
+                    endpoint,
                     new NetworkBufferSettings(Math.Max(131072, opts.IntraThreadParallelism * opts.ValueLength)),
                     tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
                 client.Connect();

benchmark/Resp.benchmark/RespPerfBench.cs

@@ -5,6 +5,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
+using System.Net;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
@@ -86,7 +87,7 @@ private unsafe void GetDBSIZE(int loadDbThreads)
             {
                 var req = Encoding.ASCII.GetBytes("*1\r\n$6\r\nDBSIZE\r\n");
                 var lighClientOnResponseDelegate = new LightClient.OnResponseDelegateUnsafe(ReqGen.OnResponse);
-                using LightClient client = new(opts.Address, opts.Port, (int)OpType.DBSIZE, lighClientOnResponseDelegate, 128, opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
+                using LightClient client = new(new IPEndPoint(IPAddress.Parse(opts.Address), opts.Port), (int)OpType.DBSIZE, lighClientOnResponseDelegate, 128, opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
 
                 client.Connect();
                 client.Authenticate(opts.Auth);
@@ -373,7 +374,7 @@ public ReqGen LightOperate(
         private unsafe void LightOperateThreadRunner(int NumOps, OpType opType, ReqGen rg)
         {
             var lighClientOnResponseDelegate = new LightClient.OnResponseDelegateUnsafe(ReqGen.OnResponse);
-            using ClientBase client = new LightClient(opts.Address, opts.Port, (int)opType, lighClientOnResponseDelegate, rg.GetBufferSize(), opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
+            using ClientBase client = new LightClient(new IPEndPoint(IPAddress.Parse(opts.Address), opts.Port), (int)opType, lighClientOnResponseDelegate, rg.GetBufferSize(), opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
 
             client.Connect();
             client.Authenticate(opts.Auth);
@@ -407,7 +408,7 @@ private void GarnetClientSessionOperateThreadRunner(int NumOps, OpType opType, R
                 default:
                     throw new Exception($"opType: {opType} benchmark not supported with GarnetClientSession!");
             }
-            var c = new GarnetClientSession(opts.Address, opts.Port, new(), tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
+            var c = new GarnetClientSession(new IPEndPoint(IPAddress.Parse(opts.Address), opts.Port), new(), tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
             c.Connect();
             if (opts.Auth != null)
             {

benchmark/Resp.benchmark/TxnPerfBench.cs

@@ -4,6 +4,7 @@
 using System;
 using System.Diagnostics;
 using System.Linq;
+using System.Net;
 using System.Threading;
 using Garnet.client;
 using Garnet.common;
@@ -107,7 +108,7 @@ public void Run()
             {
                 gcsPool = new AsyncPool<GarnetClientSession>(opts.NumThreads.First(), () =>
                 {
-                    var c = new GarnetClientSession(address, port, new(), tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
+                    var c = new GarnetClientSession(new IPEndPoint(IPAddress.Parse(address), port), new(), tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
                     c.Connect();
                     if (auth != null)
                     {
@@ -325,7 +326,7 @@ public void OpRunnerSERedis(int thread_id)
         public void LoadData()
         {
             var req = new OnlineReqGen(0, opts.DbSize, true, opts.Zipf, opts.KeyLength, opts.ValueLength);
-            GarnetClientSession client = new(address, port, new(), tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
+            GarnetClientSession client = new(new IPEndPoint(IPAddress.Parse(address), port), new(), tlsOptions: opts.EnableTLS ? BenchUtils.GetTlsOptions(opts.TlsHost, opts.CertFileName, opts.CertPassword) : null);
             client.Connect();
             if (auth != null)
             {