basic_task.hpp

Go to the documentation of this file.

#pragma once
#include "itask.hpp"
#include "task_canceled_exception.hpp"
#include "task_creation_options.hpp"
#include "task_object.hpp"
#include "task_status.hpp"
#include "../auto_reset_event.hpp"
#include "../lock.hpp"
#include "../thread_pool.hpp"
#include "../../diagnostics/stopwatch.hpp"
#include "../../helpers/throw_helper.hpp"
#include "../../exception_services/exception_dispatch_info.hpp"
#include "../../action.hpp"
#include "../../aggregate_exception.hpp"
#include "../../func.hpp"
#include "../../iasync_result.hpp"
#include "../../new_sptr.hpp"
#include "../../optional.hpp"
#include "../../ref.hpp"
#include "../../scope_exit.hpp"
#include "../../sptr.hpp"
#include <coroutine>

namespace xtd {
  namespace threading {
    namespace tasks {
      template<typename result_t = void>
      class task;
      
      template<typename result_t>
      struct task_awaiter;
 
      class task_factory;

      template<typename result_t = void>
      class basic_task : public xtd::threading::tasks::task_object, public xtd::threading::tasks::itask, public xtd::iasync_result {
      public:
        struct yield_awaiter;
        
        
        static constexpr xtd::usize wait_timeout = xtd::usize_object::max_value;
        
        basic_task(const xtd::func<result_t>& func) {
          data_->func = func;
        }
        basic_task(const xtd::func<result_t>& func, const xtd::threading::cancellation_token& cancellation_token) {
          data_->func = func;
          data_->cancellation_token = cancellation_token;
        }
        basic_task(const xtd::func<result_t, const xtd::any_object&>& func, const xtd::any_object& state) {
          data_->parameterized_func = func;
          data_->state = &state;
        }
        basic_task(const xtd::func<result_t>& func, const xtd::any_object& state, const xtd::threading::cancellation_token& cancellation_token) {
          data_->func = func;
          data_->state = &state;
          data_->cancellation_token = cancellation_token;
        }
        
        template<typename create_result_t>
        basic_task(const xtd::func<create_result_t>& func) {
          data_->func = func;
        }
        template<typename create_result_t>
        basic_task(const xtd::func<create_result_t>& func, const xtd::threading::cancellation_token& cancellation_token) {
          data_->func = func;
          data_->cancellation_token = cancellation_token;
        }
        template<typename create_result_t>
        basic_task(const xtd::func<create_result_t, const xtd::any_object&>& func, const xtd::any_object& state) {
          data_->parameterized_func = func;
          data_->state = &state;
        }
        template<typename create_result_t>
        basic_task(const xtd::func<create_result_t, const xtd::any_object&>& func, const xtd::any_object& state, const xtd::threading::cancellation_token& cancellation_token) {
          data_->parameterized_func = func;
          data_->state = &state;
          data_->cancellation_token = cancellation_token;
        }
 
        
        
        [[nodiscard]] auto async_state() const noexcept -> xtd::any_object override {return data_->async_state;}
        [[nodiscard]] auto creation_options() const noexcept -> xtd::threading::tasks::task_creation_options {return data_->creation_options;}
        [[nodiscard]] auto exception() const noexcept -> xtd::ref<xtd::exception> {return data_->exception.exception_captured() ? xtd::ref<xtd::exception> {*data_->exception.source_exception()} : xtd::ref<xtd::exception> {xtd::null};}
        [[nodiscard]] auto id() const noexcept -> xtd::usize override {return data_->id;}
        [[nodiscard]] auto is_canceled() const noexcept -> bool {return data_->status == xtd::threading::tasks::task_status::canceled;}
        [[nodiscard]] auto is_completed() const noexcept -> bool override {return is_faulted() || is_canceled() || data_->status == xtd::threading::tasks::task_status::ran_to_completion;}
        [[nodiscard]] auto is_faulted() const noexcept -> bool {return data_->status == xtd::threading::tasks::task_status::faulted;}
        [[nodiscard]] auto status() const noexcept -> xtd::threading::tasks::task_status {return data_->status;}
        
        
        [[nodiscard]] static auto completed_task() -> task<result_t>;
        [[nodiscard]] static auto current_id() noexcept -> xtd::usize {return current_id_;}
        [[nodiscard]] static auto factory() noexcept -> const xtd::threading::tasks::task_factory&;
        
        
        auto continue_with(xtd::action<> continuation) -> void {
          auto call_now = false;
          
          lock_(data_->sync_root) {
            if (is_completed()) call_now = true;
            else data_->continuation = continuation;
          }
          
          if (call_now) continuation();
        }
        
