node: AsyncListener use separate storage mechanism

Before when an AsyncListener object was created and the "create" callback returned a value, it was necessary to construct a new Object with the same callbacks but add a place for the new storage value. Now, instead, a separate storage array is kept on the context which is used for any return value of the "create" callback. This significantly reduces the number of Objects that need to be created. Also added a flags property to the context to quickly check if a specific callback was available either on the context or on the AsyncListener instance itself. Few other minor changes for readability that were difficult to separate into their own commit. This has not been optimized yet.
nodejs · Jan 8, 2014 · 66e353d · 66e353d
1 parent 3905986
commit 66e353d
Show file tree

Hide file tree

Showing 6 changed files with 324 additions and 159 deletions.
diff --git a/lib/timers.js b/lib/timers.js
@@ -35,6 +35,9 @@ var runAsyncQueue = process._runAsyncQueue;
 var loadAsyncQueue = process._loadAsyncQueue;
 var unloadAsyncQueue = process._unloadAsyncQueue;
 
+// Same as in AsyncListener in env.h
+var kHasListener = 0;
+
 // Do a little housekeeping.
 delete process._asyncFlags;
 delete process._runAsyncQueue;
@@ -57,6 +60,8 @@ var lists = {};
 
 // Make Timer as monomorphic as possible.
 Timer.prototype._asyncQueue = undefined;
+Timer.prototype._asyncData = undefined;
+Timer.prototype._asyncFlags = 0;
 
 // the main function - creates lists on demand and the watchers associated
 // with them.
@@ -190,35 +195,59 @@ exports.active = function(item) {
   // Whether or not a new TimerWrap needed to be created, this should run
   // for each item. This way each "item" (i.e. timer) can properly have
   // their own domain assigned.
-  if (asyncFlags[0] > 0)
+  if (asyncFlags[kHasListener] > 0)
     runAsyncQueue(item);
 };
 
 
