bench

joao-r-reis · joao-r-reis · commit 8b6235ea1d85 · 2024-09-26T18:47:08.000+01:00
diff --git a/src/Cassandra/Connections/Connection.cs b/src/Cassandra/Connections/Connection.cs
@@ -761,15 +761,15 @@ private static bool InvokeReadCallbacks(MemoryStream stream, ICollection<Func<Me
                 return false;
             }
             //Invoke all callbacks using the default TaskScheduler
-            Task.Run(async () =>
+            Task.Factory.StartNew(async () =>
             {
                 stream.Position = 0;
                 foreach (var cb in operationCallbacks)
                 {
                     await cb(stream, timestamp).ConfigureAwait(false);
                 }
                 stream.Dispose();
-            }, CancellationToken.None);
+            }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);
             return true;
         }
 
@@ -803,7 +803,9 @@ public OperationState Send(IRequest request, Func<IRequestError, Response, Task>
             if (_isClosed)
             {
                 // Avoid calling back before returning
-                Task.Run(() => callback(RequestError.CreateClientError(new SocketException((int)SocketError.NotConnected), true), null), CancellationToken.None);
+                Task.Factory.StartNew(
+                    () => callback(RequestError.CreateClientError(new SocketException((int)SocketError.NotConnected), true), null), 
+                    CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);
                 return null;
             }
 
@@ -827,7 +829,9 @@ public OperationState Send(IRequest request, Func<IRequestError, Response, Task>
                 catch (Exception ex)
                 {
                     // Avoid calling back before returning
-                    Task.Run(() => callback(RequestError.CreateClientError(ex, true), null), CancellationToken.None);
+                    Task.Factory.StartNew(
+                        () => callback(RequestError.CreateClientError(ex, true), null),
+                        CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);
                     return null;
                 }
             }
diff --git a/src/Cassandra/Connections/HostConnectionPool.cs b/src/Cassandra/Connections/HostConnectionPool.cs
@@ -404,7 +404,7 @@ internal void OnConnectionClosing(IConnection c = null)
                 return;
             }
             // We are using an IO thread
-            Task.Run(async () =>
+            Task.Factory.StartNew(async () =>
             {
                 // Use a lock for avoiding concurrent calls to SetNewConnectionTimeout()
                 await _allConnectionClosedEventLock.WaitAsync().ConfigureAwait(false);
@@ -427,7 +427,7 @@ internal void OnConnectionClosing(IConnection c = null)
                 {
                     _allConnectionClosedEventLock.Release();
                 }
-            }).Forget();
+            }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);
         }
 
         private void OnDistanceChanged(HostDistance previousDistance, HostDistance distance)
diff --git a/src/Cassandra/OperationState.cs b/src/Cassandra/OperationState.cs
@@ -155,7 +155,7 @@ public void InvokeCallback(IRequestError error, long timestamp)
             }
             //Invoke the callback in a new thread in the thread pool
             //This way we don't let the user block on a thread used by the Connection
-            Task.Run(() => callback(error, null, timestamp), CancellationToken.None);
+            Task.Factory.StartNew(() => callback(error, null, timestamp), CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);
         }
 
         /// <summary>
@@ -175,7 +175,7 @@ public bool MarkAsTimedOut(OperationTimedOutException ex, Func<Task> onReceive,
             Thread.MemoryBarrier();
 
             _timeoutCallbackSet = true;
-            Task.Run(() => callback(RequestError.CreateClientError(ex, false), null, timestamp), CancellationToken.None);
+            Task.Factory.StartNew(() => callback(RequestError.CreateClientError(ex, false), null, timestamp), CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);
             return true;
         }
 
@@ -204,14 +204,14 @@ public void Cancel()
         /// </summary>
         internal static void CallbackMultiple(IEnumerable<OperationState> ops, IRequestError error, long timestamp)
         {
-            Task.Run(async () =>
+            Task.Factory.StartNew(async () =>
             {
                 foreach (var state in ops)
                 {
                     var callback = state.SetCompleted();
                     await callback(error, null, timestamp).ConfigureAwait(false);
                 }
-            }, CancellationToken.None);
+            }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);
         }
     }
 }
