5. spawn_future

最終更新日時(UTC):
が更新

履歴 編集

customization point object
<execution>

std::execution::spawn_future(C++26)

namespace std::execution {
  struct spawn_future_t { unspecified };
  inline constexpr spawn_future_t spawn_future{};
}

概要

spawn_futureは、入力Senderに対して非同期トークンスコープへの関連付けを試み、成功時に入力Senderを早期開始(eagerly start)させるSenderアダプタである。

戻り値Senderの接続(connect)および開始(start)により、早期開始した入力Senderの戻り値を用いて値完了、もしくは入力Senderが開始していなければset_stoppedで完了する。

効果

説明用の式sndr, token, envを下記の通り定義する。

sender<Sndr>, scope_token<Token>, queryable<Env>のいずれかを満たさないとき、呼び出し式spawn_future(sndr, token, env)は不適格となる。

そうでなければ、呼び出し式spawn_future(sndr, token, env)に対して、説明用の式new_senderをtoken.wrap(sndr)とし、allocとsenvを次の通りとする。

呼び出し式spawn_future(sndr, token, env)は、次の効果をもつ。

  • allocを用いてメモリ確保し、alloc, token.wrap(sndr), token, senvからdecltype(spawn-future-state(alloc, token.wrap(sndr), token, senv))型のオブジェクトsを構築する。いずれかのオブジェクト構築・破棄時に例外送出されたときは、確保されたメモリが解放される。
  • 下記を満たすunique_ptrの特殊化型のオブジェクトuを構築する。
  • make-sender(spawn_future, std::move(u))を返す。

呼び出し式spawn_future(sndr, token)は、式spawn_future(sndr, token, execution::env<>())と等価である。

Senderアルゴリズムタグ spawn_future

Senderアルゴリズム動作説明用のクラステンプレートimpls-forに対して、下記の特殊化が定義される。

namespace std::execution {
  template<>
  struct impls-for<spawn_future_t> : default-impls {
    static constexpr auto start = see below;      // exposition only
    static constexpr auto get-state = see below;  // exposition only
  };
}

impls-for<spawn_future_t>::startメンバは、下記ラムダ式と等価な関数呼び出し可能なオブジェクトで初期化される。

[](auto& state, auto&) noexcept -> void {
  state.run();
}

impls-for<spawn_future_t>::get-stateメンバは、下記ラムダ式と等価な関数呼び出し可能なオブジェクトで初期化される。

[]<class Sndr, class Rcvr>(Sndr sndr, Rcvr& rcvr) noexcept {
  auto& [_, data] = sndr;
  using state_ptr = remove_cvref_t<decltype(data)>;
  return future-operation<state_ptr, Rcvr>(std::move(data), std::move(rcvr));
}

説明専用エンティティ

クラスtry-cancelable

namespace std::execution {
  struct try-cancelable {                   // exposition only
    virtual void try-cancel() noexcept = 0; // exposition only
  };
}

クラステンプレートspawn-future-state-base

namespace std::execution {
  template<class Completions>
  struct spawn-future-state-base;                                 // exposition only

  template<class... Sigs>
  struct spawn-future-state-base<completion_signatures<Sigs...>>  // exposition only
    : try-cancelable {
    using variant-t = see below;                                  // exposition only
    variant-t result;                                             // exposition only
    virtual void complete() noexcept = 0;                         // exposition only
  };
}

説明用のパックSigsを、クラステンプレートspawn-future-state-baseのパラメータに指定するcompletion_signatures特殊化の引数パックと定義する。説明用のエイリアステンプレートas-tuple<Tag(Args...)>をdecayed-tuple<Tag, Args...>と定義する。

クラステンプレートspawn-future-receiver

namespace std::execution {
  template<class Completions>
  struct spawn-future-receiver {                  // exposition only
    using receiver_concept = receiver_tag;

    spawn-future-state-base<Completions>* state;  // exposition only

    template<class... T>
    void set_value(T&&... t) && noexcept {
      set-complete<set_value_t>(std::forward<T>(t)...);
    }

    template<class E>
    void set_error(E&& e) && noexcept {
      set-complete<set_error_t>(std::forward<E>(e));
    }

