AMQNET-637 NMS 2.0 #18

lukeabsent · 2021-12-22T15:13:10Z

NMS 2.0 async await, it can be further improved/made async,

not all features are implemented, for example shared consumers, delivery delay

Havret · 2022-01-30T17:19:13Z

src/Transport/ResponseCorrelator.cs

+	        TaskCompletionSource<Response> taskCompletionSource = new TaskCompletionSource<Response>(TaskCreationOptions.RunContinuationsAsynchronously);
+			if (timeout.TotalMilliseconds > 0)
+	        {
+		        CancellationTokenSource ct = new CancellationTokenSource(timeout);


CancellationTokenSource is disposable. There is a resource leek here.

Not sure, but treid to update it anyway

Havret · 2022-01-30T17:23:57Z

src/Transport/ResponseCorrelator.cs

+			        taskCompletionSource.TrySetException(new RequestTimedOutException(timeout));
+		        }, false);
+	        }
+
            FutureResponse future = AsyncRequest(command);


This FutureResponse is some Java variation around TaskCompletionSource, I'd drop it altogether. Is doesn't add any value, but complicates logic dramatically.

Havret · 2022-01-30T17:26:17Z

src/Transport/ResponseCorrelator.cs

-            Response response = future.Response;
-            return response;
+
+            _ = future.Task.ContinueWith(t =>


The logic you are applying here should be applied as part of OnCommand. You basically should resolve promise there.

Havret · 2022-01-30T17:27:47Z

src/Transport/ResponseCorrelator.cs

+            {
+	            if (t.IsCompleted)
+	            {
+		            taskCompletionSource.SetResult(t.Result);


Please use TrySet* instead of Set* on tcs. Set may throw an InvalidOperationException, where value is already set.

Havret · 2022-01-30T17:30:35Z

src/Transport/ResponseCorrelator.cs

@@ -18,7 +18,9 @@
 using System;


I'd remove most of the synchronisation blocks and replace Hashtable.Synchronized(new Hashtable()) with thread safe ConcurrentDictionary.

I gave it a try but without this locking, had some group of tests failing.

Havret · 2022-01-30T17:33:57Z

src/Transport/FutureResponse.cs

@@ -25,8 +26,13 @@ namespace Apache.NMS.ActiveMQ.Transport
 	/// <summary>
 	/// Handles asynchronous responses
 	/// </summary>
-	public class FutureResponse
+	public class FutureResponse : TaskCompletionSource<Response>


So it seems you have 2 TCSs to handle a single request-reply. One should be enough.

seemed so : ), did a little refactoring, its bit more lean now

@Havret, so once you have some time please look again at this pr

Havret

I'd take a step back and rethink handling of async operations.

Havret · 2022-01-30T17:44:04Z

src/Transport/ResponseCorrelator.cs

+	            }
+            });
+
+            return taskCompletionSource.Task;


You should handle scenario when the operation is cancelled and remove future or TCS from list of pending operations.

Havret · 2022-03-07T21:27:29Z

src/Util/Synchronization/TaskExtensions.cs

+
+        public static T GetAsyncResult<T>(this Task<T> task)
+        {
+            return task.Await().GetAwaiter().GetResult();


When you're using GetResult there is no await so there is no point of calling ConfigureAwait.

Havret · 2022-03-07T21:33:00Z

src/AdvisoryConsumer.cs

@@ -44,7 +45,7 @@ internal AdvisoryConsumer(Connection connection, ConsumerId consumerId) : base()
            this.info.NoLocal = true;

            this.connection.AddDispatcher(consumerId, this);
-            this.connection.SyncRequest(this.info);
+            this.connection.SyncRequestAsync(this.info).GetAsyncResult();


From calling stack it seems this is used in async flow, but you're blocking it here. I guess this should be changed, so first, you asynchronously create a resource and then inject it here via the constructor.

Havret · 2022-03-07T21:35:49Z

src/AdvisoryConsumer.cs

@@ -66,7 +67,7 @@ internal void Dispose()
            }
        }

-        public void Dispatch(MessageDispatch messageDispatch)
+        public Task Dispatch_Async(MessageDispatch messageDispatch)


