async_operation_queue.hpp

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.
 
#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>
#include <mutex>
#include <thread>
#include <tuple>
 
namespace Azure { namespace Core { namespace Amqp { namespace Common { namespace _internal {
 
  template <typename... T> class AsyncOperationQueue final {
  public:
    AsyncOperationQueue() = default;
    ~AsyncOperationQueue() = default;
 
    AsyncOperationQueue(const AsyncOperationQueue&) = delete;
    AsyncOperationQueue& operator=(const AsyncOperationQueue&) = delete;
 
    AsyncOperationQueue(AsyncOperationQueue&&) = default;
    AsyncOperationQueue& operator=(AsyncOperationQueue&&) = default;
 
    void CompleteOperation(T... operationParameters)
    {
      std::unique_lock<std::mutex> lock(m_operationComplete);
      m_operationQueue.push_back(std::make_unique<std::tuple<T...>>(
          std::make_tuple<T...>(std::forward<T>(operationParameters)...)));
      lock.unlock();
      m_operationCondition.notify_one();
    }
 
    template <class... Poller>
    std::unique_ptr<std::tuple<T...>> WaitForPolledResult(
        Context const& context,
        Poller&... pollers)
    {
      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;
          }
          if (context.IsCancelled())
          {
            return nullptr;
          }
        }
        std::this_thread::yield();
 
        // Note: We need to call Poll() *outside* the lock because the poller is going to call the
        // CompleteOperation function.
        Poll(pollers...);
      } while (true);
    }
 
    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);
    }
 
    std::unique_ptr<std::tuple<T...>> TryWaitForResult()
    {
      // If the queue is not empty, return the first element.
      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;
      }
      return nullptr;
    }
 
    // 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.
    void Clear()
    {
      std::unique_lock<std::mutex> lock(m_operationComplete);
      m_operationQueue.clear();
    }
 
  private:
    std::mutex m_operationComplete;
    std::condition_variable m_operationCondition;
    std::list<std::unique_ptr<std::tuple<T...>>> m_operationQueue;
 
    void Poll() {}
 
    template <class PT, class... Ts> void Poll(PT& first, Ts&... rest)
    {
      first.Poll();
      Poll(rest...);
    }
  };
}}}}} // namespace Azure::Core::Amqp::Common::_internal