    void set_stopped() && noexcept {
      set-complete<set_stopped_t>();
    }

  private:
    template<class CPO, class... T>
      void set-complete(T&&... t) noexcept {      // exposition only
        constexpr bool nothrow = (is_nothrow_constructible_v<decay_t<T>, T> && ...);
        try {
          state->result.template emplace<decayed-tuple<CPO, T...>>(CPO{},
                                                                   std::forward<T>(t)...);
        }
        catch (...) {
          if constexpr (!nothrow) {
            using tuple_t = decayed-tuple<set_error_t, exception_ptr>;
            state->result.template emplace<tuple_t>(set_error_t{}, current_exception());
          }
        }
        state->complete();
      }
  };
}

型ssource-tをstoppable-sourceおよびdefault_initializableのモデルである未規定の型とし、型ssource-tのデフォルト構築オブジェクトは関連付けられた停止状態を持つ。 ssourceを型ssource-tの左辺値とする。 型stoken-tをdecltype(ssource.get_token())とする。

エイリアステンプレート future-spawned-sender

template<sender Sender, class Env>
using future-spawned-sender =  // exposition only
  decltype(write_env(stop-when(declval<Sender>(), declval<stoken-t>()), declval<Env>()));

クラステンプレートspawn-future-state

namespace std::execution {
  template<class Alloc, scope_token Token, sender Sender, class Env>
  struct spawn-future-state                                                 // exposition only
    : spawn-future-state-base<completion_signatures_of_t<future-spawned-sender<Sender, Env>>> {
    using sigs-t =                                                          // exposition only
      completion_signatures_of_t<future-spawned-sender<Sender, Env>>;
    using receiver-t =                                                      // exposition only
      spawn-future-receiver<sigs-t>;
    using op-t =                                                            // exposition only
      connect_result_t<future-spawned-sender<Sender, Env>, receiver-t>;

    spawn-future-state(Alloc alloc, Sender&& sndr, Token token, Env env)    // exposition only
      : alloc(std::move(alloc)),
        op(connect(
          write_env(stop-when(std::forward<Sender>(sndr), ssource.get_token()), std::move(env)),
          receiver-t(this))),
        assoc(token.try_associate()) {
          if (assoc)
            start(op);
          else
            set_stopped(receiver-t(this));
        }

    void complete() noexcept override;                                      // exposition only
    void consume(receiver auto& rcvr) noexcept;                             // exposition only
    void abandon() noexcept;                                                // exposition only
    void try-cancel() noexcept override {                                   // exposition only
      ssource.request_stop();
      try-set-stopped();
    }
    void try-set-stopped() noexcept;                                        // exposition only

  private:
    using assoc-t =                                                         // exposition only
      remove_cvref_t<decltype(declval<Token&>().try_associate())>;

    Alloc alloc;                                                            // exposition only
    ssource-t ssource;                                                      // exposition only
    op-t op;                                                                // exposition only
    assoc-t assoc;                                                          // exposition only

    void destroy() noexcept;                                                // exposition only
  };

  template<class Alloc, scope_token Token, sender Sender, class Env>        // exposition only
    spawn-future-state(Alloc alloc, Sender&& sndr, Token token, Env env)
      -> spawn-future-state<Alloc, Token, Sender, Env>;
}

データ競合の存在を判定する目的において、complete, consume, try-set-stopped, abandonはアトミック操作として振る舞う。 spawn-future-stateの特殊化である型の単一オブジェクトに対するこれらの操作は、単一の全順序で発生するように見える。

void complete() noexcept;

  • 効果 :
    • *thisに対するこのcompleteの呼び出しがconsume, try-set-stopped, abandonいずれかの呼び出しよりも前に発生するならば、効果を持たない。
    • そうではなく、*thisに対するconsumeの呼び出しがこのcompleteの呼び出しよりも前に発生し、かつ*thisに対するこのcompleteの呼び出しよりも前に発生するtry-set-stoppedの呼び出しが存在しなければ、consume(rcvr)によって登録解除と完了する登録されたReceiverrcvrが存在する。
    • そうではないとき、destroyが呼び出される。

void consume(receiver auto& rcvr) noexcept;


