Fix multi-tread creation session (#139)

* Fix multi-tread creation session

When the session is created in multi-thread way.
The fix adds _semaphoreSlim.

closes: #138

Signed-off-by: Gabriele Santomaggio <[email protected]>

* Add tests

Signed-off-by: Gabriele Santomaggio <[email protected]>

---------

Signed-off-by: Gabriele Santomaggio <[email protected]>

Author: Gsantomaggio

RabbitMQ.AMQP.Client/Impl/AmqpSessionManagement.cs

@@ -5,6 +5,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Amqp;
 using Amqp.Framing;
@@ -15,8 +16,12 @@ internal class AmqpSessionManagement
     {
         private readonly AmqpConnection _amqpConnection;
         private readonly int _maxSessionsPerItem;
+
         private readonly ConcurrentBag<Session> _sessions = new();
 
+        // lock during session creation
+        private readonly SemaphoreSlim _semaphoreSlim = new(1);
+
         internal AmqpSessionManagement(AmqpConnection amqpConnection, int maxSessionsPerItem)
         {
             _amqpConnection = amqpConnection;
@@ -26,33 +31,65 @@ internal AmqpSessionManagement(AmqpConnection amqpConnection, int maxSessionsPer
         // TODO cancellation token
         internal async Task<Session> GetOrCreateSessionAsync()
         {
-            Session rv;
-
-            if (_sessions.Count >= _maxSessionsPerItem)
+            await _semaphoreSlim.WaitAsync(TimeSpan.FromSeconds(1)).ConfigureAwait(false);
+            try
             {
-                rv = _sessions.First();
-            }
-            else
-            {
-                TaskCompletionSource<ISession> sessionBeginTcs = Utils.CreateTaskCompletionSource<ISession>();
-                void OnBegin(ISession session, Begin peerBegin)
+                Session rv;
+
+                if (_sessions.Count >= _maxSessionsPerItem)
                 {
-                    sessionBeginTcs.SetResult(session);
+                    rv = _sessions.First();
                 }
+                else
+                {
+                    TaskCompletionSource<ISession> sessionBeginTcs = Utils.CreateTaskCompletionSource<ISession>();
 
-                rv = new Session(_amqpConnection.NativeConnection, GetDefaultBegin(), OnBegin);
-                // TODO cancellation token
-                ISession awaitedSession = await sessionBeginTcs.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
-                _sessions.Add(rv);
-            }
+                    void OnBegin(ISession session, Begin peerBegin)
+                    {
+                        sessionBeginTcs.SetResult(session);
+                    }
+
+                    rv = new Session(_amqpConnection.NativeConnection, GetDefaultBegin(), OnBegin);
+                    // TODO cancellation token
+                    ISession awaitedSession =
+                        await sessionBeginTcs.Task.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
+                    _sessions.Add(rv);
+                }
 
-            return rv;
+                return rv;
+            }
+            finally
+            {
+                _semaphoreSlim.Release();
+            }
         }
 
         internal void ClearSessions()
         {
-            // TODO close open sessions?
-            _sessions.Clear();
+            _semaphoreSlim.Wait();
+            try
+            {
+                // TODO close open sessions?
+                _sessions.Clear();
+            }
+            finally
+            {
+                _semaphoreSlim.Release();
+            }
+        }
+
+        // sessions count 
+        internal int GetSessionsCount()
+        {
+            _semaphoreSlim.Wait();
+            try
+            {
+                return _sessions.Count;
+            }
+            finally
+            {
+                _semaphoreSlim.Release();
+            }
         }
 
         // Note: these values come from Amqp.NET

Tests/Sessions/SessionsTests.cs

@@ -0,0 +1,55 @@
+// This source code is dual-licensed under the Apache License, version 2.0,
+// and the Mozilla Public License, version 2.0.
+// Copyright (c) 2017-2024 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using RabbitMQ.AMQP.Client.Impl;
+using Xunit;
+using Xunit.Abstractions;
+
+namespace Tests.Sessions
+{
+    public class SessionsTests(ITestOutputHelper testOutputHelper) : IntegrationTest(testOutputHelper)
+    {
+        [Theory]
+        [InlineData(1)]
+        [InlineData(10)]
+        [InlineData(37)]
+        public async Task CreateSessionSequentiallyShouldHaveTheMaxSessionsPerItem(int maxSessions)
+        {
+            Assert.NotNull(_connection);
+            AmqpConnection? amqpConnection = _connection as AmqpConnection;
+            Assert.NotNull(amqpConnection);
+            AmqpSessionManagement s = new(amqpConnection, maxSessions);
+            for (int i = 0; i < 100; i++)
+            {
+                await s.GetOrCreateSessionAsync();
+            }
+
+            Assert.Equal(maxSessions, s.GetSessionsCount());
+        }
+
+        [Theory]
+        [InlineData(1)]
+        [InlineData(10)]
+        [InlineData(37)]
+        public async Task CreateSessionInMultiThreadingShouldHaveTheMaxSessionsPerItem(int maxSessions)
+        {
+            Assert.NotNull(_connection);
+            AmqpConnection? amqpConnection = _connection as AmqpConnection;
+            Assert.NotNull(amqpConnection);
+            AmqpSessionManagement s = new(amqpConnection, maxSessions);
+            Task[] tasks = new Task[100];
+            for (int i = 0; i < 100; i++)
+            {
+                tasks[i] = (Task.Run(async () => { await s.GetOrCreateSessionAsync(); }));
+            }
+
+            await Task.WhenAll(tasks);
+
+            Assert.Equal(maxSessions, s.GetSessionsCount());
+        }
+    }
+}