Added support for the CancelQuery() method in IStatelessSession

src/NHibernate.Test/Async/Stateless/StatelessSessionCancelQueryFixture.cs

@@ -0,0 +1,120 @@
+//------------------------------------------------------------------------------
+// <auto-generated>
+//     This code was generated by AsyncGenerator.
+//
+//     Changes to this file may cause incorrect behavior and will be lost if
+//     the code is regenerated.
+// </auto-generated>
+//------------------------------------------------------------------------------
+
+
+using System;
+using System.Data;
+using System.Data.SqlClient;
+using System.Globalization;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using NHibernate.Linq;
+using NUnit.Framework;
+
+namespace NHibernate.Test.Stateless
+{
+	[TestFixture]
+	public class StatelessSessionCancelQueryFixtureAsync : TestCase
+	{
+		protected override string MappingsAssembly
+		{
+			get { return "NHibernate.Test"; }
+		}
+
+		protected override string[] Mappings
+		{
+			get { return new[] { "Stateless.Document.hbm.xml" }; }
+		}
+
+		private const string _documentName = "SomeDocument";
+		private CultureInfo _backupCulture;
+		private CultureInfo _backupUICulture;
+
+		protected override void OnSetUp()
+		{
+			if (CultureInfo.CurrentCulture.TwoLetterISOLanguageName != CultureInfo.InvariantCulture.TwoLetterISOLanguageName)
+			{
+				// This test needs to run in English
+				_backupCulture = CultureInfo.CurrentCulture;
+				_backupUICulture = CultureInfo.CurrentUICulture;
+				CultureInfo.CurrentCulture = CultureInfo.InvariantCulture;
+				CultureInfo.CurrentUICulture = CultureInfo.InvariantCulture;
+			}
+
+			using (var s = Sfi.OpenStatelessSession())
+			using (var t = s.BeginTransaction())
+			{
+				s.Insert(new Document("Some text", _documentName));
+				t.Commit();
+			}
+		}
+
+		protected override void OnTearDown()
+		{
+			using (var s = Sfi.OpenStatelessSession())
+			using (var t = s.BeginTransaction())
+			{
+				s.CreateQuery("delete Document").ExecuteUpdate();
+				t.Commit();
+			}
+
+			if (_backupCulture != null)
+			{
+				CultureInfo.CurrentCulture = _backupCulture;
+				CultureInfo.CurrentUICulture = _backupUICulture;
+			}
+		}
+
+		protected override bool AppliesTo(Dialect.Dialect dialect)
+		{
+			return TestDialect.SupportsCancelQuery &&
+				TestDialect.SupportsSelectForUpdate;
+		}
+
+		private async Task CancelQueryTestAsync(Action<IStatelessSession> queryAction, CancellationToken cancellationToken = default(CancellationToken))
+		{
+			using (var s1 = Sfi.OpenStatelessSession())
+			using (var t1 = s1.BeginTransaction())
+			{
+				await (s1.GetAsync<Document>(_documentName, LockMode.Upgrade, cancellationToken));
+
+				using (var s2 = Sfi.OpenStatelessSession())
+				using (var t2 = s2.BeginTransaction())
+				{
+					var queryTask = Task.Factory.StartNew(() => queryAction(s2));
+
+					await (Task.Delay(200, cancellationToken));
+					s2.CancelQuery();
+					Assert.That(() => queryTask,
+						Throws.InnerException.TypeOf(typeof(OperationCanceledException))
+							.Or.InnerException.Message.Contains("cancel"));
+				}
+			}
+		}
+
+		[Test]
+		public async Task CancelHqlQueryAsync()
+		{
+			await (CancelQueryTestAsync(s => s.CreateQuery("from Document d").SetLockMode("d", LockMode.Upgrade).List<Document>()));
+		}
+
+		[Test]
+		public async Task CancelLinqQueryAsync()
+		{
+			await (CancelQueryTestAsync(s => s.Query<Document>().WithLock(LockMode.Upgrade).ToList()));
+		}
+
+		[Test]
+		public async Task CancelQueryOverQueryAsync()
+		{
+			await (CancelQueryTestAsync(s => s.QueryOver<Document>().Lock().Upgrade.List()));
+		}
+	}
+}

