timers: refactor setImmediate error handling

If an error is encountered during the processing of Immediates, schedule
the remaining queue to finish after all error handling code runs (if the
process is still alive to do so). The new changes make the Immediates
error handling behaviour entirely deterministic and predictable, as the
full queue will be flushed on each Immediates cycle, regardless of
whether an error is encountered or not.

Currently this processing is scheduled for nextTick which can yield
unpredictable results as the nextTick might happen as early as close
callbacks phase or as late as after the next event loop turns Immediates
all fully processed. The latter can result in two full cycles of
Immediates processing during one even loop turn.

The current implementation also doesn't differentiate between Immediates
scheduled for the current queue run or the next one, so Immediates that
were scheduled for the next turn of the event loop, will process
alongside the ones that were scheduled for the current turn.

PR-URL: https://github.com/nodejs/node/pull/17879
Reviewed-By: Anna Henningsen <anna@addaleax.net>
Reviewed-By: James M Snell <jasnell@gmail.com>

4 files changed, 71 insertions, 22 deletions

diff --git a/src/env-inl.h b/src/env-inl.h
index c25e126549..78a02909c4 100644
--- a/src/env-inl.h
+++ b/src/env-inl.h
@@ -217,6 +217,30 @@ inline bool Environment::AsyncCallbackScope::in_makecallback() const {
   return env_->makecallback_cntr_ > 1;
 }
 
+inline Environment::ImmediateInfo::ImmediateInfo(v8::Isolate* isolate)
+    : fields_(isolate, kFieldsCount) {}
+
+inline AliasedBuffer<uint32_t, v8::Uint32Array>&
+    Environment::ImmediateInfo::fields() {
+  return fields_;
+}
+
+inline uint32_t Environment::ImmediateInfo::count() const {
+  return fields_[kCount];
+}
+
+inline bool Environment::ImmediateInfo::has_outstanding() const {
+  return fields_[kHasOutstanding] == 1;
+}
+
+inline void Environment::ImmediateInfo::count_inc(uint32_t increment) {
+  fields_[kCount] = fields_[kCount] + increment;
+}
+
+inline void Environment::ImmediateInfo::count_dec(uint32_t decrement) {
+  fields_[kCount] = fields_[kCount] - decrement;
+}
+
 inline Environment::TickInfo::TickInfo(v8::Isolate* isolate)
     : fields_(isolate, kFieldsCount) {}
 
@@ -263,6 +287,7 @@ inline Environment::Environment(IsolateData* isolate_data,
                                 v8::Local<v8::Context> context)
     : isolate_(context->GetIsolate()),
       isolate_data_(isolate_data),
+      immediate_info_(context->GetIsolate()),
       tick_info_(context->GetIsolate()),
       timer_base_(uv_now(isolate_data->event_loop())),
       using_domains_(false),
