CSHARP-994: fix thread starvation in Mapper

Author: verdie-g

src/Cassandra/Cassandra.csproj

@@ -8,8 +8,8 @@
     <VersionPrefix>3.22.0</VersionPrefix>
     <IncludeSourceRevisionInInformationalVersion>false</IncludeSourceRevisionInInformationalVersion>
     <Authors>DataStax</Authors>
-    <TargetFrameworks Condition="'$(BuildCoreOnly)' != 'True'">net452;netstandard2.0</TargetFrameworks>
-    <TargetFramework Condition="'$(BuildCoreOnly)' == 'True'">netstandard2.0</TargetFramework>
+    <TargetFrameworks Condition="'$(BuildCoreOnly)' != 'True'">net452;netstandard2.0;net8.0</TargetFrameworks>
+    <TargetFrameworks Condition="'$(BuildCoreOnly)' == 'True'">netstandard2.0;net8.0</TargetFrameworks>
     <NoWarn>$(NoWarn);1591</NoWarn>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
     <WarningsNotAsErrors>NU1901;NU1902;NU1903;NU1904</WarningsNotAsErrors>
@@ -24,16 +24,17 @@
     <PackageLicenseFile>LICENSE.md</PackageLicenseFile>
     <RepositoryUrl>https://github.com/datastax/csharp-driver</RepositoryUrl>
     <PackageProjectUrl>https://github.com/datastax/csharp-driver</PackageProjectUrl>
-    <LangVersion>7.1</LangVersion>
-	<NuGetAudit>false</NuGetAudit>
+    <LangVersion Condition="$(TargetFramework) != 'net8.0'">7.1</LangVersion>
+    <LangVersion Condition="$(TargetFramework) == 'net8.0'">12.0</LangVersion>
+    <NuGetAudit>false</NuGetAudit>
   </PropertyGroup>
   <PropertyGroup Condition="$([System.Text.RegularExpressions.Regex]::IsMatch('$(TargetFramework)', '^net4\d'))">
     <DefineConstants>$(DefineConstants);NETFRAMEWORK</DefineConstants>
   </PropertyGroup>
   <PropertyGroup Condition="$([System.Text.RegularExpressions.Regex]::IsMatch('$(TargetFramework)', '^net\d$'))">
     <DefineConstants>$(DefineConstants);NETCOREAPP</DefineConstants>
   </PropertyGroup>
-  
+
   <ItemGroup>
     <None Include="..\..\LICENSE.md" Pack="true" PackagePath="LICENSE.md" />
   </ItemGroup>
@@ -44,6 +45,8 @@
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="1.0.0" />
     <PackageReference Include="Newtonsoft.Json" Version="9.0.1" NoWarn="NU1903" />
     <PackageReference Include="System.Management" Version="4.7.0" />
+  </ItemGroup>
+  <ItemGroup Condition="'$(TargetFramework)' != 'net8.0'">
     <PackageReference Include="System.Runtime.InteropServices.RuntimeInformation" Version="4.0.0" />
     <PackageReference Include="System.Threading.Tasks.Dataflow" Version="4.6.0" />
   </ItemGroup>

src/Cassandra/Data/Linq/ClientProjectionCqlQuery.cs

@@ -54,7 +54,7 @@ internal override IEnumerable<TResult> AdaptResult(string cql, RowSet rs)
             if (!_canCompile)
             {
                 var mapper = MapperFactory.GetMapperWithProjection<TResult>(cql, rs, _projectionExpression);
-                result = rs.Select(mapper);
+                result = Enumerable.Select(rs, mapper);
             }
             else
             {

src/Cassandra/Data/Linq/CqlConditionalCommand.cs

@@ -76,7 +76,7 @@ protected internal override string GetCql(out object[] values)
                 Cql.New(cql, values).WithExecutionProfile(executionProfile)).ConfigureAwait(false);
             this.CopyQueryPropertiesTo(stmt);
             var rs = await session.ExecuteAsync(stmt, executionProfile).ConfigureAwait(false);
-            return AppliedInfo<TEntity>.FromRowSet(_mapperFactory, cql, rs);
+            return await AppliedInfo<TEntity>.FromRowSetAsync(_mapperFactory, cql, rs);
         }
 
         /// <summary>

src/Cassandra/Data/Linq/CqlQuery.cs

@@ -134,7 +134,12 @@ public async Task<IPage<TEntity>> ExecutePagedAsync(string executionProfile)
             var cql = visitor.GetSelect(Expression, out object[] values);
             var rs = await InternalExecuteWithProfileAsync(executionProfile, cql, values).ConfigureAwait(false);
             var mapper = MapperFactory.GetMapper<TEntity>(cql, rs);
