Only execute expired tasks once per batched event group (#39013)

Summary:
Pull Request resolved: https://github.com/facebook/react-native/pull/39013

Changelog: [internal]

Reviewed By: sammy-SC

Differential Revision: D48210131

fbshipit-source-id: 1410aab2f70b75e71fd8a6ace8f4653682daaf90

packages/react-native/ReactCommon/react/renderer/core/EventPipe.h

@@ -25,4 +25,6 @@ using EventPipe = std::function<void(
     ReactEventPriority priority,
     const EventPayload &payload)>;
 
+using EventPipeConclusion = std::function<void(jsi::Runtime &runtime)>;
+
 } // namespace facebook::react

packages/react-native/ReactCommon/react/renderer/core/EventQueueProcessor.cpp

@@ -7,6 +7,7 @@
 
 #include <cxxreact/JSExecutor.h>
 #include <logger/react_native_log.h>
+#include <react/utils/CoreFeatures.h>
 #include "EventEmitter.h"
 #include "EventLogger.h"
 #include "EventQueue.h"
@@ -16,8 +17,11 @@ namespace facebook::react {
 
 EventQueueProcessor::EventQueueProcessor(
     EventPipe eventPipe,
+    EventPipeConclusion eventPipeConclusion,
     StatePipe statePipe)
-    : eventPipe_(std::move(eventPipe)), statePipe_(std::move(statePipe)) {}
+    : eventPipe_(std::move(eventPipe)),
+      eventPipeConclusion_(std::move(eventPipeConclusion)),
+      statePipe_(std::move(statePipe)) {}
 
 void EventQueueProcessor::flushEvents(
     jsi::Runtime &runtime,
@@ -67,6 +71,11 @@ void EventQueueProcessor::flushEvents(
         reactPriority,
         *event.eventPayload);
 
+    // We run the "Conclusion" per-event when unbatched
+    if (!CoreFeatures::enableDefaultAsyncBatchedPriority) {
+      eventPipeConclusion_(runtime);
+    }
+
     if (eventLogger != nullptr) {
       eventLogger->onEventEnd(event.loggingTag);
     }
@@ -75,6 +84,10 @@ void EventQueueProcessor::flushEvents(
       hasContinuousEventStarted_ = true;
     }
   }
+  // We only run the "Conclusion" once per event group when batched.
+  if (CoreFeatures::enableDefaultAsyncBatchedPriority) {
+    eventPipeConclusion_(runtime);
+  }
 
   // No need to lock `EventEmitter::DispatchMutex()` here.
   // The mutex protects from a situation when the `instanceHandle` can be

packages/react-native/ReactCommon/react/renderer/core/EventQueueProcessor.h

@@ -19,13 +19,17 @@ namespace facebook::react {
 
 class EventQueueProcessor {
  public:
-  EventQueueProcessor(EventPipe eventPipe, StatePipe statePipe);
+  EventQueueProcessor(
+      EventPipe eventPipe,
+      EventPipeConclusion eventPipeConclusion,
+      StatePipe statePipe);
 
   void flushEvents(jsi::Runtime &runtime, std::vector<RawEvent> &&events) const;
   void flushStateUpdates(std::vector<StateUpdate> &&states) const;
 
  private:
   EventPipe const eventPipe_;
+  EventPipeConclusion const eventPipeConclusion_;
   StatePipe const statePipe_;
 
   mutable bool hasContinuousEventStarted_{false};

packages/react-native/ReactCommon/react/renderer/core/tests/EventQueueProcessorTest.cpp

@@ -32,10 +32,11 @@ class EventQueueProcessorTest : public testing::Test {
       eventPriorities_.push_back(priority);
     };
 
+    auto dummyEventPipeConclusion = [](jsi::Runtime &runtime) {};
     auto dummyStatePipe = [](StateUpdate const &stateUpdate) {};
 
-    eventProcessor_ =
-        std::make_unique<EventQueueProcessor>(eventPipe, dummyStatePipe);
+    eventProcessor_ = std::make_unique<EventQueueProcessor>(
+        eventPipe, dummyEventPipeConclusion, dummyStatePipe);
   }
 
   std::unique_ptr<facebook::hermes::HermesRuntime> runtime_;

packages/react-native/ReactCommon/react/renderer/scheduler/Scheduler.cpp

@@ -67,11 +67,23 @@ Scheduler::Scheduler(
               runtime, eventTarget, type, priority, payload);
         },
         runtime);
-    if (runtimeScheduler != nullptr) {
-      runtimeScheduler->callExpiredTasks(runtime);
+
+    // We only want to run this per-event if we are not batching by default
+    if (!CoreFeatures::enableDefaultAsyncBatchedPriority) {
+      if (runtimeScheduler != nullptr) {
+        runtimeScheduler->callExpiredTasks(runtime);
+      }
     }
   };
 
+  auto eventPipeConclusion =
+      [runtimeScheduler = runtimeScheduler.get()](jsi::Runtime &runtime) {
+        if (CoreFeatures::enableDefaultAsyncBatchedPriority &&
+            runtimeScheduler != nullptr) {
+          runtimeScheduler->callExpiredTasks(runtime);
+        }
+      };
+
   auto statePipe = [uiManager](StateUpdate const &stateUpdate) {
     uiManager->updateState(stateUpdate);
   };
@@ -79,7 +91,7 @@ Scheduler::Scheduler(
   // Creating an `EventDispatcher` instance inside the already allocated
   // container (inside the optional).
   eventDispatcher_->emplace(
-      EventQueueProcessor(eventPipe, statePipe),
+      EventQueueProcessor(eventPipe, eventPipeConclusion, statePipe),
       schedulerToolbox.synchronousEventBeatFactory,
       schedulerToolbox.asynchronousEventBeatFactory,
       eventOwnerBox);