AMQP Stack polls for operations in a background thread. (Azure#4914)

There are two significant parts to this fix. The first is that the Connection object now has a method EnableAsyncOperation which registers the connection with a background thread which polls the connection. The EnableAsyncOperation needs to be callable from outside the connection because in normal client operation, the calls to open the connection are all internal to uAMQP. That means that the message sender and receiver need to call it when they're opened (and closed).

The other major part of this change is that the AsyncOperationQueue now has a WaitForRequest API which does not normally poll (there is a test hook which enables polling but that is not normally used in most scenarios).

The other part of this change fixes some concurrency issues associated with the Log::Stream functionality.

Co-authored-by: Anton Kolesnyk <[email protected]>

* Update sdk/core/azure-core/src/logger.cpp

Co-authored-by: Anton Kolesnyk <[email protected]>

* Update sdk/core/azure-core/test/ut/CMakeLists.txt

Co-authored-by: Anton Kolesnyk <[email protected]>

* Update sdk/core/azure-core/test/ut/logging_test.cpp

Co-authored-by: Anton Kolesnyk <[email protected]>

* Added ASAN as an option, converted transports to be bases to ensure consistancy

---------

Co-authored-by: Anton Kolesnyk <[email protected]>

Author: LarryOsterman

.gitignore

@@ -47,6 +47,9 @@ Generated\ Files/
 [Tt]est[Rr]esult*/
 [Bb]uild[Ll]og.*
 
+# CTest test results
+[Tt]esting/
+
 # NUNIT
 *.VisualState.xml
 TestResult.xml

.vscode/cspell.json

@@ -179,6 +179,13 @@
     "XSMB"
   ],
   "overrides": [
+    {
+      "filename": "CMakePresets.json",
+      "words": [
+        "ASAN",
+        "fsanitize"
+      ]
+    },
     {
       "filename": "**/eng/pipelines/templates/**/*.yml",
       "words": [

CMakePresets.json

@@ -166,6 +166,45 @@
         "BUILD_PERFORMANCE_TESTS": true
       }
     },
+    {
+      "name": "winhttp-transport",
+      "displayName": "Enable WinHTTP Transport",
+      "description": "Enable WinHTTP Transport (Hidden)",
+      "hidden": true,
+      "cacheVariables": {
+        "BUILD_TRANSPORT_WINHTTP": true
+      },
+      "condition": {
+        "type": "equals",
+        "lhs": "${hostSystemName}",
+        "rhs": "Windows"
+      }
+    },
+    {
+      "name": "curl-transport",
+      "displayName": "Enable CURL Transport",
+      "description": "Enable CURL Transport (Hidden)",
+      "hidden": true,
+      "cacheVariables": {
+        "BUILD_TRANSPORT_CURL": true
+      }
+    },
+    {
+      "name": "enable-address-sanitizer",
+      "displayName": "Enable Address Sanitizer",
+      "description": "Enable Address Sanitizer (Hidden). Note: ASAN can be extremely disruptive, when enabling ASAN, make sure you run your application under the debugger.",
+      "hidden": true,
+      "environment": {
+        "CFLAGS": "/fsanitize=address",
+        "CXXFLAGS": "/fsanitize=address"
+      },
+      "cacheVariables": { "DISABLE_AZURE_CORE_OPENTELEMETRY": true },
+      "condition": {
+        "type": "equals",
+        "lhs": "${hostSystemName}",
+        "rhs": "Windows"
+      }
+    },
     {
       "name": "x86-static-debug-tests",
       "displayName": "x86 Debug With Tests, static",
@@ -204,19 +243,13 @@
       "name": "x64-static-debug-tests-curl",
       "displayName": "x64 Debug With Tests, static, libcurl",
       "description": "Windows x64 Debug build with Tests configured",
-      "inherits": [ "x64-static", "debug-build", "enable-tests" ],
-      "cacheVariables": {
-        "BUILD_TRANSPORT_CURL": true
-      }
+      "inherits": [ "x64-static", "debug-build", "enable-tests", "curl-transport" ]
     },
     {
       "name": "x64-static-debug-tests-winhttp",
       "displayName": "x64 Debug With Tests, static, winhttp",
       "description": "Windows x64 Debug build with Tests configured",
-      "inherits": [ "x64-static", "debug-build", "enable-tests" ],
-      "cacheVariables": {
-        "BUILD_TRANSPORT_WINHTTP": true
-      }
+      "inherits": [ "x64-static", "debug-build", "enable-tests", "winhttp-transport" ]
     },
     {
       "name": "x64-static-debug-tests-OpenSSL111",
@@ -234,30 +267,24 @@
       "inherits": [
         "x86-static",
         "release-build",
-        "enable-tests"
-      ],
-      "cacheVariables": {
-        "BUILD_TRANSPORT_CURL": true
-      }
+        "enable-tests",
+        "curl-transport"
+      ]
     },
     {
       "name": "x86-release-tests",
       "displayName": "x86 Release With Tests (Note: Does not link because it sets BUILD_TRANSPORT_CUSTOM)",
-      "inherits": [ "x86", "release-build", "enable-tests" ],
+      "inherits": [ "x86", "release-build", "enable-tests", "curl-transport", "winhttp-transport" ],
       "cacheVariables": {
-        "BUILD_TRANSPORT_CURL": true,
-        "BUILD_TRANSPORT_CUSTOM": true,
-        "BUILD_TRANSPORT_WINHTTP": true
+        "BUILD_TRANSPORT_CUSTOM": true
       }
     },
     {
       "name": "x64-release-tests",
       "displayName": "x64 Release With Tests (Note: Does not link because it sets BUILD_TRANSPORT_CUSTOM)",
-      "inherits": [ "x64", "release-build", "enable-tests" ],
+      "inherits": [ "x64", "release-build", "enable-tests", "curl-transport", "winhttp-transport" ],
       "cacheVariables": {
-        "BUILD_TRANSPORT_CURL": true,
-        "BUILD_TRANSPORT_CUSTOM": true,
-        "BUILD_TRANSPORT_WINHTTP": true
+        "BUILD_TRANSPORT_CUSTOM": true
       }
     },
     {
@@ -268,21 +295,15 @@
         "debug-build",
         "enable-tests",
         "enable-samples",
-        "enable-perf"
-      ],
-      "cacheVariables": {
-        "BUILD_TRANSPORT_CURL": true,
-        "BUILD_TRANSPORT_WINHTTP": true
-      }
+        "enable-perf",
+        "winhttp-transport",
+        "curl-transport"
+      ]
     },
     {
       "name": "x86-msvc-static-debug-perftests",
       "displayName": "x86 MSVC Debug static With Perf Tests and samples",
-      "inherits": [ "x86-msvc-static", "debug-build", "enable-tests", "enable-perf", "enable-samples" ],
-      "cacheVariables": {
-        "BUILD_TRANSPORT_WINHTTP": true,
-        "BUILD_TRANSPORT_CURL": true
-      }
+      "inherits": [ "x86-msvc-static", "debug-build", "enable-tests", "enable-perf", "enable-samples", "curl-transport", "winhttp-transport" ]
     },
     {
       "name": "x64-static-release-perftests",

sdk/core/azure-core-amqp/CHANGELOG.md

@@ -4,10 +4,24 @@
 
 ### Features Added
 
+- AMQP moved from a polling model to an asynchronous model. 
+
 ### Breaking Changes
 
+- Removed the `QueueSend` API from `MessageSender` because it was not compatible with the new asynchronous model.
+- The new asynchronous model requires the user to call `Close()` on the `MessageSender` and `MessageReceiver` 
+to ensure operations have stabilized before destroying the object.
+- For connection listeners (primarily test scenarios), if you call `Open()` or `Listen()` on a connection, you MUST call `Close()` 
+before the connection is destroyed.
+- The `Connection::Close()` method no longer requires that the caller provide connection disconnect information.
+- The `Session::End()` method no longer requires that the caller provide session disconnect information.
+- Several asserts have been added which will force termination of the running application if invariants have not been met.
+
 ### Bugs Fixed
 
+- Several fixes related to the new asynchronous model. Ensures that message senders and receivers are always closed, 
+and that resources are released.
+
 ### Other Changes
 
 ## 1.0.0-beta.3 (2023-09-07)

sdk/core/azure-core-amqp/CMakeLists.txt

@@ -100,6 +100,7 @@ add_library(Azure::azure-core-amqp ALIAS azure-core-amqp)
 
 # coverage. Has no effect if BUILD_CODE_COVERAGE is OFF
 create_code_coverage(core azure-core-amqp azure-core-amqp-tests "tests?/*;samples?/*")
+create_code_coverage(core azure-core-amqp-uamqp azure-core-amqp-uamqp-tests "tests?/*;samples?/*")
 
 # cspell:words aziotsharedutil
 # uAMQP's headers require the manual addition of umock_c, azure_macro_utils_c, and aziotsharedutil to the target link libraries. 

sdk/core/azure-core-amqp/inc/azure/core/amqp/claims_based_security.hpp

@@ -67,7 +67,6 @@ namespace Azure { namespace Core { namespace Amqp { namespace _detail {
         std::string const& audience,
         std::string const& token,
         Context const& context = {});
-    void SetTrace(bool traceEnabled);
 
   private:
     std::shared_ptr<ClaimsBasedSecurityImpl> m_impl;

sdk/core/azure-core-amqp/inc/azure/core/amqp/common/async_operation_queue.hpp

@@ -4,6 +4,8 @@
 #pragma once
 
 #include <azure/core/context.hpp>
+#include <azure/core/diagnostics/logger.hpp>
+#include <azure/core/internal/diagnostics/log.hpp>
 
 #include <condition_variable>
 #include <list>
@@ -19,7 +21,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace Common { namespace
    * "CompleteOperation" which sets the result, and a consumer calls WaitForResult which reads from
    * the AsyncOperationQueue. WaitForResult will block until a result is available.
    */
-  template <typename... T> class AsyncOperationQueue {
+  template <typename... T> class AsyncOperationQueue final {
 
   public:
     AsyncOperationQueue() = default;
@@ -63,6 +65,58 @@ namespace Azure { namespace Core { namespace Amqp { namespace Common { namespace
       } while (true);
     }
 
+    /**
+     * @brief Wait for a result to be available.
+     *
+     * @param context The context to use for cancellation.
+     * @param pollers optional set of pollers to call.
+     * @return std::unique_ptr<std::tuple<T...>> The result.
+     *
+     * @remarks The pollers parameter is a TEST HOOK to allow test message receivers to interact
+     * with the message loop. in general clients should NOT provide a poller.
+     *
+     */
+    template <class... Poller>
+    std::unique_ptr<std::tuple<T...>> WaitForResult(Context const& context, Poller&... pollers)
+    {
+      // If the queue is not empty, return the first element.
+      do
+      {
+        {
+          std::unique_lock<std::mutex> lock(m_operationComplete);
+
+          if (!m_operationQueue.empty())
+          {
+            std::unique_ptr<std::tuple<T...>> rv;
+            rv = std::move(m_operationQueue.front());
+            m_operationQueue.pop_front();
+            return rv;
+          }
+
+          // There's nothing in the queue, wait until something is put into the queue.
+          // This will block until either something is put into the queue or the context is
+          // cancelled.
+          m_operationCondition.wait_for(
+              lock, std::chrono::milliseconds(100), [this, &context]() -> bool {
+                // If the context is cancelled, we should return immediately.
+                if (context.IsCancelled())
+                {
+                  return true;
+                }
+                return !m_operationQueue.empty();
+              });
+
+          if (context.IsCancelled())
+          {
+            return nullptr;
+          }
+        }
+        // Note: We need to call Poll() *outside* the lock because the poller is going to call the
+        // CompleteOperation function.
+        Poll(pollers...);
+      } while (true);
+    }
+
     // Clear any pending elements from the queue. This may be needed because some queued elements
     // may have ordering dependencies that need to be cleared before the object containing the queue
     // can be released.

sdk/core/azure-core-amqp/inc/azure/core/amqp/common/global_state.hpp

@@ -3,6 +3,11 @@
 
 #pragma once
 
+#include <list>
+#include <memory>
+#include <mutex>
+#include <thread>
+
 namespace Azure { namespace Core { namespace Amqp { namespace Common { namespace _detail {
 
   /**
@@ -15,11 +20,35 @@ namespace Azure { namespace Core { namespace Amqp { namespace Common { namespace
    *
    */
 
+  class Pollable {
+  public:
+    virtual void Poll() = 0;
+    virtual ~Pollable() = default;
+  };
   class GlobalStateHolder final {
     GlobalStateHolder();
     ~GlobalStateHolder();
 
+    std::list<std::shared_ptr<Pollable>> m_pollables;
+    std::mutex m_pollablesMutex;
+    std::thread m_pollingThread;
+    bool m_stopped{false};
+
   public:
     static GlobalStateHolder* GlobalStateInstance();
+
+    GlobalStateHolder(GlobalStateHolder const&) = delete;
+    GlobalStateHolder& operator=(GlobalStateHolder const&) = delete;
+
+    GlobalStateHolder(GlobalStateHolder&&) = delete;
+    GlobalStateHolder& operator=(GlobalStateHolder&&) = delete;
+
+    void AddPollable(std::shared_ptr<Pollable> pollable);
+
+    void RemovePollable(std::shared_ptr<Pollable> pollable)
+    {
+      std::lock_guard<std::mutex> lock(m_pollablesMutex);
+      m_pollables.remove(pollable);
+    }
   };
 }}}}} // namespace Azure::Core::Amqp::Common::_detail

sdk/core/azure-core-amqp/inc/azure/core/amqp/connection.hpp

@@ -38,6 +38,10 @@ namespace Azure { namespace Core { namespace Amqp { namespace Tests {
   class TestMessages_SenderOpenClose_Test;
   class TestMessages_TestLocalhostVsTls_Test;
   class TestMessages_SenderSendAsync_Test;
+  class TestMessages_SenderOpenClose_Test;
+  class TestMessages_ReceiverOpenClose_Test;
+  class TestMessages_ReceiverReceiveAsync_Test;
+
 }}}} // namespace Azure::Core::Amqp::Tests
 #endif // TESTING_BUILD
 #if defined(SAMPLES_BUILD)
@@ -153,6 +157,8 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     Error,
   };
 
+  std::ostream& operator<<(std::ostream& stream, ConnectionState const value);
+
   class Connection;
 
   /** @brief The ConnectionEvents interface defines a series of events triggered on a connection
@@ -336,6 +342,9 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
      * will be opened implicitly by a Session object derived from the connection. It primarily
      * exists as a test hook.
      *
+     * @remarks If you call Open() or Listen(), then you MUST call Close() when you are done with
+     * the connection, BEFORE destroying it.
+     *
      */
     void Open();
 
@@ -348,6 +357,8 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
      * will be opened implicitly by a Session object derived from the connection. It primarily
      * exists as a test hook.
      *
+     * @remarks If you call Open() or Listen(), then you MUST call Close() when you are done with
+     * the connection, BEFORE destroying it.
      */
     void Listen();
 
@@ -361,11 +372,13 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
      * will be closed implicitly by a Session object derived from the connection. It primarily
      * exists as a test hook.
      *
+     * @remarks If you have NOT called Open() or Listen(), then calling this is an error.
+     *
      */
     void Close(
-        std::string const& condition,
-        std::string const& description,
-        Models::AmqpValue info);
+        std::string const& condition = {},
+        std::string const& description = {},
+        Models::AmqpValue info = {});
 
     /** @brief Gets host configured by the connection.
      *
@@ -450,6 +463,8 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     friend class Azure::Core::Amqp::Tests::TestMessages_SenderOpenClose_Test;
     friend class Azure::Core::Amqp::Tests::TestMessages_TestLocalhostVsTls_Test;
     friend class Azure::Core::Amqp::Tests::TestMessages_SenderSendAsync_Test;
+    friend class Azure::Core::Amqp::Tests::TestMessages_SenderOpenClose_Test;
+
 #endif // TESTING_BUILD
 #if SAMPLES_BUILD
     friend int LocalServerSample::LocalServerSampleMain();

sdk/core/azure-core-amqp/inc/azure/core/amqp/message_receiver.hpp

@@ -32,6 +32,7 @@ namespace Azure { namespace Core { namespace Amqp { namespace _internal {
     Closing,
     Error,
   };
+  std::ostream& operator<<(std::ostream& stream, _internal::MessageReceiverState const& state);
 
   enum class ReceiverSettleMode
   {