-function timerAddAsyncListener(obj) {
+// new Array() is used here because it is more efficient for sparse
+// arrays. Please *do not* change these to simple bracket notation.
+function timerAddAsyncListener(obj, data) {
+  obj = process.createAsyncListener(obj, data);
+
   if (!this._asyncQueue)
-    this._asyncQueue = [];
+    this._asyncQueue = new Array();
+  if (!this._asyncData)
+    this._asyncData = new Array();
+
+  var inQueue = false;
   var queue = this._asyncQueue;
   // This queue will be small. Probably always <= 3 items.
   for (var i = 0; i < queue.length; i++) {
-    if (queue[i].uid === obj.uid)
-      return;
+    if (queue[i].uid === obj.uid) {
+      inQueue = true;
+      break;
+    }
   }
-  this._asyncQueue.push(obj);
+
+  if (!inQueue) {
+    queue.push(obj);
+    this._asyncData[obj.uid] = obj.data;
+    this._asyncFlags |= obj.flags;
+  }
+
+  return obj;
 }
 
 
 function timerRemoveAsyncListener(obj) {
   if (!this._asyncQueue)
     return;
   var queue = this._asyncQueue;
+  var i;
   // This queue will be small. Probably always <= 3 items.
-  for (var i = 0; i < queue.length; i++) {
+  for (i = 0; i < queue.length; i++) {
     if (queue[i].uid === obj.uid) {
       queue.splice(i, 1);
+      this._asyncData[obj.uid] = undefined;
       return;
     }
   }
+  // Rebuild flags
+  this._asyncFlags = 0;
+  for (i = 0; i < queue.length; i++) {
+    this._asyncFlags |= queue[i].flags;
+  }
 }
 
 
@@ -418,8 +447,10 @@ Immediate.prototype.removeAsyncListener = timerRemoveAsyncListener;
 Immediate.prototype.domain = undefined;
 Immediate.prototype._onImmediate = undefined;
 Immediate.prototype._asyncQueue = undefined;
+Immediate.prototype._asyncData = undefined;
 Immediate.prototype._idleNext = undefined;
 Immediate.prototype._idlePrev = undefined;
+Immediate.prototype._asyncFlags = 0;
 
 
 exports.setImmediate = function(callback) {
@@ -446,7 +477,7 @@ exports.setImmediate = function(callback) {
   }
 
   // setImmediates are handled more like nextTicks.
-  if (asyncFlags[0] > 0)
+  if (asyncFlags[kHasListener] > 0)
     runAsyncQueue(immediate);
   if (process.domain)
     immediate.domain = process.domain;
@@ -484,7 +515,6 @@ function unrefTimeout() {
   var diff, first, hasQueue, threw;
   while (first = L.peek(unrefList)) {
     diff = now - first._idleStart;
-    hasQueue = !!first._asyncQueue;
 
     if (diff < first._idleTimeout) {
       diff = first._idleTimeout - diff;
@@ -498,6 +528,7 @@ function unrefTimeout() {
 
     if (!first._onTimeout) continue;
     if (first.domain && first.domain._disposed) continue;
+    hasQueue = !!first._asyncQueue;
 
     try {
       if (hasQueue)

diff --git a/src/async-wrap-inl.h b/src/async-wrap-inl.h
@@ -38,7 +38,7 @@ namespace node {
 inline AsyncWrap::AsyncWrap(Environment* env, v8::Handle<v8::Object> object)
     : BaseObject(env, object),
       async_flags_(NO_OPTIONS) {
-  if (!env->has_async_listeners())
+  if (!env->has_async_listener())
     return;
 
   // TODO(trevnorris): Do we really need to TryCatch this call?
@@ -49,7 +49,7 @@ inline AsyncWrap::AsyncWrap(Environment* env, v8::Handle<v8::Object> object)
   env->async_listener_run_function()->Call(env->process_object(), 1, &val);
 
   if (!try_catch.HasCaught())
-    async_flags_ |= ASYNC_LISTENERS;
+    async_flags_ |= HAS_ASYNC_LISTENER;
 }
 
 
@@ -83,8 +83,8 @@ inline void AsyncWrap::remove_flag(unsigned int flag) {
 }
 
 
-inline bool AsyncWrap::has_async_queue() {
-  return async_flags() & ASYNC_LISTENERS;
+inline bool AsyncWrap::has_async_listener() {
+  return async_flags() & HAS_ASYNC_LISTENER;
 }
 
 
@@ -100,7 +100,7 @@ inline v8::Handle<v8::Value> AsyncWrap::MakeCallback(
   v8::TryCatch try_catch;
   try_catch.SetVerbose(true);
 
-  if (has_async_queue()) {
+  if (has_async_listener()) {
     v8::Local<v8::Value> val = context.As<v8::Value>();
     env()->async_listener_load_function()->Call(process, 1, &val);
 
@@ -114,7 +114,7 @@ inline v8::Handle<v8::Value> AsyncWrap::MakeCallback(
     return Undefined(env()->isolate());
   }
 
-  if (has_async_queue()) {
+  if (has_async_listener()) {
     v8::Local<v8::Value> val = context.As<v8::Value>();
     env()->async_listener_unload_function()->Call(process, 1, &val);
 
@@ -135,9 +135,6 @@ inline v8::Handle<v8::Value> AsyncWrap::MakeCallback(
 
   tick_info->set_in_tick(true);
 
-  // TODO(trevnorris): Consider passing "context" to _tickCallback so it
-  // can then be passed as the first argument to the nextTick callback.
-  // That should greatly help needing to create closures.
   env()->tick_callback_function()->Call(process, 0, NULL);
 
   tick_info->set_in_tick(false);
@@ -191,7 +188,7 @@ inline void AsyncWrap::AddAsyncListener(
   Type* wrap = static_cast<Type*>(
       handle->GetAlignedPointerFromInternalField(0));
   assert(wrap != NULL);
-  wrap->set_flag(ASYNC_LISTENERS);
+  wrap->set_flag(HAS_ASYNC_LISTENER);
 }
 
 
@@ -213,7 +210,7 @@ inline void AsyncWrap::RemoveAsyncListener(
     Type* wrap = static_cast<Type*>(
         handle->GetAlignedPointerFromInternalField(0));
     assert(wrap != NULL);
-    wrap->remove_flag(ASYNC_LISTENERS);
+    wrap->remove_flag(HAS_ASYNC_LISTENER);
   }
 }
 

diff --git a/src/async-wrap.h b/src/async-wrap.h
@@ -32,7 +32,7 @@ class AsyncWrap : public BaseObject {
  public:
   enum AsyncFlags {
     NO_OPTIONS = 0,
-    ASYNC_LISTENERS = 1
+    HAS_ASYNC_LISTENER = 1
   };
 
   inline AsyncWrap(Environment* env, v8::Handle<v8::Object> object);
@@ -48,7 +48,7 @@ class AsyncWrap : public BaseObject {
 
   inline void remove_flag(unsigned int flag);
 
-  inline bool has_async_queue();
+  inline bool has_async_listener();
 
   // Only call these within a valid HandleScope.
   inline v8::Handle<v8::Value> MakeCallback(const v8::Handle<v8::Function> cb,
@@ -62,6 +62,8 @@ class AsyncWrap : public BaseObject {
                                             v8::Handle<v8::Value>* argv);
 
  private:
+  inline AsyncWrap();
+
   // Add an async listener to an existing handle.
   template <typename Type>
   static inline void AddAsyncListener(

diff --git a/src/env-inl.h b/src/env-inl.h
@@ -82,8 +82,8 @@ inline int Environment::AsyncListener::fields_count() const {
   return kFieldsCount;
 }
 
-inline uint32_t Environment::AsyncListener::count() const {
-  return fields_[kCount];
+inline bool Environment::AsyncListener::has_listener() const {
+  return fields_[kHasListener] > 0;
 }
 
 inline Environment::TickInfo::TickInfo() : in_tick_(false), last_threw_(false) {
@@ -188,9 +188,9 @@ inline v8::Isolate* Environment::isolate() const {
   return isolate_;
 }
 
-inline bool Environment::has_async_listeners() const {
+inline bool Environment::has_async_listener() const {
   // The const_cast is okay, it doesn't violate conceptual const-ness.
-  return const_cast<Environment*>(this)->async_listener()->count() > 0;
+  return const_cast<Environment*>(this)->async_listener()->has_listener();
 }
 
 inline Environment* Environment::from_immediate_check_handle(

diff --git a/src/env.h b/src/env.h
@@ -175,14 +175,14 @@ class Environment {
    public:
     inline uint32_t* fields();
     inline int fields_count() const;
-    inline uint32_t count() const;
+    inline bool has_listener() const;
 
    private:
     friend class Environment;  // So we can call the constructor.
     inline AsyncListener();
 
     enum Fields {
-      kCount,
+      kHasListener,
       kFieldsCount
     };
 
@@ -231,7 +231,7 @@ class Environment {
 
   inline v8::Isolate* isolate() const;
   inline uv_loop_t* event_loop() const;
-  inline bool has_async_listeners() const;
+  inline bool has_async_listener() const;
 
   static inline Environment* from_immediate_check_handle(uv_check_t* handle);
   inline uv_check_t* immediate_check_handle();