@@ -271,7 +296,6 @@ inline Environment::Environment(IsolateData* isolate_data,
       abort_on_uncaught_exception_(false),
       emit_napi_warning_(true),
       makecallback_cntr_(0),
-      scheduled_immediate_count_(isolate_, 1),
       should_abort_on_uncaught_toggle_(isolate_, 1),
 #if HAVE_INSPECTOR
       inspector_agent_(new inspector::Agent(this)),
@@ -371,6 +395,10 @@ inline Environment::AsyncHooks* Environment::async_hooks() {
   return &async_hooks_;
 }
 
+inline Environment::ImmediateInfo* Environment::immediate_info() {
+  return &immediate_info_;
+}
+
 inline Environment::TickInfo* Environment::tick_info() {
   return &tick_info_;
 }
@@ -508,11 +536,6 @@ inline void Environment::set_fs_stats_field_array(double* fields) {
   fs_stats_field_array_ = fields;
 }
 
-inline AliasedBuffer<uint32_t, v8::Uint32Array>&
-Environment::scheduled_immediate_count() {
-  return scheduled_immediate_count_;
-}
-
 void Environment::SetImmediate(native_immediate_callback cb,
                                void* data,
                                v8::Local<v8::Object> obj) {
@@ -522,9 +545,9 @@ void Environment::SetImmediate(native_immediate_callback cb,
     std::unique_ptr<v8::Persistent<v8::Object>>(
         obj.IsEmpty() ? nullptr : new v8::Persistent<v8::Object>(isolate_, obj))
   });
-  if (scheduled_immediate_count_[0] == 0)
+  if (immediate_info()->count() == 0)
     ActivateImmediateCheck();
-  scheduled_immediate_count_[0] = scheduled_immediate_count_[0] + 1;
+  immediate_info()->count_inc(1);
 }
 
 inline performance::performance_state* Environment::performance_state() {
diff --git a/src/env.cc b/src/env.cc
index eb8cad15ed..fb23a8c1c9 100644
--- a/src/env.cc
+++ b/src/env.cc
@@ -283,14 +283,14 @@ void Environment::RunAndClearNativeImmediates() {
     }
 
 #ifdef DEBUG
-    CHECK_GE(scheduled_immediate_count_[0], count);
+    CHECK_GE(immediate_info()->count(), count);
 #endif
-    scheduled_immediate_count_[0] = scheduled_immediate_count_[0] - count;
+    immediate_info()->count_dec(count);
   }
 }
 
 static bool MaybeStopImmediate(Environment* env) {
-  if (env->scheduled_immediate_count()[0] == 0) {
+  if (env->immediate_info()->count() == 0) {
     uv_check_stop(env->immediate_check_handle());
     uv_idle_stop(env->immediate_idle_handle());
     return true;
@@ -309,12 +309,14 @@ void Environment::CheckImmediate(uv_check_t* handle) {
 
   env->RunAndClearNativeImmediates();
 
-  MakeCallback(env->isolate(),
-               env->process_object(),
-               env->immediate_callback_function(),
-               0,
-               nullptr,
-               {0, 0}).ToLocalChecked();
+  do {
+    MakeCallback(env->isolate(),
+                 env->process_object(),
+                 env->immediate_callback_function(),
+                 0,
+                 nullptr,
+                 {0, 0}).ToLocalChecked();
+  } while (env->immediate_info()->has_outstanding());
 
   MaybeStopImmediate(env);
 }
diff --git a/src/env.h b/src/env.h
index 268abd441d..abd089dddd 100644
--- a/src/env.h
+++ b/src/env.h
@@ -453,6 +453,30 @@ class Environment {
     DISALLOW_COPY_AND_ASSIGN(AsyncCallbackScope);
   };
 
+  class ImmediateInfo {
+   public:
+    inline AliasedBuffer<uint32_t, v8::Uint32Array>& fields();
+    inline uint32_t count() const;
+    inline bool has_outstanding() const;
+
+    inline void count_inc(uint32_t increment);
+    inline void count_dec(uint32_t decrement);
+
+   private:
+    friend class Environment;  // So we can call the constructor.
+    inline explicit ImmediateInfo(v8::Isolate* isolate);
+
+    enum Fields {
+      kCount,
+      kHasOutstanding,
+      kFieldsCount
+    };
+
+    AliasedBuffer<uint32_t, v8::Uint32Array> fields_;
+
+    DISALLOW_COPY_AND_ASSIGN(ImmediateInfo);
+  };
+
   class TickInfo {
    public:
     inline AliasedBuffer<uint8_t, v8::Uint8Array>& fields();
@@ -532,6 +556,7 @@ class Environment {
   inline void FinishHandleCleanup(uv_handle_t* handle);
 
   inline AsyncHooks* async_hooks();
+  inline ImmediateInfo* immediate_info();
   inline TickInfo* tick_info();
   inline uint64_t timer_base() const;
 
@@ -582,8 +607,6 @@ class Environment {
   inline double* fs_stats_field_array() const;
   inline void set_fs_stats_field_array(double* fields);
 
-  inline AliasedBuffer<uint32_t, v8::Uint32Array>& scheduled_immediate_count();
-
   inline performance::performance_state* performance_state();
   inline std::map<std::string, uint64_t>* performance_marks();
 
@@ -704,6 +727,7 @@ class Environment {
   uv_check_t idle_check_handle_;
 
   AsyncHooks async_hooks_;
+  ImmediateInfo immediate_info_;
   TickInfo tick_info_;
   const uint64_t timer_base_;
   bool using_domains_;
@@ -714,7 +738,6 @@ class Environment {
   size_t makecallback_cntr_;
   std::vector<double> destroy_async_id_list_;
 
-  AliasedBuffer<uint32_t, v8::Uint32Array> scheduled_immediate_count_;
   AliasedBuffer<uint32_t, v8::Uint32Array> should_abort_on_uncaught_toggle_;
 
   int should_not_abort_scope_counter_ = 0;
diff --git a/src/timer_wrap.cc b/src/timer_wrap.cc
index 5c3f499d16..ab450fcb3e 100644
--- a/src/timer_wrap.cc
+++ b/src/timer_wrap.cc
@@ -90,8 +90,9 @@ class TimerWrap : public HandleWrap {
         env->NewFunctionTemplate(activate_cb)->GetFunction(env->context())
         .ToLocalChecked();
     auto result = Array::New(env->isolate(), 2);
-    result->Set(0, activate_function);
-    result->Set(1, env->scheduled_immediate_count().GetJSArray());
+    result->Set(env->context(), 0, activate_function).FromJust();
+    result->Set(env->context(), 1,
+                env->immediate_info()->fields().GetJSArray()).FromJust();
     args.GetReturnValue().Set(result);
   }