pw_async2/value_future_thread_test.cc

// Copyright 2025 The Pigweed Authors
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not
// use this file except in compliance with the License. You may obtain a copy of
// the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations under
// the License.

#include <atomic>

#include "pw_async2/callback_task.h"
#include "pw_async2/dispatcher_for_test.h"
#include "pw_async2/value_future.h"
#include "pw_thread/sleep.h"
#include "pw_thread/test_thread_context.h"
#include "pw_thread/thread.h"
#include "pw_thread/yield.h"
#include "pw_unit_test/framework.h"

namespace {

using pw::async2::BroadcastValueProvider;
using pw::async2::CallbackTask;
using pw::async2::DispatcherForTest;
using pw::async2::OptionalBroadcastValueProvider;
using pw::async2::OptionalValueProvider;
using pw::async2::ValueProvider;

using namespace std::chrono_literals;

constexpr int kIterations = 5000;

TEST(ValueFuture, ResolveFromOtherThread) {
  DispatcherForTest dispatcher;
  ValueProvider<int> provider;

  std::optional<int> result;
  CallbackTask task([&](int value) { result = value; }, provider.Get());
  dispatcher.Post(task);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&provider]() {
    pw::this_thread::sleep_for(1ms);
    provider.Resolve(42);
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  ASSERT_TRUE(result.has_value());
  EXPECT_EQ(*result, 42);
}

TEST(ValueFuture, Broadcast_ResolveFromOtherThread) {
  DispatcherForTest dispatcher;
  BroadcastValueProvider<int> provider;

  std::optional<int> result1;
  CallbackTask task1([&](int value) { result1 = value; }, provider.Get());
  std::optional<int> result2;
  CallbackTask task2([&](int value) { result2 = value; }, provider.Get());

  dispatcher.Post(task1);
  dispatcher.Post(task2);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&provider]() {
    pw::this_thread::sleep_for(1ms);
    provider.Resolve(123);
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  ASSERT_TRUE(result1.has_value());
  EXPECT_EQ(*result1, 123);
  ASSERT_TRUE(result2.has_value());
  EXPECT_EQ(*result2, 123);
}

TEST(ValueFuture, Void_ResolveFromOtherThread) {
  DispatcherForTest dispatcher;
  ValueProvider<void> provider;

  bool completed = false;
  CallbackTask task([&]() { completed = true; }, provider.Get());
  dispatcher.Post(task);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&provider]() {
    pw::this_thread::sleep_for(1ms);
    provider.Resolve();
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  EXPECT_TRUE(completed);
}

TEST(OptionalValueProvider, ResolveFromOtherThread) {
  DispatcherForTest dispatcher;
  OptionalValueProvider<int> provider;

  std::optional<int> result;
  CallbackTask task([&](std::optional<int> value) { result = value; },
                    provider.Get());
  dispatcher.Post(task);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&provider]() {
    pw::this_thread::sleep_for(1ms);
    provider.Resolve(42);
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  ASSERT_TRUE(result.has_value());
  EXPECT_EQ(*result, 42);
}

TEST(OptionalValueProvider, CancelFromOtherThread) {
  DispatcherForTest dispatcher;
  OptionalValueProvider<int> provider;

  std::optional<int> result = 5;
  CallbackTask task([&](std::optional<int> value) { result = value; },
                    provider.Get());
  dispatcher.Post(task);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&provider]() {
    pw::this_thread::sleep_for(1ms);
    provider.Cancel();
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  EXPECT_FALSE(result.has_value());
}

TEST(ValueProvider, MoveToThread) {
  DispatcherForTest dispatcher;
  ValueProvider<int> provider;

  std::optional<int> result;
  CallbackTask task([&](int value) { result = value; }, provider.Get());
  dispatcher.Post(task);

  // Move the provider into the thread lambda.
  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(),
                             [moved_provider = std::move(provider)]() mutable {
                               pw::this_thread::sleep_for(1ms);
                               moved_provider.Resolve(42);
                             });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  ASSERT_TRUE(result.has_value());
  EXPECT_EQ(*result, 42);
}