I don't like this dash here. Why not call this method just DispatchAsync.

i recall there was some naming collision

Havret · 2022-03-07T21:41:51Z

src/Transport/ResponseCorrelator.cs

+			if (timeout.TotalMilliseconds > 0)
+	        {
+		        CancellationTokenSource ct = new CancellationTokenSource(timeout);
+		        ct.Token.Register(() =>


Register returns CancellationTokenRegistration that also should be disposed when you're dealt with this request.

Havret · 2022-03-17T20:13:06Z

src/Commands/ActiveMQMessage.cs

-	    public virtual void ClearBody()
+		public Task AcknowledgeAsync()
+		{
+			Acknowledge();


This is blocking. Shouldn't be as the internal implementation isn't.

Havret · 2022-03-17T20:18:40Z

src/Util/Synchronization/NmsSynchronizationMonitor.cs

+    /// Goal of this is to replace lock(syncRoot) for sync and async methods, and also have Wait and Pulse(All) capabilities
+    /// Relies on AsyncLocal construct, and should be valid along the flow of executioncontext
+    /// </summary>
+    public class NmsSynchronizationMonitor


I wonder why you didn't use sth like https://www.nuget.org/packages/Nito.AsyncEx/ and reimplemented this from scratch yourself. 🤔

afaik this is non reentrant, non resursive, the other one tries to be

Havret · 2022-03-17T20:22:09Z

src/IDispatcher.cs

@@ -25,6 +26,6 @@ namespace Apache.NMS.ActiveMQ
    /// </summary>
    public interface IDispatcher
    {
-        void Dispatch(MessageDispatch messageDispatch);
+        Task Dispatch_Async(MessageDispatch messageDispatch);


Why not DispatchAsync?

there was collision with Session.DispatchAsync

Havret · 2022-03-17T20:23:25Z

src/ISynchronization.cs

 namespace Apache.NMS.ActiveMQ
 {
 	public interface ISynchronization
    {
        /// <summary>
        /// Called before a commit or rollback is applied.
        /// </summary>
-        void BeforeEnd();
+        Task BeforeEndAsync();


I hope nobody was using this public API.

Havret · 2022-03-17T20:27:08Z

src/MessageConsumer.cs

@@ -667,7 +693,8 @@ public void DeliverAcks()
                        this.executor = new ThreadPoolExecutor();
                    }

-                    this.executor.QueueUserWorkItem(AsyncDeliverAck, ack);
+                    // queue to sync pool so we have to make it sync method
+                    this.executor.QueueUserWorkItem( (obj) => AsyncDeliverAckAsync(obj).GetAsyncResult(), ack);


That's pretty strange, but I guess it makes sense.

Havret · 2022-03-17T20:36:10Z

src/Transport/FutureResponse.cs

-
-				latch.countDown();
-			}
+			set => SetResult(value);


This is not safe, it will throw an exception when you call SetCanceled or SetException first. I'd change it here and in other places to Try* equivalent.

Havret

I reviewed this PR as thoroughly as I could. I left some suggestions and remarks, but I don't know this code-base well enough to officially give you the green light.

michaelandrepearce · 2022-04-01T17:09:12Z

@Havret i think you should be able to give green light, as long as all pre-existing tests pass and your comments are addressed around the new changes.

lukeabsent · 2022-04-05T17:02:04Z

Hello @killnine perhaps you could also have a look ?

lukeabsent · 2022-04-27T12:16:46Z

There was a little update to Task runners,

Here is my tests run on my local machine, except dtc, the same tests are failing on the pre-2.0/old version.
And as much as I can see the same situation was in here
#9 (comment)

Here's the report:
nms-openwire-test (27 tests) Failed: 27 tests failed
Apache.NMS.ActiveMQ.Test (27 tests) Failed: 27 tests failed
Commands (2 tests) Failed: 2 tests failed
ActiveMQMessageTest (net472) (1 test) Failed: One or more child tests had errors: 1 test failed
TestSetEmptyPropertyName Failed: Should have thrown exception
ActiveMQMessageTest (netcoreapp3.1) (1 test) Failed: One or more child tests had errors: 1 test failed
TestSetEmptyPropertyName Failed: Should have thrown exception
DtcBasicTransactionsTest (netcoreapp3.1) (2 tests) Failed: One or more child tests had errors: 2 tests failed
TestTransacteDequeueAndDbWrite Failed: wrong number of rows in DB
TestTransactedDBReadAndProduce Failed: wrong number of rows in DB
DtcConsumerTransactionsTest (netcoreapp3.1) (13 tests) Failed: One or more child tests had errors: 13 tests failed
TestConsumeWithDBInsertLogLocation Failed: Expected: 1
TestIterativeTransactedConsume Failed: wrong number of rows in DB
TestRecoverAfterFailOnTransactionAfterPrepareSent Failed: Expected: 0
TestRecoverAfterFailOnTransactionBeforePrepareSent Failed: Expected: 5
TestRecoverAfterRollbackFailWhenScopeAborted Failed: Expected: 5
TestRecoverAfterTransactionScopeAborted Failed: Expected: 5
TestRecoveryAfterCommitFailsAfterSent Failed: Expected: 0
TestRecoveryAfterCommitFailsBeforeSent Failed: Expected: 0
TestRedelivered Failed: System.PlatformNotSupportedException : This platform does not support distributed transactions.
TestRedeliveredCase2 Failed: System.PlatformNotSupportedException : This platform does not support distributed transactions.
TestRedeliveredCase3 Failed: System.PlatformNotSupportedException : This platform does not support distributed transactions.
TestRedeliveredNoComplete Failed: System.PlatformNotSupportedException : This platform does not support distributed transactions.
TestTransactedAsyncConsumption Failed: System.Transactions.TransactionAbortedException : The transaction has aborted.
DtcProducerTransactionsTest (netcoreapp3.1) (4 tests) Failed: One or more child tests had errors: 4 tests failed
TestIterativeTransactedProduceWithDBDelete Failed: wrong number of rows in DB
TestNoRecoverAfterFailOnTransactionWhenLogDeleted Failed: wrong number of rows in DB
TestRecoverAfterFailOnTransactionCommit Failed: wrong number of rows in DB
TestRecoverAfterFailOnTransactionPostCommitSend Failed: Expected: 5
MaxInactivityDurationTest (net472) (2 tests) Failed: One or more child tests had errors: 2 tests failed
TestInactivityMonitorThreadLeak (2 tests) Failed: One or more child tests had errors: 2 tests failed
TestInactivityMonitorThreadLeak(0) Failed: Handle count grew beyond maximum of 500 on iteration #25.
TestInactivityMonitorThreadLeak(1000) Failed: Handle count grew beyond maximum of 500 on iteration #19.
MaxInactivityDurationTest (netcoreapp3.1) (2 tests) Failed: One or more child tests had errors: 2 tests failed
TestInactivityMonitorThreadLeak (2 tests) Failed: One or more child tests had errors: 2 tests failed
TestInactivityMonitorThreadLeak(0) Failed: Handle count grew beyond maximum of 500 on iteration #44.
TestInactivityMonitorThreadLeak(1000) Failed: Handle count grew beyond maximum of 500 on iteration #26.
TempDestinationTest (net472) (1 test) Failed: One or more child tests had errors: 1 test failed
TestTmpQueueWorksUnderLoad Failed: Expected: <ActiveMQBytesMessage[ commandId = 0, responseRequired = False,
TempDestinationTest (netcoreapp3.1) (1 test) Failed: One or more child tests had errors: 1 test failed
TestTmpQueueWorksUnderLoad Failed: Expected: <ActiveMQBytesMessage[ commandId = 0, responseRequired = False,

When run in batch (all at once) sometimes these fail: InactivityMonitorTest.TestWriteMessageFail, AMQNET375Test.TestZeroPRefetchConsumerGetsAllMessages) but then after running separately they turn green.

Havret · 2022-05-29T18:12:06Z

src/Threads/CompositeTaskRunner.cs

@@ -28,6 +28,8 @@ public class CompositeTaskRunner : TaskRunner
    {
        private readonly Mutex mutex = new Mutex();
        private readonly Thread theThread = null;
+        private AsyncLocal<bool> workExecutionContextCurrentlyProcessing = new AsyncLocal<bool>();


What's the reasoning behind this flag? What is it supposed to do?

Havret

I guess it's ok, if this helps the tests pass, but the reasoning behind this implementation exceeds my technical understanding of this library.

michaelandrepearce · 2022-06-06T10:07:40Z

+1 from me, with same test coverage as previous. Main thing here is to note this will be first pass of 2.0 support, so its expected there maybe issues raised from end users, and we simply will have to bug fix those.

lukeabsent force-pushed the AMQNET-637 branch from 4c90a49 to f2832bc Compare December 22, 2021 15:17

AMQNET-637 NMS 2.0

fafd30a

lukeabsent force-pushed the AMQNET-637 branch from f2832bc to fafd30a Compare December 30, 2021 11:09

lukeabsent marked this pull request as ready for review December 30, 2021 15:10

Havret reviewed Jan 30, 2022

View reviewed changes

Havret requested changes Jan 30, 2022

View reviewed changes

Havret reviewed Jan 30, 2022

View reviewed changes

AMQNET-637 NMS 2.0

61a3de5

Havret reviewed Mar 7, 2022

View reviewed changes

Havret reviewed Mar 17, 2022

View reviewed changes

michaelandrepearce mentioned this pull request Apr 1, 2022

AMQNET-735 Add AsyncMessageListener to Consumer API apache/activemq-nms-api#26

Merged

AMQNET-637 NMS 2.0

1e667f2

lukeabsent force-pushed the AMQNET-637 branch from b758a2e to 1e667f2 Compare April 4, 2022 10:44

AMQNET-637 NMS 2.0

c155c5a

Havret reviewed May 29, 2022

View reviewed changes

Havret approved these changes May 31, 2022

View reviewed changes

Havret merged commit fde8a9b into apache:main Jun 6, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

AMQNET-637 NMS 2.0 #18

AMQNET-637 NMS 2.0 #18

lukeabsent commented Dec 22, 2021

Havret Jan 30, 2022

lukeabsent Feb 11, 2022

Havret Jan 30, 2022

Havret Jan 30, 2022

Havret Jan 30, 2022 •

edited

Loading

Havret Jan 30, 2022

lukeabsent Feb 11, 2022

Havret Jan 30, 2022

lukeabsent Feb 11, 2022

lukeabsent Feb 11, 2022

Havret left a comment

Havret Jan 30, 2022

Havret Mar 7, 2022

Havret Mar 7, 2022

Havret Mar 7, 2022

lukeabsent Mar 29, 2022

Havret Mar 7, 2022

Havret Mar 17, 2022

Havret Mar 17, 2022

lukeabsent Mar 29, 2022

Havret Mar 17, 2022

lukeabsent Mar 29, 2022

Havret Mar 17, 2022

Havret Mar 17, 2022

Havret Mar 17, 2022

Havret left a comment •

edited

Loading

michaelandrepearce commented Apr 1, 2022

lukeabsent commented Apr 5, 2022

lukeabsent commented Apr 27, 2022 •

edited

Loading

Havret May 29, 2022

Havret left a comment •

edited

Loading

michaelandrepearce commented Jun 6, 2022

AMQNET-637 NMS 2.0 #18

AMQNET-637 NMS 2.0 #18

Conversation

lukeabsent commented Dec 22, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Havret Jan 30, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Havret left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Havret left a comment • edited Loading

Choose a reason for hiding this comment

michaelandrepearce commented Apr 1, 2022

lukeabsent commented Apr 5, 2022

lukeabsent commented Apr 27, 2022 • edited Loading

Choose a reason for hiding this comment

Havret left a comment • edited Loading

Choose a reason for hiding this comment

michaelandrepearce commented Jun 6, 2022

Havret Jan 30, 2022 •

edited

Loading

Havret left a comment •

edited

Loading

lukeabsent commented Apr 27, 2022 •

edited

Loading

Havret left a comment •

edited

Loading