pw_async: Create HeapDispatcher, FunctionDispatcher

Create abstract class FunctionDispatcher that extends Dispatcher with
Post*() methods that take a TaskFunction instead of a task.

Create concrete class HeapDispatcher that implements FunctionDispatcher
using new/delete to manage Task memory.

Update Dispatcher::Cancel() docs to explain that tasks can be
canceled from within a TaskFunction.

Change-Id: I6cabe9614cb0b26c4e21f31b4d06713b4e936e42
Reviewed-on: https://pigweed-review.googlesource.com/c/pigweed/pigweed/+/135192
Reviewed-by: Taylor Cramer <cramertj@google.com>
Commit-Queue: Ben Lawson <benlawson@google.com>

3 files changed, 88 insertions, 0 deletions

diff --git a/pw_async_basic/BUILD.bazel b/pw_async_basic/BUILD.bazel
index 8a1fb3aca..8a689c3b8 100644
--- a/pw_async_basic/BUILD.bazel
+++ b/pw_async_basic/BUILD.bazel
@@ -19,6 +19,7 @@ filegroup(
         "dispatcher_test.cc",
         "fake_dispatcher.cc",
         "fake_dispatcher_fixture_test.cc",
+        "heap_dispatcher_test.cc",
         "public/pw_async_basic/dispatcher.h",
         "public/pw_async_basic/fake_dispatcher.h",
         "public/pw_async_basic/task.h",
diff --git a/pw_async_basic/BUILD.gn b/pw_async_basic/BUILD.gn
index 9ccce5e85..f3d753cce 100644
--- a/pw_async_basic/BUILD.gn
+++ b/pw_async_basic/BUILD.gn
@@ -17,6 +17,7 @@ import("//build_overrides/pigweed.gni")
 import("$dir_pw_async/async.gni")
 import("$dir_pw_async/fake_dispatcher_fixture.gni")
 import("$dir_pw_async/fake_dispatcher_test.gni")
+import("$dir_pw_async/heap_dispatcher.gni")
 import("$dir_pw_bloat/bloat.gni")
 import("$dir_pw_build/target_types.gni")
 import("$dir_pw_chrono/backend.gni")
@@ -122,11 +123,27 @@ pw_test("dispatcher_test") {
   sources = [ "dispatcher_test.cc" ]
 }
 
+pw_async_heap_dispatcher_source_set("heap_dispatcher") {
+  task_backend = ":task"
+  visibility = [ ":*" ] + pw_async_EXPERIMENTAL_MODULE_VISIBILITY
+}
+
+pw_test("heap_dispatcher_test") {
+  enable_if = pw_chrono_SYSTEM_CLOCK_BACKEND != "" &&
+              pw_thread_THREAD_BACKEND == "$dir_pw_thread_stl:thread"
+  sources = [ "heap_dispatcher_test.cc" ]
+  deps = [
+    ":fake_dispatcher_fixture",
+    ":heap_dispatcher",
+  ]
+}
+
 pw_test_group("tests") {
   tests = [
     ":dispatcher_test",
     ":fake_dispatcher_test",
     ":fake_dispatcher_fixture_test",
+    ":heap_dispatcher_test",
   ]
 }
 
diff --git a/pw_async_basic/heap_dispatcher_test.cc b/pw_async_basic/heap_dispatcher_test.cc
new file mode 100644
index 000000000..3c744971d
--- /dev/null
+++ b/pw_async_basic/heap_dispatcher_test.cc
@@ -0,0 +1,70 @@
+// Copyright 2023 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 "pw_async/heap_dispatcher.h"
+
+#include "gtest/gtest.h"
+#include "pw_async/fake_dispatcher_fixture.h"
+
+using namespace std::chrono_literals;
+
+namespace pw::async {
+namespace {
+
+using HeapDispatcherTest = test::FakeDispatcherFixture;
+
+class DestructionChecker {
+ public:
+  DestructionChecker(bool* flag) : flag_(flag) {}
+  DestructionChecker(DestructionChecker&& other) {
+    flag_ = other.flag_;
+    other.flag_ = nullptr;
+  }
+  ~DestructionChecker() {
+    if (flag_) {
+      *flag_ = true;
+    }
+  }
+
+ private:
+  bool* flag_;
+};
+
+TEST_F(HeapDispatcherTest, RunUntilIdleRunsPostedTask) {
+  HeapDispatcher heap_dispatcher(dispatcher());
+
+  int count = 0;
+  Status status = heap_dispatcher.Post(
+      [&count](Context& /*ctx*/, Status /*status*/) { ++count; });
+  EXPECT_TRUE(status.ok());
+  ASSERT_EQ(count, 0);
+  RunUntilIdle();
+  ASSERT_EQ(count, 1);
+}
+
+TEST_F(HeapDispatcherTest, TaskFunctionIsDestroyedAfterBeingCalled) {
+  HeapDispatcher heap_dispatcher(dispatcher());
+
+  // Test that the lambda is destroyed after being called.
+  bool flag = false;
+  Status status =
+      heap_dispatcher.Post([checker = DestructionChecker(&flag)](
+                               Context& /*ctx*/, Status /*status*/) {});
+  EXPECT_TRUE(status.ok());
+  EXPECT_FALSE(flag);
+  RunUntilIdle();
+  EXPECT_TRUE(flag);
+}
+
+}  // namespace
+}  // namespace pw::async