        [[noreturn]] auto rethrow_exception() -> void {
          if (is_faulted () && data_->exception) throw *data_->exception.source_exception();
          xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::invalid_operation, "This task is not faulted.");
        }
        
        auto run_synchronously() -> void {
          if (is_completed()) xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::invalid_operation, "Start may not be called on a task that has completed.");
          data_->status = xtd::threading::tasks::task_status::waiting_to_run;
          data_->task_proc(data_->state, false);
        }
        
        auto start() -> void override {
          if (is_completed()) xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::invalid_operation, "Start may not be called on a task that has completed.");
          data_->status = xtd::threading::tasks::task_status::waiting_for_activation;
          thread_pool::register_wait_for_single_object(data_->start_event, data_->task_proc, *data_->state, xtd::threading::timeout::infinite, true);
          data_->status = xtd::threading::tasks::task_status::waiting_to_run;
          data_->start_event.set();
        }
        
        auto wait() -> void override {wait(xtd::threading::timeout::infinite);}
        auto wait(xtd::int32 milliseconds_timeout) -> bool override {
          auto cancellation_token = xtd::threading::cancellation_token {};
          return wait(milliseconds_timeout, data_->cancellation_token ? *data_->cancellation_token : cancellation_token);
        }

        auto wait(xtd::int32 milliseconds_timeout, xtd::threading::cancellation_token& cancellation_token) -> bool override {
          auto& cancellation_token_wait_handle = cancellation_token.wait_handle();
          auto result = xtd::threading::wait_handle::wait_any(milliseconds_timeout, &cancellation_token_wait_handle, &data_->end_event);
          if (result == 0) xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::task_canceled);
          if (data_->status == xtd::threading::tasks::task_status::faulted) rethrow_exception();
          return result == 1;
        }

        auto wait(const xtd::time_span& timeout) -> bool override {return wait(xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()));}
        auto wait(const xtd::time_span& timeout, xtd::threading::cancellation_token& cancellation_token) -> bool override {return wait(xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()), cancellation_token);}
        
        
        [[nodiscard]] static auto from_cancelation(const xtd::threading::cancellation_token& cancellation_token) -> task<result_t>;
        
        template<typename from_exception_t>
        [[nodiscard]] static auto from_exception(from_exception_t exception) -> task<result_t>;
        
        [[nodiscard]] static auto delay(const xtd::time_span& delay) -> task<>;
        [[nodiscard]] static auto delay(const xtd::time_span& delay, const xtd::threading::cancellation_token& cancellation_token) -> task<>;
        [[nodiscard]] static auto delay(xtd::int32 milliseconds_delay) -> task<>;
        [[nodiscard]] static auto delay(xtd::int32 milliseconds_delay, const xtd::threading::cancellation_token& cancellation_token) -> task<>;
 
        [[nodiscard]] static auto run(const xtd::func<result_t>& func) -> task<result_t>;
        [[nodiscard]] static auto run(const xtd::func<result_t>& func, const xtd::threading::cancellation_token& cancellation_token) -> task<result_t>;
        [[nodiscard]] static auto run(const xtd::func<result_t, const xtd::any_object&>& func, const xtd::any_object& state) -> task<result_t>;
        [[nodiscard]] static auto run(const xtd::func<result_t, const xtd::any_object&>& func, const xtd::any_object& state, const xtd::threading::cancellation_token& cancellation_token) -> task<result_t>;
        
        template<typename collection_t>
        static auto wait_all(const collection_t& tasks) -> bool {return wait_all(tasks, xtd::threading::timeout::infinite);}
 
        template<typename collection_t>
        static auto wait_all(const collection_t& tasks, xtd::int32 milliseconds_timeout) -> bool {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(const_cast<decltype(&task)>(&task));
          return wait_all(array<itask*> {task_pointers}, milliseconds_timeout);
        }
 
        template<typename collection_t>
        static auto wait_all(const collection_t& tasks, const xtd::time_span& timeout) -> bool {return wait_all(tasks, xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()));}
 
        template<typename collection_t>
        static xtd::usize wait_any(const collection_t& tasks) {return wait_any(tasks, timeout::infinite);}
        
        template<typename collection_t>
        static xtd::usize wait_any(const collection_t& tasks, int32 milliseconds_timeout) {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(const_cast<decltype(&task)>(&task));
          return wait_any(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        
        template<typename collection_t>
        static xtd::usize wait_any(const collection_t& tasks, const time_span& timeout) {return wait_any(tasks, as<int32>(timeout.total_milliseconds_duration().count()));}
 
        template<typename ...items_t>
        static auto when_all(items_t&&... items) -> task<>;
 
        template<typename ...items_t>
        static auto when_any(items_t&&... items) -> task<xtd::usize>;
        
        template<typename ...items_t>
        static auto wait_all(items_t... items) -> bool {return wait_all(xtd::threading::timeout::infinite, items...);}
        template<typename ...items_t>
        static auto wait_all(const xtd::time_span& timeout, items_t... items) -> bool {return wait_all(xtd::as<xtd::int32>(timeout.total_milliseconds()), items...);}
        template<typename ...items_t>
        static auto wait_all(xtd::int32 milliseconds_timeout, items_t... items) -> bool {
          auto task_pointers = std::vector<itask*> {};
          fill_task_pointers(task_pointers, items...);
          return wait_all(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        template<typename item_t>
        static auto wait_all(const std::initializer_list<item_t>& tasks) -> bool {return wait_all(tasks, timeout::infinite);}
        template<typename item_t>
        static auto wait_all(const std::initializer_list<item_t>& tasks, xtd::int32 milliseconds_timeout) -> bool {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(const_cast<item_t*>(&task));
          return wait_all(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        template<typename item_t>
        static auto wait_all(const std::initializer_list<item_t>& tasks, const xtd::time_span& timeout) -> bool {return wait_all(tasks, as<int32>(timeout.total_milliseconds_duration().count()));}
        static auto wait_all(const std::initializer_list<xtd::sptr<itask>>& tasks) -> bool {return wait_all(tasks, xtd::threading::timeout::infinite);}
        static auto wait_all(const std::initializer_list<xtd::sptr<itask>>& tasks, xtd::int32 milliseconds_timeout) -> bool {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(task.get());
          return wait_all(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        static auto wait_all(const std::initializer_list<xtd::sptr<itask>>& tasks, const xtd::time_span& timeout) -> bool {return wait_all(tasks, xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()));}
        static auto wait_all(const std::initializer_list<xtd::uptr<itask>>& tasks) -> bool {return wait_all(tasks, xtd::threading::timeout::infinite);}
        static auto wait_all(const std::initializer_list<xtd::uptr<itask>>& tasks, xtd::int32 milliseconds_timeout) -> bool {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(task.get());
          return wait_all(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        static auto wait_all(const std::initializer_list<xtd::uptr<itask>>& tasks, const xtd::time_span& timeout) -> bool {return wait_all(tasks, xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()));}
        static auto wait_all(const xtd::array<xtd::sptr<itask>>& tasks) -> bool {return wait_all(tasks, timeout::infinite);}
        static auto wait_all(const xtd::array<xtd::sptr<itask>>& tasks, xtd::int32 milliseconds_timeout) -> bool {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(task.get());
          return wait_all(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        static auto wait_all(const xtd::array<xtd::sptr<itask>>& tasks, const xtd::time_span& timeout) -> bool {return wait_all(tasks, xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()));}
        static auto wait_all(const xtd::array<xtd::uptr<itask>>& tasks) -> bool {return wait_all(tasks, timeout::infinite);}
        static auto wait_all(const xtd::array<xtd::uptr<itask>>& tasks, xtd::int32 milliseconds_timeout) -> bool {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(task.get());
          return wait_all(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        static auto wait_all(const xtd::array<xtd::uptr<itask>>& tasks, const xtd::time_span& timeout) -> bool {return wait_all(tasks, xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()));}
        static auto wait_all(const xtd::array<itask*>& tasks, xtd::int32 milliseconds_timeout) -> bool {
          if (milliseconds_timeout < timeout::infinite) xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::argument);
          
          if (milliseconds_timeout == timeout::infinite) {
            for (auto task : tasks)
              task->wait();
            return true;
          }
          
          auto sw = xtd::diagnostics::stopwatch::start_new();
          for (auto& task : tasks)
            if (sw.elapsed_milliseconds() > milliseconds_timeout || task->wait(milliseconds_timeout - xtd::as<xtd::int32>(sw.elapsed_milliseconds()))) return false;
          return true;
        }
 
        template<typename ...items_t>
        static auto wait_any(items_t... items) -> xtd::usize {return wait_any(xtd::threading::timeout::infinite, items...);}
        template<typename ...items_t>
        static auto wait_any(const xtd::time_span& timeout, items_t... items) -> xtd::usize {return wait_any(xtd::as<xtd::int32>(timeout.total_milliseconds()), items...);}
        template<typename ...items_t>
        static auto wait_any(xtd::int32 milliseconds_timeout, items_t... items) -> xtd::usize {
          auto task_pointers = std::vector<itask*> {};
          fill_task_pointers(task_pointers, items...);
          return wait_any(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        template<typename item_t>
        static auto wait_any(const std::initializer_list<item_t>& tasks) -> xtd::usize {return wait_any(tasks, timeout::infinite);}
        template<typename item_t>
        static auto wait_any(const std::initializer_list<item_t>& tasks, xtd::int32 milliseconds_timeout) -> xtd::usize {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(const_cast<item_t*>(&task));
          return wait_any(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        template<typename item_t>
        static auto wait_any(const std::initializer_list<item_t>& tasks, const xtd::time_span& timeout) -> xtd::usize {return wait_any(tasks, as<int32>(timeout.total_milliseconds_duration().count()));}
        static auto wait_any(const std::initializer_list<xtd::sptr<itask>>& tasks) -> xtd::usize {return wait_any(tasks, xtd::threading::timeout::infinite);}
        static auto wait_any(const std::initializer_list<xtd::sptr<itask>>& tasks, xtd::int32 milliseconds_timeout) -> xtd::usize {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(task.get());
          return wait_any(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        static auto wait_any(const std::initializer_list<xtd::sptr<itask>>& tasks, const xtd::time_span& timeout) -> xtd::usize {return wait_any(tasks, xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()));}
        static auto wait_any(const std::initializer_list<xtd::uptr<itask>>& tasks) -> xtd::usize {return wait_any(tasks, xtd::threading::timeout::infinite);}
        static auto wait_any(const std::initializer_list<xtd::uptr<itask>>& tasks, xtd::int32 milliseconds_timeout) -> xtd::usize {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(task.get());
          return wait_any(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        static auto wait_any(const std::initializer_list<xtd::uptr<itask>>& tasks, const xtd::time_span& timeout) -> xtd::usize {return wait_any(tasks, xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()));}
        static auto wait_any(const xtd::array<xtd::sptr<itask>>& tasks) -> xtd::usize {return wait_any(tasks, timeout::infinite);}
        static auto wait_any(const xtd::array<xtd::sptr<itask>>& tasks, xtd::int32 milliseconds_timeout) -> xtd::usize {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(task.get());
          return wait_any(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        static auto wait_any(const xtd::array<xtd::sptr<itask>>& tasks, const xtd::time_span& timeout) -> xtd::usize {return wait_any(tasks, xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()));}
        static auto wait_any(const xtd::array<xtd::uptr<itask>>& tasks) -> xtd::usize {return wait_any(tasks, timeout::infinite);}
        static auto wait_any(const xtd::array<xtd::uptr<itask>>& tasks, xtd::int32 milliseconds_timeout) -> xtd::usize {
          auto task_pointers = std::vector<itask*> {};
          for (auto& task : tasks)
            task_pointers.push_back(task.get());
          return wait_any(xtd::array<itask*> {task_pointers}, milliseconds_timeout);
        }
        static auto wait_any(const xtd::array<xtd::uptr<itask>>& tasks, const xtd::time_span& timeout) -> xtd::usize {return wait_any(tasks, xtd::as<xtd::int32>(timeout.total_milliseconds_duration().count()));}
        static auto wait_any(const xtd::array<itask*>& tasks, xtd::int32 milliseconds_timeout) -> xtd::usize {
          if (milliseconds_timeout < xtd::threading::timeout::infinite) xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::argument);
          
          auto sleep_duration = 1;
          const auto max_sleep = 10;
          auto sw = xtd::diagnostics::stopwatch::start_new();
          
          while (milliseconds_timeout == xtd::threading::timeout::infinite || sw.elapsed_milliseconds() <= milliseconds_timeout) {
            for (auto index = xtd::usize {0}; index < tasks.length(); ++index)
              if (tasks[index]->wait(0)) return index;
 
            xtd::threading::thread::sleep(sleep_duration);
            sleep_duration = std::min(sleep_duration + 1, max_sleep);
          }
          
          return wait_timeout;
        }
 
        static auto yield() -> task<result_t>;
        
      private:
        template<typename task_result_t>
        friend class task;
        template<typename batic_task_result_t>
        friend class batic_task;
        
        basic_task() = default;
        basic_task(basic_task&&) = default;
        basic_task(const basic_task&) = default;
        auto operator=(basic_task&&) -> basic_task& = default;
        auto operator=(const basic_task&) -> basic_task& = default;
        
        [[nodiscard]] auto async_wait_handle() noexcept -> xtd::threading::wait_handle& override {return data_->async_event;}
        [[nodiscard]] auto completed_synchronously() const noexcept -> bool override {return false;}
 
        template<typename item_t, typename ...items_t>
        static auto fill_task_pointers(std::vector<itask*>& itask_pointer, item_t& first, items_t& ... rest) -> void {
          itask_pointer.push_back(&first);
          fill_task_pointers(itask_pointer, rest...);
        }
        template<typename item_t>
        static auto fill_task_pointers(std::vector<itask*>& itask_pointer, item_t& item) -> void {
          itask_pointer.push_back(&item);
        }
 
        struct data {
          using result_type = std::conditional_t<std::is_same_v<result_t, void>, std::uint8_t, result_t>;
          
          xtd::async_callback async_callback;
          xtd::threading::manual_reset_event async_event;
          xtd::any_object async_state;
          xtd::optional<xtd::threading::cancellation_token> cancellation_token;
          xtd::action<> continuation;
          xtd::threading::tasks::task_creation_options creation_options = xtd::threading::tasks::task_creation_options::none;
          const xtd::any_object empty_state;
          xtd::threading::auto_reset_event end_event;
          xtd::exception_services::exception_dispatch_info exception;
          xtd::func<result_t> func;
          xtd::usize id = generate_id();
          xtd::usize previous_current_id;
          xtd::func<result_t, const xtd::any_object&> parameterized_func;
          result_type result;
          const xtd::any_object* state = &empty_state;
          xtd::threading::auto_reset_event start_event;
          xtd::threading::tasks::task_status status = xtd::threading::tasks::task_status::created;
          xtd::object sync_root;
          
          xtd::threading::wait_or_timer_callback task_proc {delegate_(const xtd::any_object& state, bool timed_out) {
            previous_current_id = current_id_;
            current_id_ = id;
            
            scope_exit_ {
              end_event.set();
              if (!continuation.is_empty()) continuation();
              current_id_ = previous_current_id;
            };
            
            status = xtd::threading::tasks::task_status::running;
            try {
              if (cancellation_token && cancellation_token->wait_handle().wait_one(0)) xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::task_canceled);
              if constexpr(std::is_same_v<result_t, void>) {
                if (!func.is_empty()) func();
                else if (!parameterized_func.is_empty()) parameterized_func(state);
              } else {
                if (!func.is_empty()) result = func();
                else if (!parameterized_func.is_empty()) result = parameterized_func(state);
              }
              if (cancellation_token && cancellation_token->wait_handle().wait_one(0)) xtd::helpers::throw_helper::throws(xtd::helpers::exception_case::task_canceled);
              status = xtd::threading::tasks::task_status::ran_to_completion;
            } catch (const xtd::threading::tasks::task_canceled_exception& e) {
              exception = xtd::exception_services::exception_dispatch_info::capture(e);
              status = xtd::threading::tasks::task_status::canceled;
            } catch (const xtd::exception& e) {
              exception = xtd::exception_services::exception_dispatch_info::capture(e);
              status = xtd::threading::tasks::task_status::faulted;
            } catch (const std::exception& e) {
              exception = xtd::exception_services::exception_dispatch_info::capture(xtd::exception {e.what()});
              status = xtd::threading::tasks::task_status::faulted;
            } catch (...) {
              exception = xtd::exception_services::exception_dispatch_info::capture(xtd::exception {"Unknown exception"});
              status = xtd::threading::tasks::task_status::faulted;
            }
          }};
        };
        
        xtd::sptr<data> data_ = xtd::new_sptr<data>();
      };
    }
  }
}