diff --git a/src/Cassandra/ProtocolEvents/ProtocolEventDebouncer.cs b/src/Cassandra/ProtocolEvents/ProtocolEventDebouncer.cs
@@ -16,6 +16,7 @@
 
 using System;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using System.Threading.Tasks.Dataflow;
 
@@ -184,7 +185,7 @@ private void Process()
             _queue = null;
 
             // not necessary to enqueue within the exclusive scheduler
-            Task.Run(async () =>
+            Task.Factory.StartNew(async () =>
             {
                 var sent = false;
                 try
@@ -203,7 +204,7 @@ private void Process()
                         cb?.TrySetException(new DriverInternalError("Could not process events in the ProtocolEventDebouncer."));
                     }
                 }
-            }).Forget();
+            }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);
         }
 
         private static async Task ProcessQueue(EventQueue queue)
diff --git a/src/Cassandra/Requests/RequestExecution.cs b/src/Cassandra/Requests/RequestExecution.cs
@@ -21,6 +21,7 @@
 using System.Linq;
 using System.Net;
 using System.Net.Sockets;
+using System.Threading;
 using System.Threading.Tasks;
 using Cassandra.Connections;
 using Cassandra.Observers.Abstractions;
@@ -573,11 +574,11 @@ bool SearchBoundStatement(Statement s) =>
                     preparedKeyspace, _session.Keyspace));
 
                 var c = _connection;
-                Task.Run(async () =>
+                Task.Factory.StartNew(async () =>
                 {
                     await c.SetKeyspace(preparedKeyspace).ConfigureAwait(false);
                     await SendAsync(request, nodeRequestInfo.Host, NewReprepareResponseHandler(ex)).ConfigureAwait(false);
-                }).Forget();
+                }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default).Forget();
                 return;
             }
             await SendAsync(request, nodeRequestInfo.Host, NewReprepareResponseHandler(ex)).ConfigureAwait(false);
