AsyncSystem.h

#pragma once
 
#include "Impl/ContinuationFutureType.h"
#include "Impl/RemoveFuture.h"
#include "Impl/WithTracing.h"
#include "Impl/cesium-async++.h"
 
#include <CesiumAsync/Future.h>
#include <CesiumAsync/Library.h>
#include <CesiumAsync/Promise.h>
#include <CesiumAsync/ThreadPool.h>
#include <CesiumUtility/Tracing.h>
#include <CesiumUtility/transformTuple.h>
 
#include <memory>
#include <type_traits>
 
namespace CesiumAsync {
class ITaskProcessor;
 
class AsyncSystem;

class CESIUMASYNC_API AsyncSystem final {
public:
  AsyncSystem(const std::shared_ptr<ITaskProcessor>& pTaskProcessor) noexcept;

  template <typename T, typename Func> Future<T> createFuture(Func&& f) const {
    std::shared_ptr<async::event_task<T>> pEvent =
        std::make_shared<async::event_task<T>>();
 
    Promise<T> promise(this->_pSchedulers, pEvent);
 
    try {
      f(promise);
    } catch (...) {
      promise.reject(std::current_exception());
    }
 
    return Future<T>(this->_pSchedulers, pEvent->get_task());
  }

  template <typename T> Promise<T> createPromise() const {
    return Promise<T>(
        this->_pSchedulers,
        std::make_shared<async::event_task<T>>());
  }

  template <typename Func>
  CesiumImpl::ContinuationFutureType_t<Func, void>
  runInWorkerThread(Func&& f) const {
    static const char* tracingName = "waiting for worker thread";
 
    CESIUM_TRACE_BEGIN_IN_TRACK(tracingName);
 
    return CesiumImpl::ContinuationFutureType_t<Func, void>(
        this->_pSchedulers,
        async::spawn(
            this->_pSchedulers->workerThread.immediate,
            CesiumImpl::WithTracing<void>::end(
                tracingName,
                std::forward<Func>(f))));
  }

  template <typename Func>
  CesiumImpl::ContinuationFutureType_t<Func, void>
  runInMainThread(Func&& f) const {
    static const char* tracingName = "waiting for main thread";
 
    CESIUM_TRACE_BEGIN_IN_TRACK(tracingName);
 
    return CesiumImpl::ContinuationFutureType_t<Func, void>(
        this->_pSchedulers,
        async::spawn(
            this->_pSchedulers->mainThread.immediate,
            CesiumImpl::WithTracing<void>::end(
                tracingName,
                std::forward<Func>(f))));
  }

  template <typename Func>
  CesiumImpl::ContinuationFutureType_t<Func, void>
  runInThreadPool(const ThreadPool& threadPool, Func&& f) const {
    static const char* tracingName = "waiting for thread pool";
 
    CESIUM_TRACE_BEGIN_IN_TRACK(tracingName);
 
    return CesiumImpl::ContinuationFutureType_t<Func, void>(
        this->_pSchedulers,
        async::spawn(
            threadPool._pScheduler->immediate,
            CesiumImpl::WithTracing<void>::end(
                tracingName,
                std::forward<Func>(f))));
  }

  template <typename T>
  using AllValueType =
      std::conditional_t<std::is_void_v<T>, void, std::vector<T>>;

  template <typename T>
  Future<AllValueType<T>> all(std::vector<Future<T>>&& futures) const {
    return this->all<T, Future<T>>(
        std::forward<std::vector<Future<T>>>(futures));
  }

  template <typename T>
  Future<AllValueType<T>> all(std::vector<SharedFuture<T>>&& futures) const {
    return this->all<T, SharedFuture<T>>(
        std::forward<std::vector<SharedFuture<T>>>(futures));
  }

  template <typename... Futures>
  using AllTupleType =
      std::tuple<typename CesiumImpl::RemoveFuture<Futures>::type...>;

  template <typename... Futures>
  Future<AllTupleType<Futures...>> all(Futures&&... futures) const {
    // Helper function so that we can get the `_task` member of each future and
    // pass as a variadic pack.
    auto getTaskFromFuture = [](auto&& future) {
      return std::move(future._task);
    };
 
    using AggregateTaskType = std::tuple<
        async::task<typename CesiumImpl::RemoveFuture<Futures>::type>...>;
 
    async::task<AllTupleType<Futures...>> task =
        async::when_all(getTaskFromFuture(std::forward<Futures>(futures))...)
            .then(async::inline_scheduler(), [](AggregateTaskType&& tasks) {
              return CesiumUtility::transformTuple(
                  std::move(tasks),
                  [](auto&& task) { return task.get(); });
            });
 
    return Future<AllTupleType<Futures...>>(
        this->_pSchedulers,
        std::move(task));
  }

  template <typename T> Future<T> createResolvedFuture(T&& value) const {
    return Future<T>(
        this->_pSchedulers,
        async::make_task<T>(std::forward<T>(value)));
  }

  Future<void> createResolvedFuture() const {
    return Future<void>(this->_pSchedulers, async::make_task());
  }

  void dispatchMainThreadTasks();

  bool dispatchOneMainThreadTask();

  using MainThreadScope = CesiumImpl::ImmediateScheduler<
      CesiumImpl::QueuedScheduler>::SchedulerScope;

  MainThreadScope enterMainThread() const;

  ThreadPool createThreadPool(int32_t numberOfThreads) const;

  bool operator==(const AsyncSystem& rhs) const noexcept;

  bool operator!=(const AsyncSystem& rhs) const noexcept;
 
private:
  // Common implementation of 'all' for both Future and SharedFuture.
  template <typename T, typename TFutureType>
  Future<AllValueType<T>> all(std::vector<TFutureType>&& futures) const {
    using TTaskType = decltype(TFutureType::_task);
    std::vector<TTaskType> tasks;
    tasks.reserve(futures.size());
 
    for (auto it = futures.begin(); it != futures.end(); ++it) {
      tasks.emplace_back(std::move(it->_task));
    }
 
    futures.clear();
 
    async::task<AllValueType<T>> task =
        async::when_all(tasks.begin(), tasks.end())
            .then(
                async::inline_scheduler(),
                [](std::vector<TTaskType>&& tasks) {
                  if constexpr (std::is_void_v<T>) {
                    // Tasks return void. "Get" each task so that error
                    // information is propagated.
                    for (auto it = tasks.begin(); it != tasks.end(); ++it) {
                      it->get();
                    }
                  } else {
                    // Get all the results. If any tasks rejected, we'll bail
                    // with an exception.
                    std::vector<T> results;
                    results.reserve(tasks.size());
 
                    for (auto it = tasks.begin(); it != tasks.end(); ++it) {
                      results.emplace_back(std::move(it->get()));
                    }
                    return results;
                  }
                });
    return Future<AllValueType<T>>(this->_pSchedulers, std::move(task));
  }
 
  std::shared_ptr<CesiumImpl::AsyncSystemSchedulers> _pSchedulers;
 
  template <typename T> friend class Future;
};
} // namespace CesiumAsync