Consecutive & interlaced session test, async compatible concurrent test. To be squashed.

Author: fredericDelaporte

src/NHibernate.Test/Async/NHSpecificTest/Logs/LogsFixture.cs

@@ -12,9 +12,13 @@
 using System.Collections.Concurrent;
 using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
+using NHibernate.Cfg;
 using NHibernate.Impl;
-
+using NHibernate.SqlCommand;
+using NHibernate.Type;
 using NUnit.Framework;
+using NHibernate.Linq;
 
 namespace NHibernate.Test.NHSpecificTest.Logs
 {
@@ -26,7 +30,6 @@ namespace NHibernate.Test.NHSpecificTest.Logs
 	using log4net.Core;
 	using log4net.Layout;
 	using log4net.Repository.Hierarchy;
-	using System.Threading.Tasks;
 
 	[TestFixture]
 	public class LogsFixtureAsync : TestCase
@@ -41,6 +44,33 @@ protected override string MappingsAssembly
 			get { return "NHibernate.Test"; }
 		}
 
+		protected override void Configure(Configuration configuration)
+		{
+			base.Configure(configuration);
+			configuration.SetProperty(Cfg.Environment.UseSecondLevelCache, "false");
+		}
+
+		protected override void OnSetUp()
+		{
+			using (var s = Sfi.OpenSession())
+			using (var t = s.BeginTransaction())
+			{
+				s.Save(new Person());
+				s.Save(new Person());
+				t.Commit();
+			}
+		}
+
+		protected override void OnTearDown()
+		{
+			using (var s = Sfi.OpenSession())
+			using (var t = s.BeginTransaction())
+			{
+				s.CreateQuery("delete from Person").ExecuteUpdate();
+				t.Commit();
+			}
+		}
+
 		[Test]
 		public async Task WillGetSessionIdFromSessionLogsAsync()
 		{
@@ -49,27 +79,92 @@ public async Task WillGetSessionIdFromSessionLogsAsync()
 			using (var spy = new TextLogSpy("NHibernate.SQL", "%message | SessionId: %property{sessionId}"))
 			using (var s = Sfi.OpenSession())
 			{
-				var sessionId = ((SessionImpl)s).SessionId;
+				var sessionId = ((SessionImpl) s).SessionId;
 
-				await (s.GetAsync<Person>(1));//will execute some sql
+				await (s.GetAsync<Person>(1)); //will execute some sql
 
 				var loggingEvent = spy.GetWholeLog();
 				Assert.That(loggingEvent.Contains(sessionId.ToString()), Is.True);
 			}
 		}
 