diff --git a/src/Cassandra/Requests/RequestHandler.cs b/src/Cassandra/Requests/RequestHandler.cs
@@ -253,7 +253,7 @@ private async Task<bool> SetCompletedAsync(Exception ex, RowSet result, Func<Tas
             if (action != null)
             {
                 //Create a new Task using the default scheduler, invoke the action and set the result
-                Task.Run(async () =>
+                Task.Factory.StartNew(async () =>
                 {
                     try
                     {
@@ -268,7 +268,7 @@ private async Task<bool> SetCompletedAsync(Exception ex, RowSet result, Func<Tas
 
                     await ClearNodeExecutionsAsync().ConfigureAwait(false);
                     await _requestResultHandler.TrySetResultAsync(result, _sessionRequestInfo).ConfigureAwait(false);
-                }, CancellationToken.None).Forget();
+                }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default).Forget();
                 return true;
             }
 
@@ -506,7 +506,7 @@ private void ScheduleNext(Host currentHost)
             _nextExecutionTimeout = _session.Cluster.Configuration.Timer.NewTimeout(_ =>
             {
                 // Start the speculative execution outside the IO thread
-                Task.Run(() =>
+                Task.Factory.StartNew(() =>
                 {
                     if (HasCompleted())
                     {
@@ -516,7 +516,7 @@ private void ScheduleNext(Host currentHost)
                     RequestHandler.Logger.Info("Starting new speculative execution after {0} ms. Last used host: {1}", delay, currentHost.Address);
                     _requestObserver.OnSpeculativeExecution(currentHost, delay);
                     return StartNewExecutionAsync();
-                }, CancellationToken.None);
+                }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);
             }, null, delay);
         }
 

Original file line number	Diff line number	Diff line change
`@@ -761,15 +761,15 @@ private static bool InvokeReadCallbacks(MemoryStream stream, ICollection<Func<Me`
`761`	`761`	`return false;`
`762`	`762`	`}`
`763`	`763`	`//Invoke all callbacks using the default TaskScheduler`
`764`		`- Task.Run(async () =>`
	`764`	`+ Task.Factory.StartNew(async () =>`
`765`	`765`	`{`
`766`	`766`	`stream.Position = 0;`
`767`	`767`	`foreach (var cb in operationCallbacks)`
`768`	`768`	`{`
`769`	`769`	`await cb(stream, timestamp).ConfigureAwait(false);`
`770`	`770`	`}`
`771`	`771`	`stream.Dispose();`
`772`		`- }, CancellationToken.None);`
	`772`	`+ }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);`
`773`	`773`	`return true;`
`774`	`774`	`}`
`775`	`775`
`@@ -803,7 +803,9 @@ public OperationState Send(IRequest request, Func<IRequestError, Response, Task>`
`803`	`803`	`if (_isClosed)`
`804`	`804`	`{`
`805`	`805`	`// Avoid calling back before returning`
`806`		`- Task.Run(() => callback(RequestError.CreateClientError(new SocketException((int)SocketError.NotConnected), true), null), CancellationToken.None);`
	`806`	`+ Task.Factory.StartNew(`
	`807`	`+ () => callback(RequestError.CreateClientError(new SocketException((int)SocketError.NotConnected), true), null),`
	`808`	`+ CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);`
`807`	`809`	`return null;`
`808`	`810`	`}`
`809`	`811`
`@@ -827,7 +829,9 @@ public OperationState Send(IRequest request, Func<IRequestError, Response, Task>`
`827`	`829`	`catch (Exception ex)`
`828`	`830`	`{`
`829`	`831`	`// Avoid calling back before returning`
`830`		`- Task.Run(() => callback(RequestError.CreateClientError(ex, true), null), CancellationToken.None);`
	`832`	`+ Task.Factory.StartNew(`
	`833`	`+ () => callback(RequestError.CreateClientError(ex, true), null),`
	`834`	`+ CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);`
`831`	`835`	`return null;`
`832`	`836`	`}`
`833`	`837`	`}`
Original file line number	Diff line number	Diff line change
`@@ -404,7 +404,7 @@ internal void OnConnectionClosing(IConnection c = null)`
`404`	`404`	`return;`
`405`	`405`	`}`
`406`	`406`	`// We are using an IO thread`
`407`		`- Task.Run(async () =>`
	`407`	`+ Task.Factory.StartNew(async () =>`
`408`	`408`	`{`
`409`	`409`	`// Use a lock for avoiding concurrent calls to SetNewConnectionTimeout()`
`410`	`410`	`await _allConnectionClosedEventLock.WaitAsync().ConfigureAwait(false);`
`@@ -427,7 +427,7 @@ internal void OnConnectionClosing(IConnection c = null)`
`427`	`427`	`{`
`428`	`428`	`_allConnectionClosedEventLock.Release();`
`429`	`429`	`}`
`430`		`- }).Forget();`
	`430`	`+ }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);`
`431`	`431`	`}`
`432`	`432`
`433`	`433`	`private void OnDistanceChanged(HostDistance previousDistance, HostDistance distance)`
Original file line number	Diff line number	Diff line change
`@@ -155,7 +155,7 @@ public void InvokeCallback(IRequestError error, long timestamp)`
`155`	`155`	`}`
`156`	`156`	`//Invoke the callback in a new thread in the thread pool`
`157`	`157`	`//This way we don't let the user block on a thread used by the Connection`
`158`		`- Task.Run(() => callback(error, null, timestamp), CancellationToken.None);`
	`158`	`+ Task.Factory.StartNew(() => callback(error, null, timestamp), CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);`
`159`	`159`	`}`
`160`	`160`
`161`	`161`	`/// <summary>`
`@@ -175,7 +175,7 @@ public bool MarkAsTimedOut(OperationTimedOutException ex, Func<Task> onReceive,`
`175`	`175`	`Thread.MemoryBarrier();`
`176`	`176`
`177`	`177`	`_timeoutCallbackSet = true;`
`178`		`- Task.Run(() => callback(RequestError.CreateClientError(ex, false), null, timestamp), CancellationToken.None);`
	`178`	`+ Task.Factory.StartNew(() => callback(RequestError.CreateClientError(ex, false), null, timestamp), CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);`
`179`	`179`	`return true;`
`180`	`180`	`}`
`181`	`181`
`@@ -204,14 +204,14 @@ public void Cancel()`
`204`	`204`	`/// </summary>`
`205`	`205`	`internal static void CallbackMultiple(IEnumerable<OperationState> ops, IRequestError error, long timestamp)`
`206`	`206`	`{`
`207`		`- Task.Run(async () =>`
	`207`	`+ Task.Factory.StartNew(async () =>`
`208`	`208`	`{`
`209`	`209`	`foreach (var state in ops)`
`210`	`210`	`{`
`211`	`211`	`var callback = state.SetCompleted();`
`212`	`212`	`await callback(error, null, timestamp).ConfigureAwait(false);`
`213`	`213`	`}`
`214`		`- }, CancellationToken.None);`
	`214`	`+ }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);`
`215`	`215`	`}`
`216`	`216`	`}`
`217`	`217`	`}`
Original file line number	Diff line number	Diff line change
`@@ -16,6 +16,7 @@`
`16`	`16`
`17`	`17`	`using System;`
`18`	`18`	`using System.Linq;`
	`19`	`+using System.Threading;`
`19`	`20`	`using System.Threading.Tasks;`
`20`	`21`	`using System.Threading.Tasks.Dataflow;`
`21`	`22`
`@@ -184,7 +185,7 @@ private void Process()`
`184`	`185`	`_queue = null;`
`185`	`186`
`186`	`187`	`// not necessary to enqueue within the exclusive scheduler`
`187`		`- Task.Run(async () =>`
	`188`	`+ Task.Factory.StartNew(async () =>`
`188`	`189`	`{`
`189`	`190`	`var sent = false;`
`190`	`191`	`try`
`@@ -203,7 +204,7 @@ private void Process()`
`203`	`204`	`cb?.TrySetException(new DriverInternalError("Could not process events in the ProtocolEventDebouncer."));`
`204`	`205`	`}`
`205`	`206`	`}`
`206`		`- }).Forget();`
	`207`	`+ }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);`
`207`	`208`	`}`
`208`	`209`
`209`	`210`	`private static async Task ProcessQueue(EventQueue queue)`
Original file line number	Diff line number	Diff line change
`@@ -253,7 +253,7 @@ private async Task<bool> SetCompletedAsync(Exception ex, RowSet result, Func<Tas`
`253`	`253`	`if (action != null)`
`254`	`254`	`{`
`255`	`255`	`//Create a new Task using the default scheduler, invoke the action and set the result`
`256`		`- Task.Run(async () =>`
	`256`	`+ Task.Factory.StartNew(async () =>`
`257`	`257`	`{`
`258`	`258`	`try`
`259`	`259`	`{`
`@@ -268,7 +268,7 @@ private async Task<bool> SetCompletedAsync(Exception ex, RowSet result, Func<Tas`
`268`	`268`
`269`	`269`	`await ClearNodeExecutionsAsync().ConfigureAwait(false);`
`270`	`270`	`await _requestResultHandler.TrySetResultAsync(result, _sessionRequestInfo).ConfigureAwait(false);`
`271`		`- }, CancellationToken.None).Forget();`
	`271`	`+ }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default).Forget();`
`272`	`272`	`return true;`
`273`	`273`	`}`
`274`	`274`
`@@ -506,7 +506,7 @@ private void ScheduleNext(Host currentHost)`
`506`	`506`	`_nextExecutionTimeout = _session.Cluster.Configuration.Timer.NewTimeout(_ =>`
`507`	`507`	`{`
`508`	`508`	`// Start the speculative execution outside the IO thread`
`509`		`- Task.Run(() =>`
	`509`	`+ Task.Factory.StartNew(() =>`
`510`	`510`	`{`
`511`	`511`	`if (HasCompleted())`
`512`	`512`	`{`
`@@ -516,7 +516,7 @@ private void ScheduleNext(Host currentHost)`
`516`	`516`	`RequestHandler.Logger.Info("Starting new speculative execution after {0} ms. Last used host: {1}", delay, currentHost.Address);`
`517`	`517`	`_requestObserver.OnSpeculativeExecution(currentHost, delay);`
`518`	`518`	`return StartNewExecutionAsync();`
`519`		`- }, CancellationToken.None);`
	`519`	`+ }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.Default);`
`520`	`520`	`}, null, delay);`
`521`	`521`	`}`
`522`	`522`