TEST(OptionalBroadcastValueProvider, ResolveFromOtherThread) {
  DispatcherForTest dispatcher;
  OptionalBroadcastValueProvider<int> provider;

  std::optional<int> result1;
  CallbackTask task1([&](std::optional<int> value) { result1 = value; },
                     provider.Get());

  std::optional<int> result2;
  CallbackTask task2([&](std::optional<int> value) { result2 = value; },
                     provider.Get());

  dispatcher.Post(task1);
  dispatcher.Post(task2);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&provider]() {
    pw::this_thread::sleep_for(1ms);
    provider.Resolve(42);
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  ASSERT_TRUE(result1.has_value());
  EXPECT_EQ(*result1, 42);

  ASSERT_TRUE(result2.has_value());
  EXPECT_EQ(*result2, 42);
}

TEST(OptionalBroadcastValueProvider, CancelFromOtherThread) {
  DispatcherForTest dispatcher;
  OptionalBroadcastValueProvider<int> provider;

  std::optional<int> result1 = 5;
  CallbackTask task1([&](std::optional<int> value) { result1 = value; },
                     provider.Get());

  std::optional<int> result2 = 5;
  CallbackTask task2([&](std::optional<int> value) { result2 = value; },
                     provider.Get());

  dispatcher.Post(task1);
  dispatcher.Post(task2);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&provider]() {
    pw::this_thread::sleep_for(1ms);
    provider.Cancel();
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  EXPECT_FALSE(result1.has_value());

  EXPECT_FALSE(result2.has_value());
}

TEST(OptionalValueProvider, DestructFromOtherThread) {
  DispatcherForTest dispatcher;
  std::optional<OptionalValueProvider<int>> provider;
  provider.emplace();

  std::optional<int> result = 5;  // nullopt after provider is destroyed
  CallbackTask task([&](std::optional<int> value) { result = value; },
                    provider->Get());
  dispatcher.Post(task);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&provider]() {
    pw::this_thread::sleep_for(1ms);
    provider.reset();
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  EXPECT_FALSE(result.has_value());
}

TEST(ValueFuture, Void_MoveRace_LoopingTask) {
  DispatcherForTest dispatcher;

  struct {
    ValueProvider<void> provider;
    std::atomic<bool> running{true};
  } test_context;

  // Continuously vend, move, and discard futures while the provider thread
  // resolves them.
  class GetterTask : public pw::async2::Task {
   public:
    GetterTask(ValueProvider<void>& provider, std::atomic<bool>& running)
        : provider_(&provider), running_(&running) {}

   private:
    pw::async2::Poll<> DoPend(pw::async2::Context&) override {
      while (running_->load(std::memory_order_relaxed)) {
        auto f = provider_->Get();
        auto f2 = std::move(f);
      }
      return pw::async2::Ready();
    }

    ValueProvider<void>* provider_;
    std::atomic<bool>* running_;
  };

  GetterTask task(test_context.provider, test_context.running);
  dispatcher.Post(task);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&test_context]() {
    for (int i = 0; i < kIterations; ++i) {
      test_context.provider.Resolve();
      pw::this_thread::yield();
    }
    test_context.running.store(false, std::memory_order_relaxed);
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();
}

template <typename T = int>
class DerivedTestFutureForThreadTest : public pw::async2::ValueFuture<T> {
 public:
  constexpr DerivedTestFutureForThreadTest() = default;
  DerivedTestFutureForThreadTest(pw::async2::ValueFuture<T>&& base, int value)
      : pw::async2::ValueFuture<T>(std::move(base)), value_(value) {}

  int value() const { return value_; }

 private:
  int value_ = 0;
};