+		[Test]
+		public async Task WillGetSessionIdFromConsecutiveSessionsLogsAsync()
+		{
+			GlobalContext.Properties["sessionId"] = new SessionIdCapturer();
+
+			using (var spy = new TextLogSpy("NHibernate.SQL", "%message | SessionId: %property{sessionId}"))
+			{
+				var sessions = Enumerable.Range(1, 10).Select(i => Sfi.OpenSession()).ToArray();
+				try
+				{
+					for (var i = 0; i < 10; i++)
+					for (var j = 0; j < 10; j++)
+					{
+						var s = sessions[j];
+						await (s.GetAsync<Person>(i * 10 + j)); //will execute some sql
+					}
+				}
+				finally
+				{
+					foreach (var s in sessions)
+					{
+						s.Dispose();
+					}
+				}
+
+				var loggingEvent = spy.GetWholeLog();
+				for (var i = 0; i < 10; i++)
+				for (var j = 0; j < 10; j++)
+				{
+					var sessionId = sessions[j].GetSessionImplementation().SessionId;
+					Assert.That(loggingEvent, Does.Contain($"p0 = {i * 10 + j} [Type: Int32 (0:0:0)] | SessionId: {sessionId}"));
+				}
+			}
+		}
+
+		[Test]
+		public async Task WillGetSessionIdFromInterlacedSessionsLogsAsync()
+		{
+			GlobalContext.Properties["sessionId"] = new SessionIdCapturer();
+			var interceptor = new InterlacedSessionInterceptor(Sfi);
+			using (var spy = new TextLogSpy("NHibernate.SQL", "%message | SessionId: %property{sessionId}"))
+			using (var s = Sfi.WithOptions().Interceptor(interceptor).OpenSession())
+			{
+				// Trigger an operation which will fire many interceptor events, before and after s own logging.
+				var persons = await (s.Query<Person>().ToListAsync());
+
+				var loggingEvent = spy.GetWholeLog();
+				for (var i = 0; i < interceptor.SessionIds.Count; i++)
+				{
+					var sessionId = interceptor.SessionIds[i];
+					Assert.That(loggingEvent, Does.Contain($"p0 = {i + 1} [Type: Int32 (0:0:0)] | SessionId: {sessionId}"));
+				}
+				Assert.That(loggingEvent, Does.Contain($"Person person0_ | SessionId: {s.GetSessionImplementation().SessionId}"));
+			}
+		}
+
 		[Test]
 		public async Task WillGetSessionIdFromSessionLogsConcurrentAsync()
 		{
 			GlobalContext.Properties["sessionId"] = new SessionIdCapturer();
 
-			var semaphore = new ManualResetEventSlim();
+			// Do not use a ManualResetEventSlim, it does not support async and exhausts the task thread pool in the
+			// async counterparts of this test. SemaphoreSlim has the async support and release the thread when waiting.
+			var semaphore = new SemaphoreSlim(0);
 			var failures = new ConcurrentBag<Exception>();
 			var sessionIds = new ConcurrentDictionary<int, Guid>();
-			var array = Enumerable.Range(1, 10).Select(
-				i => new Thread(
-					() =>
+			using (var spy = new TextLogSpy("NHibernate.SQL", "%message | SessionId: %property{sessionId}"))
+			{
+				await (Task.WhenAll(Enumerable.Range(					1, 12).Select(					async i =>
 					{
+						if (i > 10)
+						{
+							// Give some time to threads for reaching the wait, having all of them ready to do most of their job concurrently.
+							await (Task.Delay(100));
+							semaphore.Release(10);
+							return;
+						}
 						try
 						{
 							using (var s = Sfi.OpenSession())
@@ -80,29 +175,21 @@ public async Task WillGetSessionIdFromSessionLogsConcurrentAsync()
 									(ti, old) => throw new InvalidOperationException(
 										$"Thread number {ti} has already session id {old}, while attempting to set it to" +
 										$" {s.GetSessionImplementation().SessionId}"));
-								semaphore.Wait();
+								await (semaphore.WaitAsync());
 
 								for (int j = 0; j < 10; j++)
 								{
-									s.Get<Person>(i * 10 + j); //will execute some sql
+									await (s.GetAsync<Person>(i * 10 + j)); //will execute some sql
 								}
 							}
 						}
 						catch (Exception e)
 						{
 							failures.Add(e);
 						}
-					})).ToArray();
-
-			using (var spy = new TextLogSpy("NHibernate.SQL", "%message | SessionId: %property{sessionId}"))
-			{
-				Array.ForEach(array, thread => thread.Start());
-				// Give some time to threads for reaching the wait, having all of them ready to do most of their job concurrently.
-				await (Task.Delay(100));
-				semaphore.Set();
-				Array.ForEach(array, thread => thread.Join());
+					})));
 
-				Assert.That(failures, Is.Empty, $"{failures.Count} thread(s) failed.");
+				Assert.That(failures, Is.Empty, $"{failures.Count} task(s) failed.");
 
 				var loggingEvent = spy.GetWholeLog();
 				for (var i = 1; i < 11; i++)
@@ -141,7 +228,7 @@ public TextLogSpy(string loggerName, string pattern)
 					Threshold = Level.All,
 					Writer = new StringWriter(stringBuilder)
 				};
-				loggerImpl = (Logger)LogManager.GetLogger(typeof(LogsFixtureAsync).Assembly, loggerName).Logger;
+				loggerImpl = (Logger) LogManager.GetLogger(typeof(LogsFixtureAsync).Assembly, loggerName).Logger;
 				loggerImpl.AddAppender(appender);
 				previousLevel = loggerImpl.Level;
 				loggerImpl.Level = Level.All;
@@ -158,7 +245,37 @@ public void Dispose()
 				loggerImpl.Level = previousLevel;
 			}
 		}
-	}
 
+		public class InterlacedSessionInterceptor : EmptyInterceptor
+		{
+			private readonly ISessionFactory _sfi;
+
+			public System.Collections.Generic.List<Guid> SessionIds { get; } = new System.Collections.Generic.List<Guid>();
+
+			public InterlacedSessionInterceptor(ISessionFactory sfi)
+			{
+				_sfi = sfi;
+			}
+
+			public override SqlString OnPrepareStatement(SqlString sql)
+			{
+				using (var s = _sfi.OpenSession())
+				{
+					SessionIds.Add(s.GetSessionImplementation().SessionId);
+					s.Get<Person>(SessionIds.Count); //will execute some sql
+				}
+				return base.OnPrepareStatement(sql);
+			}
 
+			public override bool OnLoad(object entity, object id, object[] state, string[] propertyNames, IType[] types)
+			{
+				using (var s = _sfi.OpenSession())
+				{
+					SessionIds.Add(s.GetSessionImplementation().SessionId);
+					s.Get<Person>(SessionIds.Count); //will execute some sql
+				}
+				return base.OnLoad(entity, id, state, propertyNames, types);
+			}
+		}
+	}
 }