Wakeup primitive between actors in the same activity (#27081)

* promise sequences

* Activities for promises

* Wakeups between actors in the same activity

* Automated change: Fix sanity tests

* Automated change: Fix sanity tests

Co-authored-by: ctiller <ctiller@users.noreply.github.com>

BUILD

@@ -1116,6 +1116,18 @@ grpc_cc_library(
     ],
 )
 
+grpc_cc_library(
+    name = "intra_activity_waiter",
+    language = "c++",
+    public_hdrs = [
+        "src/core/lib/promise/intra_activity_waiter.h",
+    ],
+    deps = [
+        "activity",
+        "gpr_platform",
+    ],
+)
+
 grpc_cc_library(
     name = "ref_counted",
     language = "c++",

build_autogenerated.yaml

@@ -4224,6 +4224,7 @@ targets:
   - src/core/ext/upb-generated/google/rpc/status.upb.h
   - src/core/lib/gpr/alloc.h
   - src/core/lib/gpr/env.h
+  - src/core/lib/gpr/log_internal.h
   - src/core/lib/gpr/murmur_hash.h
   - src/core/lib/gpr/spinlock.h
   - src/core/lib/gpr/string.h

src/core/lib/promise/intra_activity_waiter.h

@@ -0,0 +1,48 @@
+// Copyright 2021 gRPC 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
+//
+//     http://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.
+
+#ifndef GRPC_CORE_LIB_PROMISE_INTRA_ACTIVITY_WAITER_H
+#define GRPC_CORE_LIB_PROMISE_INTRA_ACTIVITY_WAITER_H
+
+#include <grpc/impl/codegen/port_platform.h>
+
+#include "src/core/lib/promise/activity.h"
+
+namespace grpc_core {
+
+// Helper type to track wakeups between objects in the same activity.
+// Can be fairly fast as no ref counting or locking needs to occur.
+class IntraActivityWaiter {
+ public:
+  // Register for wakeup, return Pending(). If state is not ready to proceed,
+  // Promises should bottom out here.
+  Pending pending() {
+    waiting_ = true;
+    return Pending();
+  }
+  // Wake the activity
+  void Wake() {
+    if (waiting_) {
+      waiting_ = false;
+      Activity::WakeupCurrent();
+    }
+  }
+
+ private:
+  bool waiting_ = false;
+};
+
+}  // namespace grpc_core
+
+#endif  // GRPC_CORE_LIB_PROMISE_INTRA_ACTIVITY_WAITER_H