src/NHibernate.Test/Stateless/StatelessSessionCancelQueryFixture.cs

@@ -0,0 +1,110 @@
+using System;
+using System.Data;
+using System.Data.SqlClient;
+using System.Globalization;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using NHibernate.Linq;
+using NUnit.Framework;
+
+namespace NHibernate.Test.Stateless
+{
+	[TestFixture]
+	public class StatelessSessionCancelQueryFixture : TestCase
+	{
+		protected override string MappingsAssembly
+		{
+			get { return "NHibernate.Test"; }
+		}
+
+		protected override string[] Mappings
+		{
+			get { return new[] { "Stateless.Document.hbm.xml" }; }
+		}
+
+		private const string _documentName = "SomeDocument";
+		private CultureInfo _backupCulture;
+		private CultureInfo _backupUICulture;
+
+		protected override void OnSetUp()
+		{
+			if (CultureInfo.CurrentCulture.TwoLetterISOLanguageName != CultureInfo.InvariantCulture.TwoLetterISOLanguageName)
+			{
+				// This test needs to run in English
+				_backupCulture = CultureInfo.CurrentCulture;
+				_backupUICulture = CultureInfo.CurrentUICulture;
+				CultureInfo.CurrentCulture = CultureInfo.InvariantCulture;
+				CultureInfo.CurrentUICulture = CultureInfo.InvariantCulture;
+			}
+
+			using (var s = Sfi.OpenStatelessSession())
+			using (var t = s.BeginTransaction())
+			{
+				s.Insert(new Document("Some text", _documentName));
+				t.Commit();
+			}
+		}
+
+		protected override void OnTearDown()
+		{
+			using (var s = Sfi.OpenStatelessSession())
+			using (var t = s.BeginTransaction())
+			{
+				s.CreateQuery("delete Document").ExecuteUpdate();
+				t.Commit();
+			}
+
+			if (_backupCulture != null)
+			{
+				CultureInfo.CurrentCulture = _backupCulture;
+				CultureInfo.CurrentUICulture = _backupUICulture;
+			}
+		}
+
+		protected override bool AppliesTo(Dialect.Dialect dialect)
+		{
+			return TestDialect.SupportsCancelQuery &&
+				TestDialect.SupportsSelectForUpdate;
+		}
+
+		private void CancelQueryTest(Action<IStatelessSession> queryAction)
+		{
+			using (var s1 = Sfi.OpenStatelessSession())
+			using (var t1 = s1.BeginTransaction())
+			{
+				s1.Get<Document>(_documentName, LockMode.Upgrade);
+
+				using (var s2 = Sfi.OpenStatelessSession())
+				using (var t2 = s2.BeginTransaction())
+				{
+					var queryTask = Task.Factory.StartNew(() => queryAction(s2));
+
+					Thread.Sleep(200);
+					s2.CancelQuery();
+					Assert.That(() => queryTask,
+						Throws.InnerException.TypeOf(typeof(OperationCanceledException))
+							.Or.InnerException.Message.Contains("cancel"));
+				}
+			}
+		}
+
+		[Test]
+		public void CancelHqlQuery()
+		{
+			CancelQueryTest(s => s.CreateQuery("from Document d").SetLockMode("d", LockMode.Upgrade).List<Document>());
+		}
+
+		[Test]
+		public void CancelLinqQuery()
+		{
+			CancelQueryTest(s => s.Query<Document>().WithLock(LockMode.Upgrade).ToList());
+		}
+
+		[Test]
+		public void CancelQueryOverQuery()
+		{
+			CancelQueryTest(s => s.QueryOver<Document>().Lock().Upgrade.List());
+		}
+	}
+}

src/NHibernate.Test/TestDialect.cs

@@ -198,5 +198,10 @@ public bool SupportsSqlType(SqlType sqlType)
 		/// Returns true if you can modify the same table which you use in the SELECT part.
 		/// </summary>
 		public virtual bool SupportsModifyAndSelectSameTable => true;