-            return new Page<TEntity>(rs.Select(mapper), PagingState, rs.PagingState);
+#if NET8_0_OR_GREATER
+            var items = await AsyncEnumerable.Select(rs, mapper).ToListAsync();
+#else
+            var items = Enumerable.Select(rs, mapper).ToList();
+#endif
+            return new Page<TEntity>(items, PagingState, rs.PagingState);
         }
 
         /// <summary>

src/Cassandra/Data/Linq/CqlQueryBase.cs

@@ -123,7 +123,7 @@ protected async Task<RowSet> InternalExecuteWithProfileAsync(string executionPro
         internal virtual IEnumerable<TEntity> AdaptResult(string cql, RowSet rs)
         {
             var mapper = MapperFactory.GetMapper<TEntity>(cql, rs);
-            return rs.Select(mapper);
+            return Enumerable.Select(rs ,mapper);
         }
 
         /// <summary>

src/Cassandra/DataStax/Graph/GraphResultSet.cs

@@ -75,7 +75,7 @@ public IEnumerator<GraphNode> GetEnumerator()
         /// </summary>
         private IEnumerable<GraphNode> YieldNodes()
         {
-            foreach (var node in _rs.Select(_factory))
+            foreach (var node in Enumerable.Select(_rs, _factory))
             {
                 for (var i = 0; i < node.Bulk; i++)
                 {

src/Cassandra/Helpers/AsyncEnumerable.cs

@@ -0,0 +1,110 @@
+// Copyright (C) DataStax Inc.
+//
+//    Licensed under the Apache License, Version 2.0 (the "License");
+//    you may not use this file except in compliance with the License.
+//    You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+//    Unless required by applicable law or agreed to in writing, software
+//    distributed under the License is distributed on an "AS IS" BASIS,
+//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//    See the License for the specific language governing permissions and
+//    limitations under the License.
+
+#if NET8_0_OR_GREATER && !NET10_OR_GREATER // These methods are implemented in .NET 10.
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+// ReSharper disable once CheckNamespace
+namespace System.Linq;
+
+internal static class AsyncEnumerable
+{
+    public static async ValueTask<TSource> FirstAsync<TSource>(
+        this IAsyncEnumerable<TSource> source,
+        CancellationToken cancellationToken = default)
+    {
+        await using var e = source.GetAsyncEnumerator(cancellationToken);
+
+        if (!await e.MoveNextAsync())
+        {
+            throw new InvalidOperationException("Sequence contains no elements");
+        }
+
+        return e.Current;
+    }
+
+    public static async ValueTask<TSource> FirstOrDefaultAsync<TSource>(
+        this IAsyncEnumerable<TSource> source,
+        CancellationToken cancellationToken = default)
+    {
+        await using var e = source.GetAsyncEnumerator(cancellationToken);
+        return await e.MoveNextAsync() ? e.Current : default;
+    }
+
+    public static async ValueTask<TSource> SingleAsync<TSource>(
+        this IAsyncEnumerable<TSource> source,
+        CancellationToken cancellationToken = default)
+    {
+        await using var e = source.GetAsyncEnumerator(cancellationToken);
+
+        if (!await e.MoveNextAsync())
+        {
+            throw new InvalidOperationException("Sequence contains no elements");
+        }
+
+        TSource result = e.Current;
+        if (await e.MoveNextAsync())
+        {
+            throw new InvalidOperationException("Sequence contains more than one element");
+        }
+
+        return result;
+    }
+
+    public static async ValueTask<TSource> SingleOrDefaultAsync<TSource>(
+        this IAsyncEnumerable<TSource> source,
+        CancellationToken cancellationToken = default)
+    {
+        await using var e = source.GetAsyncEnumerator(cancellationToken);
+
+        if (!await e.MoveNextAsync())
+        {
+            return default;
+        }
+
+        TSource result = e.Current;
+        if (await e.MoveNextAsync())
+        {
+            throw new InvalidOperationException("Sequence contains more than one element");
+        }
+
+        return result;
+    }
+
+    public static async IAsyncEnumerable<TResult> Select<TSource, TResult>(
+        this IAsyncEnumerable<TSource> source,
+        Func<TSource, TResult> selector)
+    {
+        await foreach (TSource element in source)
+        {
+            yield return selector(element);
+        }
+    }
+
+    public static async ValueTask<List<TSource>> ToListAsync<TSource>(
+        this IAsyncEnumerable<TSource> source,
+        CancellationToken cancellationToken = default)
+    {
+        List<TSource> list = [];
+        await foreach (TSource element in source.WithCancellation(cancellationToken))
+        {
+            list.Add(element);
+        }
+
+        return list;
+    }
+}
+#endif

src/Cassandra/Mapping/AppliedInfo.cs

@@ -15,6 +15,11 @@
 //
 
 using System.Linq;
+using System.Threading.Tasks;
+
+#if !NET8_0_OR_GREATER
+#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
+#endif
 
 namespace Cassandra.Mapping
 {
@@ -54,9 +59,13 @@ public AppliedInfo(T existing)
         /// <summary>
         /// Adapts a LWT RowSet and returns a new AppliedInfo
         /// </summary>
-        internal static AppliedInfo<T> FromRowSet(MapperFactory mapperFactory, string cql, RowSet rs)
+        internal static async Task<AppliedInfo<T>> FromRowSetAsync(MapperFactory mapperFactory, string cql, RowSet rs)
         {
+#if NET8_0_OR_GREATER
+            var row = await rs.FirstOrDefaultAsync();
+#else
             var row = rs.FirstOrDefault();
+#endif
             const string appliedColumn = "[applied]";
             if (row == null || row.GetColumn(appliedColumn) == null || row.GetValue<bool>(appliedColumn))
             {

src/Cassandra/Mapping/ICqlQueryAsyncClient.cs

@@ -17,6 +17,10 @@
 using System.Collections.Generic;
 using System.Threading.Tasks;
 
+#if !NET8_0_OR_GREATER
+#pragma warning disable CS1574, CS1584, CS1581, CS1580 // Cannot resolve reference in XML comment
+#endif
+
 namespace Cassandra.Mapping
 {
     /// <summary>
@@ -25,20 +29,43 @@ namespace Cassandra.Mapping
     public interface ICqlQueryAsyncClient
     {
         /// <summary>
-        /// Gets a list of all T from Cassandra.
+        /// Gets a list of all T from Cassandra. Loading new pages when enumerating the result may block the thread. For that reason,
+        /// <see cref="FetchAsAsyncEnumerable{T}(CqlQueryOptions)"/> should be preferred.
         /// </summary>
         Task<IEnumerable<T>> FetchAsync<T>(CqlQueryOptions queryOptions = null);
-        
+
         /// <summary>
-        /// Gets a list of T from Cassandra using the CQL statement and parameter values specified.
+        /// Gets a list of T from Cassandra using the CQL statement and parameter values specified. Loading new pages when enumerating the result may
+        /// block the thread. For that reason, <see cref="FetchAsAsyncEnumerable{T}(string, object[])"/> should be preferred.
         /// </summary>
         Task<IEnumerable<T>> FetchAsync<T>(string cql, params object[] args);
 
         /// <summary>
-        /// Gets a list of T from Cassandra using the CQL statement specified.
+        /// Gets a list of T from Cassandra using the CQL statement specified. Loading new pages when enumerating the result may block the thread.
+        /// For that reason, <see cref="FetchAsAsyncEnumerable{T}(Cql)"/> should be preferred.
         /// </summary>
         Task<IEnumerable<T>> FetchAsync<T>(Cql cql);
 
+#if NET8_0_OR_GREATER
+        /// <summary>
+        /// Gets an <see cref="IAsyncEnumerable{T}"/> of all T from Cassandra. Unlike <see cref="FetchAsync{T}(CqlQueryOptions)"/>, loading new
+        /// pages when enumerating the result does not block the thread.
+        /// </summary>
+        IAsyncEnumerable<T> FetchAsAsyncEnumerable<T>(CqlQueryOptions options = null);
+
+        /// <summary>
+        /// Gets an <see cref="IAsyncEnumerable{T}"/> of all T from Cassandra using the CQL statement and parameter values specified. Unlike
+        /// <see cref="FetchAsync{T}(string, object[])"/>, loading new pages when enumerating the result does not block the thread.
+        /// </summary>
+        IAsyncEnumerable<T> FetchAsAsyncEnumerable<T>(string cql, params object[] args);
+
+        /// <summary>
+        /// Gets an <see cref="IAsyncEnumerable{T}"/> of all T from Cassandra using the CQL statement specified. Unlike
+        /// <see cref="FetchAsync{T}(Cql)"/>, loading new pages when enumerating the result does not block the thread.
+        /// </summary>
+        IAsyncEnumerable<T> FetchAsAsyncEnumerable<T>(Cql cql);
+#endif
+
         /// <summary>
         /// Gets a paged list of T results from Cassandra.
         /// Suitable for manually page through all the results of a query.