TEST(DerivedValueProvider, ResolveIfFromOtherThread) {
  DispatcherForTest dispatcher;
  pw::async2::DerivedValueProvider<DerivedTestFutureForThreadTest<int>>
      provider;

  std::optional<int> result;
  CallbackTask task([&](int value) { result = value; }, provider.Get(42));
  dispatcher.Post(task);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&provider]() {
    pw::this_thread::sleep_for(1ms);
    bool resolved = provider.ResolveIf(
        [](DerivedTestFutureForThreadTest<int>& f) -> std::optional<int> {
          if (f.value() == 42) {
            return f.value() * 2;
          }
          return std::nullopt;
        });
    EXPECT_TRUE(resolved);
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  ASSERT_TRUE(result.has_value());
  EXPECT_EQ(*result, 84);
}

TEST(DerivedValueProvider, TryGetFromOtherThread) {
  DispatcherForTest dispatcher;
  struct {
    pw::async2::DerivedValueProvider<DerivedTestFutureForThreadTest<int>>
        provider;
    std::atomic<bool> try_get_checked{false};
  } test_context;

  std::optional<int> result;
  CallbackTask task([&](int value) { result = value; },
                    test_context.provider.Get(10));
  dispatcher.Post(task);

  pw::thread::test::TestThreadContext context;
  pw::Thread thread(context.options(), [&test_context]() {
    EXPECT_FALSE(test_context.provider.TryGet(20).has_value());
    test_context.try_get_checked.store(true, std::memory_order_release);
  });

  dispatcher.AllowBlocking();
  while (!test_context.try_get_checked.load(std::memory_order_acquire)) {
    pw::this_thread::yield();
  }
  test_context.provider.Resolve(100);
  dispatcher.RunToCompletion();
  thread.join();

  std::optional<DerivedTestFutureForThreadTest<int>> future2 =
      test_context.provider.TryGet(20);
  ASSERT_TRUE(future2.has_value());
  EXPECT_EQ(future2->value(), 20);
  test_context.provider.Resolve(200);
}

TEST(ValueListProvider, ResolveFirstFromOtherThread) {
  DispatcherForTest dispatcher;
  pw::async2::ValueListProvider<int> provider;

  std::optional<int> result1;
  CallbackTask task1([&](int value) { result1 = value; }, provider.Get());
  std::optional<int> result2;
  CallbackTask task2([&](int value) { result2 = value; }, provider.Get());

  dispatcher.Post(task1);
  dispatcher.Post(task2);

  pw::thread::test::TestThreadContext context;
  pw::Thread resolver_thread(context.options(), [&provider]() {
    pw::this_thread::sleep_for(1ms);
    provider.ResolveFirst(111);
    provider.ResolveFirst(222);
  });

  dispatcher.AllowBlocking();
  dispatcher.RunToCompletion();
  resolver_thread.join();

  ASSERT_TRUE(result1.has_value());
  EXPECT_EQ(*result1, 111);
  ASSERT_TRUE(result2.has_value());
  EXPECT_EQ(*result2, 222);
}

TEST(ValueListProvider, ConcurrentGetAndResolveFirstMatching) {
  struct {
    pw::async2::DerivedValueListProvider<
        DerivedTestFutureForThreadTest<pw::Status>>
        provider;
    std::atomic<bool> running{true};
  } test_context;

  pw::thread::test::TestThreadContext context1;
  pw::thread::test::TestThreadContext context2;

  pw::Thread getter_thread(context1.options(), [&test_context]() {
    while (test_context.running.load(std::memory_order_relaxed)) {
      auto f = test_context.provider.Get(42);
      pw::this_thread::yield();
    }
  });

  pw::Thread resolver_thread(context2.options(), [&test_context]() {
    for (int i = 0; i < kIterations; ++i) {
      test_context.provider.ResolveFirstMatching(
          [](DerivedTestFutureForThreadTest<pw::Status>& f)
              -> std::optional<pw::Status> {
            if (f.value() == 42) {
              return pw::OkStatus();
            }
            return std::nullopt;
          });
      pw::this_thread::yield();
    }
    test_context.running.store(false, std::memory_order_relaxed);
  });

  getter_thread.join();
  resolver_thread.join();

  test_context.provider.ResolveAll(
      [](DerivedTestFutureForThreadTest<pw::Status>&) {
        return pw::OkStatus();
      });
}

}  // namespace