+
+		/// <summary>
+		/// Returns true if you can cancel a query.
+		/// </summary>
+		public virtual bool SupportsCancelQuery => true;
 	}
 }

src/NHibernate.Test/TestDialects/MsSql2008TestDialect.cs

@@ -1,4 +1,6 @@
-namespace NHibernate.Test.TestDialects
+using System.Runtime.InteropServices;
+
+namespace NHibernate.Test.TestDialects
 {
 	public class MsSql2008TestDialect : TestDialect
 	{
@@ -11,5 +13,9 @@ public MsSql2008TestDialect(Dialect.Dialect dialect)
 		/// Does not support SELECT FOR UPDATE with paging
 		/// </summary>
 		public override bool SupportsSelectForUpdateWithPaging => false;
+
+		/// <inheritdoc />
+		/// <remarks>Canceling a query hangs under Linux with Sql2008ClientDriver. (It may be a data provider bug fixed with MicrosoftDataSqlClientDriver.)</remarks>
+		public override bool SupportsCancelQuery => !RuntimeInformation.IsOSPlatform(OSPlatform.Linux);
 	}
 }

src/NHibernate.Test/TestDialects/Oracle10gTestDialect.cs

@@ -1,4 +1,6 @@
-namespace NHibernate.Test.TestDialects
+using System.Runtime.InteropServices;
+
+namespace NHibernate.Test.TestDialects
 {
 	public class Oracle10gTestDialect : TestDialect
 	{
@@ -12,5 +14,9 @@ public Oracle10gTestDialect(Dialect.Dialect dialect) : base(dialect)
 		public override bool SupportsSelectForUpdateWithPaging => false;
 
 		public override bool SupportsAggregateInSubSelect => true;
+
+		/// <inheritdoc />
+		/// <remarks>Canceling a query hangs under Linux with OracleManagedDataClientDriver 21.6.1.</remarks>
+		public override bool SupportsCancelQuery => !RuntimeInformation.IsOSPlatform(OSPlatform.Linux);
 	}
 }

src/NHibernate/AdoNet/AbstractBatcher.cs

@@ -253,6 +253,11 @@ private DbDataReader DoExecuteReader(DbCommand cmd)
 			try
 			{
 				var reader = cmd.ExecuteReader();
+				if (reader == null)
+				{
+					// MySql may return null instead of an exception, by example when the query is canceled by another thread.
+					throw new InvalidOperationException("The query execution has yielded a null reader. (Has it been canceled?)");
+				}
 				return _factory.ConnectionProvider.Driver.SupportsMultipleOpenReaders
 					? reader
 					: NHybridDataReader.Create(reader);

src/NHibernate/Async/AdoNet/AbstractBatcher.cs

@@ -168,6 +168,11 @@ private async Task<DbDataReader> DoExecuteReaderAsync(DbCommand cmd, Cancellatio
 			try
 			{
 				var reader = await (cmd.ExecuteReaderAsync(cancellationToken)).ConfigureAwait(false);
+				if (reader == null)
+				{
+					// MySql may return null instead of an exception, by example when the query is canceled by another thread.
+					throw new InvalidOperationException("The query execution has yielded a null reader. (Has it been canceled?)");
+				}
 				return _factory.ConnectionProvider.Driver.SupportsMultipleOpenReaders
 					? reader
 					: await (NHybridDataReader.CreateAsync(reader, cancellationToken)).ConfigureAwait(false);

src/NHibernate/IStatelessSession.cs

@@ -74,6 +74,22 @@ public static void FlushBatcher(this IStatelessSession session)
 		{
 			session.GetSessionImplementation().Flush();
 		}
+
+		/// <summary>
+		/// Cancel execution of the current query.
+		/// </summary>
+		/// <remarks>
+		/// May be called from one thread to stop execution of a query in another thread.
+		/// Use with care!
+		/// </remarks>
+		public static void CancelQuery(this IStatelessSession session)
+		{
+			var implementation = session.GetSessionImplementation();
+			using (implementation.BeginProcess())
+			{
+				implementation.Batcher.CancelLastQuery();
+			}
+		}
 	}
 
 	/// <summary>