void try-set-stopped() noexcept;

  • 効果 :
    • *thisに対するconsumeの呼び出しがこのtry-set-stoppedの呼び出しよりも前に発生し、かつ*thisに対するこのtry-set-stoppedの呼び出しよりも前に発生するcompleteの呼び出しが存在しなければ、set_stopped(std::move(rcvr)), destroy()によって登録解除と完了する登録されたReceiverrcvrが存在する。
    • そうではないとき、効果を持たない。

void abandon() noexcept;

  • 効果 :

    • *thisに対するこのabandonの呼び出しがcompleteの呼び出しよりも前に発生するならば、下記と等価。

      ssource.request_stop();

    • そうでなければ、destroyが呼び出される。

void destroy() noexcept;

  • 効果 : 下記と等価。

    auto associated = std::move(this->associated);
{
  using traits = allocator_traits<Alloc>::template rebind_traits<spawn-future-state>;
  typename traits::allocator_type alloc(std::move(this->alloc));
  traits::destroy(alloc, this);
  traits::deallocate(alloc, this, 1);
}

クラステンプレートfuture-operation

namespace std::execution {
  template<class StatePtr, class Rcvr>
  struct future-operation {                              // exposition only
    struct callback {                                    // exposition only
      try-cancelable* state;                             // exposition only

      void operator()() noexcept {
        state->try-cancel();
      };
    };

    using stop-token-t =                                 // exposition only
      stop_token_of_t<env_of_t<Rcvr>>;

    using stop-callback-t =                              // exposition only
      stop_callback_for_t<stop-token-t, callback>;

    struct receiver {                                    // exposition only
      using receiver_concept = receiver_tag;
      future-operation* op;                              // exposition only

      template<class... T>
      void set_value(T&&... ts) && noexcept {
        op->set-complete<set_value_t>(std::forward<T>(ts)...);
      }

      template<class E>
      void set_error(E&& e) && noexcept {
        op->set-complete<set_error_t>(std::forward<E>(e));
      }

      void set_stopped() && noexcept {
        op->set-complete<set_stopped_t>();
      }

      env_of_t<Rcvr> get_env() const noexcept {
        return execution::get_env(op->rcvr);
      }
    };

    Rcvr rcvr;                                       // exposition only
    StatePtr state;                                  // exposition only
    receiver inner;                                  // exposition only
    optional<stop-callback-t> stopCallback;          // exposition only


    future-operation(StatePtr state, Rcvr rcvr) noexcept // exposition only
      : rcvr(std::move(rcvr)), state(std::move(state)), inner(this)
    {}

    future-operation(future-operation&&) = delete;

    void run() & noexcept {                              // exposition only
      constexpr bool nothrow =
        is_nothrow_constructible_v<stop-callback-t, stop-token-t, callback>;
      try {
          stopCallback.emplace(get_stop_token(rcvr), callback(state.get()));
      }
      catch (...) {
        if constexpr (!nothrow) {
          set_error(std::move(rcvr), current_exception());
          return;
        }
      }

      state.release()->consume(inner);
    }

    template<class CPO, class... T>
    void set-complete(T&&... ts) noexcept {              // exposition only
      stopCallback.reset();
      CPO{}(std::move(rcvr), std::forward<T>(ts)...);
    }
  };
}

例

#include <print>
#include <execution>
namespace ex = std::execution;

int main()
{
  // スレッドプール上の実行タスクを定義
  ex::scheduler auto sch = ex::get_parallel_scheduler();
  ex::sender auto snd0 =
    ex::schedule(sch)
    | ex::then([]{
        // ワーカースレッド処理
        std::println("hello async");
        return 42;
      });

  // 非同期スコープを定義
  ex::counting_scope scope;

  // タスクを早期開始させる
  std::println("spawn");
  ex::sender auto snd1 = ex::spawn_future(std::move(snd0), scope.get_token());

  // (メインスレッド処理)

  // 非同期スコープを合流
  auto result = std::this_thread::sync_wait(ex::when_all(std::move(snd1), scope.join()));
  auto [value] = *result;
  std::println("value={}", value);
}

出力

spawn
hello async
value=42

バージョン

言語

  • C++26

処理系

関連項目

参照