add hooks to ASIO to allow building of dependency graphs in PHP

In order to build a dependency graph of continuation execution and data-fetching in PHP-land, we need a few instrumentation points in the asio_ext HHVM extension. There are 4 additions required:
   1. Callback when a continuation finishes successfully.
   2. Callback when a continuation blocks on a wait_handle.
   3. Get array of WaitHandles a GenArrayWaitHandle is waiting on.
   4. Get WaitHandle that the SetResultToRefWaitHandle is waiting on.
I don't think this should really affect performance, as in the normal case, nothing has changed, but you never know... I'm also not sure who should be reviewing this, so I've just added @jan for now. If you could pile other people on, that would be cool.

sandcastle appears to be broken.

hphp/idl/asio.idl.json

@@ -49,7 +49,7 @@
         },
         {
             "name": "asio_set_on_failed_callback",
-            "desc": "Set callback to be called when wait handle fails",
+            "desc": "DEPRECATED: use ContinuationWaitHandle::setOnFailCallback()",
             "flags": [
                 "HasDocComment"
             ],
@@ -66,7 +66,7 @@
         },
         {
             "name": "asio_set_on_started_callback",
-            "desc": "Set callback to be called when a continuation is started",
+            "desc": "DEPRECATED: use ContinuationWaitHandle::setOnCreateCallback()",
             "flags": [
                 "HasDocComment"
             ],
@@ -107,6 +107,25 @@
                     "args": [
                     ]
                 },
+                {
+                    "name": "setOnJoinCallback",
+                    "desc": "Set callback for when join() is called",
+                    "flags": [
+                        "IsStatic",
+                        "HasDocComment",
+                        "HipHopSpecific"
+                    ],
+                    "return": {
+                        "type": null
+                    },
+                    "args": [
+                        {
+                            "name": "on_join_cb",
+                            "type": "Variant",
+                            "desc": "A Closure to be called on join()"
+                        }
+                    ]
+                },
                 {
                     "name": "getWaitHandle",
                     "desc": "Return this wait handle (for Awaitable interface)",
@@ -449,6 +468,82 @@
                     "args": [
                     ]
                 },
+                {
+                    "name": "setOnCreateCallback",
+                    "desc": "Set callback to be called when a continuation wait handle is created",
+                    "flags": [
+                        "IsStatic",
+                        "HasDocComment",
+                        "HipHopSpecific"
+                    ],
+                    "return": {
+                        "type": null
+                    },
+                    "args": [
+                        {
+                            "name": "on_create_cb",
+                            "type": "Variant",
+                            "desc": "A Closure to be called when continuation is created"
+                        }
+                    ]
+                },
+                {
+                    "name": "setOnYieldCallback",
+                    "desc": "Set callback to be called when a continuation yields",
+                    "flags": [
+                        "IsStatic",
+                        "HasDocComment",
+                        "HipHopSpecific"
+                    ],
+                    "return": {
+                        "type": null
+                    },
+                    "args": [
+                        {
+                            "name": "on_yield_cb",
+                            "type": "Variant",
+                            "desc": "A Closure to be called when continuations yields"
+                        }
+                    ]
+                },
+                {
+                    "name": "setOnSuccessCallback",
+                    "desc": "Set callback to be called when a continuation finishes successfully",
+                    "flags": [
+                        "IsStatic",
+                        "HasDocComment",
+                        "HipHopSpecific"
+                    ],
+                    "return": {
+                        "type": null
+                    },
+                    "args": [
+                        {
+                            "name": "on_success_cb",
+                            "type": "Variant",
+                            "desc": "A Closure to be called when continuations finishes"
+                        }
+                    ]
+                },
+                {
+                    "name": "setOnFailCallback",
+                    "desc": "Set callback to be called when a continuation fails",
+                    "flags": [
+                        "IsStatic",
+                        "HasDocComment",
+                        "HipHopSpecific"
+                    ],
+                    "return": {
+                        "type": null
+                    },
+                    "args": [
+                        {
+                            "name": "on_fail_cb",
+                            "type": "Variant",
+                            "desc": "A Closure to be called when continuations fails"
+                        }
+                    ]
+                },
                 {
                     "name": "getPrivData",
                     "desc": "Get user's private data",
@@ -521,6 +616,25 @@
                             "desc": "An Array of dependencies to wait for"
                         }
                     ]
+                },
+                {
+                    "name": "setOnCreateCallback",
+                    "desc": "Set callback for when a GenArrayWaitHandle is created",
+                    "flags": [
+                        "IsStatic",
+                        "HasDocComment",
+                        "HipHopSpecific"
+                    ],
+                    "return": {
+                        "type": null
+                    },
+                    "args": [
+                        {
+                            "name": "on_create_cb",
+                            "type": "Variant",
+                            "desc": "A Closure to be called on creation"
+                        }
+                    ]
                 }
             ],
             "consts": [
@@ -570,6 +684,25 @@
                             "ref": true
                         }
                     ]
+                },
+                {
+                    "name": "setOnCreateCallback",
+                    "desc": "Set callback for when a SetResultToRefWaitHandle is created",
+                    "flags": [
+                        "IsStatic",
+                        "HasDocComment",
+                        "HipHopSpecific"
+                    ],
+                    "return": {
+                        "type": null
+                    },
+                    "args": [
+                        {
+                            "name": "on_create_cb",
+                            "type": "Variant",
+                            "desc": "A Closure to be called on creation"
+                        }
+                    ]
                 }
             ],
             "consts": [

hphp/runtime/ext/asio/asio_session.cpp

@@ -35,8 +35,7 @@ void AsioSession::Init() {
 AsioSession::AsioSession()
     : m_contexts(), m_readyExternalThreadEvents(nullptr),
       m_readyExternalThreadEventsMutex(),
-      m_readyExternalThreadEventsCondition(),
-      m_onFailedCallback(nullptr), m_onStartedCallback(nullptr) {
+      m_readyExternalThreadEventsCondition() {
 }
 
 void AsioSession::enterContext() {
@@ -118,20 +117,8 @@ void AsioSession::enqueueExternalThreadEvent(c_ExternalThreadEventWaitHandle* wa
   }
 }
 
-void AsioSession::setOnFailedCallback(ObjectData* on_failed_callback) {
-  if (on_failed_callback) {
-    on_failed_callback->incRefCount();
-  }
-
-  if (m_onFailedCallback) {
-    decRefObj(m_onFailedCallback);
-  }
-
-  m_onFailedCallback = on_failed_callback;
-}
-
 void AsioSession::onFailed(CObjRef exception) {
-  if (m_onFailedCallback) {
+  if (m_onFailedCallback.get()) {
     try {
       vm_call_user_func(m_onFailedCallback, Array::Create(exception));
     } catch (const Object& callback_exception) {
@@ -140,24 +127,78 @@ void AsioSession::onFailed(CObjRef exception) {
   }
 }
 
-void AsioSession::setOnStartedCallback(ObjectData* on_started_callback) {
-  if (on_started_callback) {
-    on_started_callback->incRefCount();
+void AsioSession::onContinuationCreate(c_ContinuationWaitHandle* cont) {
+  assert(m_onContinuationCreateCallback.get());
+  try {
+    vm_call_user_func(
+      m_onContinuationCreateCallback,
+      Array::Create(cont));
+  } catch (const Object& callback_exception) {
+    raise_warning("[asio] Ignoring exception thrown by ContinuationWaitHandle::onCreate callback");
   }
-
-  if (m_onStartedCallback) {
-    decRefObj(m_onStartedCallback);
-  }
-
-  m_onStartedCallback = on_started_callback;
 }
 
-void AsioSession::onStarted(CObjRef wait_handle) {
-  assert(m_onStartedCallback);
+void AsioSession::onContinuationYield(c_ContinuationWaitHandle* cont, c_WaitHandle* child) {
+  assert(m_onContinuationYieldCallback.get());
   try {
-    vm_call_user_func(m_onStartedCallback, Array::Create(wait_handle));
+    vm_call_user_func(
+      m_onContinuationYieldCallback,
+      CREATE_VECTOR2(cont, child));
   } catch (const Object& callback_exception) {
-    raise_warning("[asio] Ignoring exception thrown by onStarted callback");
+    raise_warning("[asio] Ignoring exception thrown by ContinuationWaitHandle::onYield callback");
+  }
+}
+
+void AsioSession::onContinuationSuccess(c_ContinuationWaitHandle* cont, CVarRef result) {
+  assert(m_onContinuationSuccessCallback.get());
+  try {
+    vm_call_user_func(
+      m_onContinuationSuccessCallback,
+      CREATE_VECTOR2(cont, result));
+  } catch (const Object& callback_exception) {
+    raise_warning("[asio] Ignoring exception thrown by ContinuationWaitHandle::onSuccess callback");
+  }
+}
+
+void AsioSession::onContinuationFail(c_ContinuationWaitHandle* cont, CObjRef exception) {
+  assert(m_onContinuationFailCallback.get());
+  try {
+    vm_call_user_func(
+      m_onContinuationFailCallback,
+      CREATE_VECTOR2(cont, exception));
+  } catch (const Object& callback_exception) {
+    raise_warning("[asio] Ignoring exception thrown by ContinuationWaitHandle::onFail callback");
+  }
+}
+
+void AsioSession::onJoin(c_WaitHandle* wait_handle) {
+  assert(m_onJoinCallback.get());
+  try {
+    vm_call_user_func(m_onJoinCallback, Array::Create(wait_handle));
+  } catch (const Object& callback_exception) {
+    raise_warning("[asio] Ignoring exception thrown by WaitHandle::onJoin callback");
+  }
+}
+
+void AsioSession::onGenArrayCreate(c_GenArrayWaitHandle* wait_handle, CVarRef dependencies) {
+  assert(m_onGenArrayCreateCallback.get());
+  try {
+    vm_call_user_func(
+      m_onGenArrayCreateCallback,
+      CREATE_VECTOR2(wait_handle, dependencies));
+  } catch (const Object& callback_exception) {
+    raise_warning("[asio] Ignoring exception thrown by GenArrayWaitHandle::onCreate callback");
+  }
+}
+
+void AsioSession::onSetResultToRefCreate(c_SetResultToRefWaitHandle* wait_handle, CObjRef child) {
+  assert(m_onSetResultToRefCreateCallback.get());
+  try {
+    vm_call_user_func(
+      m_onSetResultToRefCreateCallback,
+      CREATE_VECTOR2(wait_handle, child));
+  } catch (const Object& callback_exception) {
+    raise_warning("[asio] Ignoring exception thrown by SetResultToRefWaitHandle::onCreate callback");
   }
 }
 

hphp/runtime/ext/asio/asio_session.h

@@ -23,10 +23,14 @@
 #include <mutex>
 #include <runtime/base/base_includes.h>
 #include <runtime/ext/asio/asio_context.h>
+#include <runtime/ext/ext_closure.h>
 
 namespace HPHP {
 ///////////////////////////////////////////////////////////////////////////////
 
+FORWARD_DECLARE_CLASS_BUILTIN(WaitHandle);
+FORWARD_DECLARE_CLASS_BUILTIN(GenArrayWaitHandle);
+FORWARD_DECLARE_CLASS_BUILTIN(SetResultToRefWaitHandle);
 FORWARD_DECLARE_CLASS_BUILTIN(ContinuationWaitHandle);
 FORWARD_DECLARE_CLASS_BUILTIN(ExternalThreadEventWaitHandle);
 
@@ -79,13 +83,62 @@ class AsioSession {
     void enqueueExternalThreadEvent(c_ExternalThreadEventWaitHandle* wait_handle);
 
     // callback: on failed
-    void setOnFailedCallback(ObjectData* on_failed_callback);
+    void setOnFailedCallback(ObjectData* on_failed_callback) {
+      assert(!on_failed_callback || on_failed_callback->instanceof(c_Closure::s_cls));
+      m_onFailedCallback = on_failed_callback;
+    }
     void onFailed(CObjRef exception);
 
-    // callback: on started
-    void setOnStartedCallback(ObjectData* on_started_callback);
-    void onStarted(CObjRef wait_handle);
-    bool hasOnStartedCallback() { return m_onStartedCallback; }
+    // ContinuationWaitHandle callbacks:
+    void setOnContinuationCreateCallback(ObjectData* on_start) {
+      assert(!on_start || on_start->instanceof(c_Closure::s_cls));
+      m_onContinuationCreateCallback = on_start;
+    }
+    void setOnContinuationYieldCallback(ObjectData* on_yield) {
+      assert(!on_yield || on_yield->instanceof(c_Closure::s_cls));
+      m_onContinuationYieldCallback = on_yield;
+    }
+    void setOnContinuationSuccessCallback(ObjectData* on_success) {
+      assert(!on_success || on_success->instanceof(c_Closure::s_cls));
+      m_onContinuationSuccessCallback = on_success;
+    }
+    void setOnContinuationFailCallback(ObjectData* on_fail) {
+      assert(!on_fail || on_fail->instanceof(c_Closure::s_cls));
+      m_onContinuationFailCallback = on_fail;
+    }
+    bool hasOnContinuationCreateCallback() { return m_onContinuationCreateCallback.get(); }
+    bool hasOnContinuationYieldCallback() { return m_onContinuationYieldCallback.get(); }
+    bool hasOnContinuationSuccessCallback() { return m_onContinuationSuccessCallback.get(); }
+    bool hasOnContinuationFailCallback() { return m_onContinuationFailCallback.get(); }
+    void onContinuationCreate(c_ContinuationWaitHandle* cont);
+    void onContinuationYield(c_ContinuationWaitHandle* cont, c_WaitHandle* child);
+    void onContinuationSuccess(c_ContinuationWaitHandle* cont, CVarRef result);
+    void onContinuationFail(c_ContinuationWaitHandle* cont, CObjRef exception);
+
+    // WaitHandle callbacks:
+    void setOnJoinCallback(ObjectData* on_join) {
+      assert(!on_join || on_join->instanceof(c_Closure::s_cls));
+      m_onJoinCallback = on_join;
+    }
+    bool hasOnJoinCallback() { return m_onJoinCallback.get(); }
+    void onJoin(c_WaitHandle* wait_handle);
+
+    // GenArrayWaitHandle callbacks:
+    void setOnGenArrayCreateCallback(ObjectData* on_create) {
+      assert(!on_create || on_create->instanceof(c_Closure::s_cls));
+      m_onGenArrayCreateCallback = on_create;
+    }
+    bool hasOnGenArrayCreateCallback() { return m_onGenArrayCreateCallback.get(); }
+    void onGenArrayCreate(c_GenArrayWaitHandle* wait_handle, CVarRef dependencies);
+
+    // SetResultToRefWaitHandle callbacks:
+    void setOnSetResultToRefCreateCallback(ObjectData* on_create) {
+      assert(!on_create || on_create->instanceof(c_Closure::s_cls));
+      m_onSetResultToRefCreateCallback = on_create;
+    }
+    bool hasOnSetResultToRefCreateCallback() { return m_onSetResultToRefCreateCallback.get(); }
+    void onSetResultToRefCreate(c_SetResultToRefWaitHandle* wait_handle, CObjRef child);
+
 
   private:
     static DECLARE_THREAD_LOCAL_PROXY(AsioSession, false, s_current);
@@ -100,8 +153,16 @@ class AsioSession {
     std::mutex m_readyExternalThreadEventsMutex;
     std::condition_variable m_readyExternalThreadEventsCondition;
 
-    ObjectData* m_onFailedCallback;
-    ObjectData* m_onStartedCallback;
+    Object m_onContinuationCreateCallback;
+    Object m_onContinuationYieldCallback;
+    Object m_onContinuationSuccessCallback;
+    Object m_onContinuationFailCallback;
+    Object m_onGenArrayCreateCallback;
+    Object m_onSetResultToRefCreateCallback;
+    Object m_onJoinCallback;
+
+    // Legacy callback for backwards compatibility.
+    Object m_onFailedCallback;
 };
 
 ///////////////////////////////////////////////////////////////////////////////

hphp/runtime/ext/asio/continuation_wait_handle.cpp

@@ -16,6 +16,7 @@
 */
 
 #include <runtime/ext/ext_asio.h>
+#include <runtime/ext/ext_closure.h>
 #include <runtime/ext/ext_continuation.h>
 #include <runtime/ext/asio/asio_context.h>
 #include <runtime/ext/asio/asio_session.h>
@@ -47,6 +48,42 @@ void c_ContinuationWaitHandle::t___construct() {
   throw e;
 }
 
+void c_ContinuationWaitHandle::ti_setoncreatecallback(CVarRef callback) {
+  if (!callback.isNull() && !callback.instanceof(c_Closure::s_cls)) {
+    Object e(SystemLib::AllocInvalidArgumentExceptionObject(
+      "Unable to set ContinuationWaitHandle::onStart: on_start_cb not a closure"));
+    throw e;
+  }
+  AsioSession::Get()->setOnContinuationCreateCallback(callback.getObjectDataOrNull());
+}
+
+void c_ContinuationWaitHandle::ti_setonyieldcallback(CVarRef callback) {
+  if (!callback.isNull() && !callback.instanceof(c_Closure::s_cls)) {
+    Object e(SystemLib::AllocInvalidArgumentExceptionObject(
+      "Unable to set ContinuationWaitHandle::onYield: on_yield_cb not a closure"));
+    throw e;
+  }
+  AsioSession::Get()->setOnContinuationYieldCallback(callback.getObjectDataOrNull());
+}
+
+void c_ContinuationWaitHandle::ti_setonsuccesscallback(CVarRef callback) {
+  if (!callback.isNull() && !callback.instanceof(c_Closure::s_cls)) {
+    Object e(SystemLib::AllocInvalidArgumentExceptionObject(
+      "Unable to set ContinuationWaitHandle::onSuccess: on_success_cb not a closure"));
+    throw e;
+  }
+  AsioSession::Get()->setOnContinuationSuccessCallback(callback.getObjectDataOrNull());
+}
+
+void c_ContinuationWaitHandle::ti_setonfailcallback(CVarRef callback) {
+  if (!callback.isNull() && !callback.instanceof(c_Closure::s_cls)) {
+    Object e(SystemLib::AllocInvalidArgumentExceptionObject(
+      "Unable to set ContinuationWaitHandle::onFail: on_fail_cb not a closure"));
+    throw e;
+  }
+  AsioSession::Get()->setOnContinuationFailCallback(callback.getObjectDataOrNull());
+}
+
 void c_ContinuationWaitHandle::Create(c_Continuation* continuation) {
   assert(continuation);
   assert(continuation->m_waitHandle.isNull());
@@ -71,8 +108,8 @@ void c_ContinuationWaitHandle::Create(c_Continuation* continuation) {
   continuation->m_waitHandle->initialize(continuation, depth + 1);
 
   // needs to be called after continuation->m_waitHandle is set
-  if (UNLIKELY(session->hasOnStartedCallback())) {
-    session->onStarted(continuation->m_waitHandle);
+  if (UNLIKELY(session->hasOnContinuationCreateCallback())) {
+    session->onContinuationCreate(continuation->m_waitHandle.get());
   }
 }
 
@@ -139,6 +176,12 @@ void c_ContinuationWaitHandle::run() {
               "Expected yield argument to be an instance of WaitHandle"));
           throw e;
         }
+
+        AsioSession* session = AsioSession::Get();
+        if (UNLIKELY(session->hasOnContinuationYieldCallback())) {
+          session->onContinuationYield(this, child);
+        }
+
         m_child = child;
       }
     } while (m_child.isNull() || m_child->isFinished());
@@ -160,6 +203,11 @@ void c_ContinuationWaitHandle::onUnblocked() {
 }
 
 void c_ContinuationWaitHandle::markAsSucceeded(const TypedValue* result) {
+  AsioSession* session = AsioSession::Get();
+  if (UNLIKELY(session->hasOnContinuationSuccessCallback())) {
+    session->onContinuationSuccess(this, tvAsCVarRef(result));
+  }
+
   setResult(result);
 
   // free m_continuation / m_child later, result may be stored there
@@ -168,7 +216,11 @@ void c_ContinuationWaitHandle::markAsSucceeded(const TypedValue* result) {
 }
 
 void c_ContinuationWaitHandle::markAsFailed(CObjRef exception) {
-  AsioSession::Get()->onFailed(exception);
+  AsioSession* session = AsioSession::Get();
+  session->onFailed(exception);
+  if (UNLIKELY(session->hasOnContinuationFailCallback())) {
+    session->onContinuationFail(this, exception);
+  }
   setException(exception.get());
 
   m_continuation = nullptr;

hphp/runtime/ext/asio/gen_array_wait_handle.cpp

@@ -16,6 +16,7 @@
 */
 
 #include <runtime/ext/ext_asio.h>
+#include <runtime/ext/ext_closure.h>
 #include <runtime/ext/asio/asio_context.h>
 #include <runtime/ext/asio/asio_session.h>
 #include <system/lib/systemlib.h>
@@ -49,6 +50,15 @@ void c_GenArrayWaitHandle::t___construct() {
   throw e;
 }
 
+void c_GenArrayWaitHandle::ti_setoncreatecallback(CVarRef callback) {
+  if (!callback.isNull() && !callback.instanceof(c_Closure::s_cls)) {
+    Object e(SystemLib::AllocInvalidArgumentExceptionObject(
+      "Unable to set GenArrayWaitHandle::onCreate: on_create_cb not a closure"));
+    throw e;
+  }
+  AsioSession::Get()->setOnGenArrayCreateCallback(callback.getObjectDataOrNull());
+}
+
 Object c_GenArrayWaitHandle::ti_create(CArrRef dependencies) {
   Array deps = dependencies->copy();
   for (ssize_t iter_pos = deps->iter_begin();
@@ -92,8 +102,14 @@ Object c_GenArrayWaitHandle::ti_create(CArrRef dependencies) {
       assert(dynamic_cast<c_WaitableWaitHandle*>(child));
       auto child_wh = static_cast<c_WaitableWaitHandle*>(child);
 
-      c_GenArrayWaitHandle* my_wh = NEWOBJ(c_GenArrayWaitHandle)();
+      p_GenArrayWaitHandle my_wh = NEWOBJ(c_GenArrayWaitHandle)();
       my_wh->initialize(exception, deps, iter_pos, child_wh);
+
+      AsioSession* session = AsioSession::Get();
+      if (UNLIKELY(session->hasOnGenArrayCreateCallback())) {
+        session->onGenArrayCreate(my_wh.get(), dependencies);
+      }
+
       return my_wh;
     }
   }

hphp/runtime/ext/asio/set_result_to_ref_wait_handle.cpp

@@ -16,6 +16,7 @@
 */
 
 #include <runtime/ext/ext_asio.h>
+#include <runtime/ext/ext_closure.h>
 #include <runtime/ext/asio/asio_context.h>
 #include <runtime/ext/asio/asio_session.h>
 #include <system/lib/systemlib.h>
@@ -43,6 +44,15 @@ void c_SetResultToRefWaitHandle::t___construct() {
   throw e;
 }
 
+void c_SetResultToRefWaitHandle::ti_setoncreatecallback(CVarRef callback) {
+  if (!callback.isNull() && !callback.instanceof(c_Closure::s_cls)) {
+    Object e(SystemLib::AllocInvalidArgumentExceptionObject(
+      "Unable to set SetResultToRefWaitHandle::onCreate: on_create_cb not a closure"));
+    throw e;
+  }
+  AsioSession::Get()->setOnSetResultToRefCreateCallback(callback.getObjectDataOrNull());
+}
+
 Object c_SetResultToRefWaitHandle::ti_create(CObjRef wait_handle, VRefParam ref) {
   TypedValue* var_or_cell = ref->asTypedValue();
   if (wait_handle.isNull()) {
@@ -78,8 +88,14 @@ Object c_SetResultToRefWaitHandle::ti_create(CObjRef wait_handle, VRefParam ref)
     tvBox(var_or_cell);
   }
 
-  c_SetResultToRefWaitHandle* my_wh = NEWOBJ(c_SetResultToRefWaitHandle)();
+  p_SetResultToRefWaitHandle my_wh = NEWOBJ(c_SetResultToRefWaitHandle)();
   my_wh->initialize(child_wh, var_or_cell->m_data.pref);
+
+  AsioSession* session = AsioSession::Get();
+  if (UNLIKELY(session->hasOnSetResultToRefCreateCallback())) {
+    session->onSetResultToRefCreate(my_wh.get(), child_wh);
+  }
+
   return my_wh;
 }
 

hphp/runtime/ext/asio/wait_handle.cpp

@@ -16,6 +16,7 @@
 */
 
 #include <runtime/ext/ext_asio.h>
+#include <runtime/ext/ext_closure.h>
 #include <runtime/ext/asio/asio_session.h>
 
 namespace HPHP {
@@ -32,6 +33,15 @@ void c_WaitHandle::t___construct() {
   throw NotSupportedException(__func__, "WTF? This is an abstract class");
 }
 
+void c_WaitHandle::ti_setonjoincallback(CVarRef callback) {
+  if (!callback.isNull() && !callback.instanceof(c_Closure::s_cls)) {
+    Object e(SystemLib::AllocInvalidArgumentExceptionObject(
+      "Unable to set WaitHandle::onJoin: on_join_cb not a closure"));
+    throw e;
+  }
+  AsioSession::Get()->setOnJoinCallback(callback.getObjectDataOrNull());
+}
+
 Object c_WaitHandle::t_getwaithandle() {
   return this;
 }

hphp/runtime/ext/asio/waitable_wait_handle.cpp

@@ -134,6 +134,10 @@ void c_WaitableWaitHandle::join() {
   assert(!isFinished());
   assert(!session->isInContext() || session->getCurrentContext()->isRunning());
 
+  if (UNLIKELY(session->hasOnJoinCallback())) {
+    session->onJoin(this);
+  }
+
   // enter new asio context and set up guard that will exit once we are done
   session->enterContext();
 

hphp/runtime/ext/ext_asio.cpp

@@ -57,18 +57,11 @@ void f_asio_set_on_failed_callback(CVarRef on_failed_cb) {
       "Unable to set asio on failed callback: on_failed_cb not a closure"));
     throw e;
   }
-
   AsioSession::Get()->setOnFailedCallback(on_failed_cb.getObjectDataOrNull());
 }
 
 void f_asio_set_on_started_callback(CVarRef on_started_cb) {
-  if (!on_started_cb.isNull() && !on_started_cb.instanceof(c_Closure::s_cls)) {
-    Object e(SystemLib::AllocInvalidArgumentExceptionObject(
-      "Unable to set asio on started callback: on_started_cb not a closure"));
-    throw e;
-  }
-
-  AsioSession::Get()->setOnStartedCallback(on_started_cb.getObjectDataOrNull());
+  c_ContinuationWaitHandle::ti_setoncreatecallback(on_started_cb);
 }
 
 ///////////////////////////////////////////////////////////////////////////////

hphp/runtime/ext/ext_asio.h

@@ -63,6 +63,7 @@ class c_WaitHandle : public ExtObjectData {
   public: c_WaitHandle(VM::Class* cls = c_WaitHandle::s_cls);
   public: ~c_WaitHandle();
   public: void t___construct();
+  public: static void ti_setonjoincallback(CVarRef callback);
   public: Object t_getwaithandle();
   public: void t_import();
   public: Variant t_join();
@@ -272,10 +273,13 @@ class c_ContinuationWaitHandle : public c_BlockableWaitHandle {
   public: c_ContinuationWaitHandle(VM::Class* cls = c_ContinuationWaitHandle::s_cls);
   public: ~c_ContinuationWaitHandle();
   public: void t___construct();
+  public: static void ti_setoncreatecallback(CVarRef callback);
+  public: static void ti_setonyieldcallback(CVarRef callback);
+  public: static void ti_setonsuccesscallback(CVarRef callback);
+  public: static void ti_setonfailcallback(CVarRef callback);
   public: Object t_getprivdata();
   public: void t_setprivdata(CObjRef data);
 
-
  public:
   static void Create(c_Continuation* continuation);
   void run();
@@ -320,8 +324,10 @@ class c_GenArrayWaitHandle : public c_BlockableWaitHandle {
   public: c_GenArrayWaitHandle(VM::Class* cls = c_GenArrayWaitHandle::s_cls);
   public: ~c_GenArrayWaitHandle();
   public: void t___construct();
+  public: static void ti_setoncreatecallback(CVarRef callback);
   public: static Object ti_create(CArrRef dependencies);
 
+
  public:
   String getName();
   void enterContext(context_idx_t ctx_idx);
@@ -354,8 +360,10 @@ class c_SetResultToRefWaitHandle : public c_BlockableWaitHandle {
   public: c_SetResultToRefWaitHandle(VM::Class* cls = c_SetResultToRefWaitHandle::s_cls);
   public: ~c_SetResultToRefWaitHandle();
   public: void t___construct();
+  public: static void ti_setoncreatecallback(CVarRef callback);
   public: static Object ti_create(CObjRef wait_handle, VRefParam ref);
 
+
  public:
   String getName();
   void enterContext(context_idx_t ctx_idx);

hphp/system/class_map.cpp

@@ -13815,13 +13815,13 @@ const char *g_class_map[] = {
   NULL,
   NULL,
   (const char *)0x10006040, "asio_set_on_failed_callback", "", (const char*)0, (const char*)0,
-  "/**\n * ( excerpt from\n * http://php.net/manual/en/function.asio-set-on-failed-callback.php )\n *\n * Set callback to be called when wait handle fails\n *\n * @on_failed_cb\n *             mixed   A Closure to be called when wait handle fails\n */",
+  "/**\n * ( excerpt from\n * http://php.net/manual/en/function.asio-set-on-failed-callback.php )\n *\n * DEPRECATED: use ContinuationWaitHandle::setOnFailCallback()\n *\n * @on_failed_cb\n *             mixed   A Closure to be called when wait handle fails\n */",
   (const char *)0x8 /* KindOfNull */, (const char *)0x2000, "on_failed_cb", "", (const char *)0xffffffff /* KindOfUnknown: $t: Variant */, "", (const char *)0, "", (const char *)0, NULL,
   NULL,
   NULL,
   NULL,
   (const char *)0x10006040, "asio_set_on_started_callback", "", (const char*)0, (const char*)0,
-  "/**\n * ( excerpt from\n * http://php.net/manual/en/function.asio-set-on-started-callback.php )\n *\n * Set callback to be called when a continuation is started\n *\n * @on_started_cb\n *             mixed   A Closure to be called when wait handle is started\n */",
+  "/**\n * ( excerpt from\n * http://php.net/manual/en/function.asio-set-on-started-callback.php )\n *\n * DEPRECATED: use ContinuationWaitHandle::setOnCreateCallback()\n *\n * @on_started_cb\n *             mixed   A Closure to be called when wait handle is started\n */",
   (const char *)0x8 /* KindOfNull */, (const char *)0x2000, "on_started_cb", "", (const char *)0xffffffff /* KindOfUnknown: $t: Variant */, "", (const char *)0, "", (const char *)0, NULL,
   NULL,
   NULL,
@@ -24708,6 +24708,12 @@ const char *g_class_map[] = {
   (const char *)0x8 /* KindOfNull */, NULL,
   NULL,
   NULL,
+  (const char *)0x10016240, "setOnJoinCallback", "", (const char*)0, (const char*)0,
+  "/**\n * ( HipHop specific )\n *\n * Set callback for when join() is called\n *\n * @on_join_cb mixed   A Closure to be called on join()\n */",
+  (const char *)0x8 /* KindOfNull */, (const char *)0x2000, "on_join_cb", "", (const char *)0xffffffff /* KindOfUnknown: $t: Variant */, "", (const char *)0, "", (const char *)0, NULL,
+  NULL,
+  NULL,
+  NULL,
   (const char *)0x10006040, "getWaitHandle", "", (const char*)0, (const char*)0,
   "/**\n * ( excerpt from http://php.net/manual/en/waithandle.getwaithandle.php )\n *\n * Return this wait handle (for Awaitable interface)\n *\n * @return     object\n */",
   (const char *)0x40 /* KindOfObject */, NULL,
@@ -24852,6 +24858,30 @@ const char *g_class_map[] = {
   (const char *)0x8 /* KindOfNull */, NULL,
   NULL,
   NULL,
+  (const char *)0x10016240, "setOnCreateCallback", "", (const char*)0, (const char*)0,
+  "/**\n * ( HipHop specific )\n *\n * Set callback to be called when a continuation wait handle is created\n *\n * @on_create_cb\n *             mixed   A Closure to be called when continuation is created\n */",
+  (const char *)0x8 /* KindOfNull */, (const char *)0x2000, "on_create_cb", "", (const char *)0xffffffff /* KindOfUnknown: $t: Variant */, "", (const char *)0, "", (const char *)0, NULL,
+  NULL,
+  NULL,
+  NULL,
+  (const char *)0x10016240, "setOnYieldCallback", "", (const char*)0, (const char*)0,
+  "/**\n * ( HipHop specific )\n *\n * Set callback to be called when a continuation yields\n *\n * @on_yield_cb\n *             mixed   A Closure to be called when continuations yields\n */",
+  (const char *)0x8 /* KindOfNull */, (const char *)0x2000, "on_yield_cb", "", (const char *)0xffffffff /* KindOfUnknown: $t: Variant */, "", (const char *)0, "", (const char *)0, NULL,
+  NULL,
+  NULL,
+  NULL,
+  (const char *)0x10016240, "setOnSuccessCallback", "", (const char*)0, (const char*)0,
+  "/**\n * ( HipHop specific )\n *\n * Set callback to be called when a continuation finishes successfully\n *\n * @on_success_cb\n *             mixed   A Closure to be called when continuations finishes\n */",
+  (const char *)0x8 /* KindOfNull */, (const char *)0x2000, "on_success_cb", "", (const char *)0xffffffff /* KindOfUnknown: $t: Variant */, "", (const char *)0, "", (const char *)0, NULL,
+  NULL,
+  NULL,
+  NULL,
+  (const char *)0x10016240, "setOnFailCallback", "", (const char*)0, (const char*)0,
+  "/**\n * ( HipHop specific )\n *\n * Set callback to be called when a continuation fails\n *\n * @on_fail_cb mixed   A Closure to be called when continuations fails\n */",
+  (const char *)0x8 /* KindOfNull */, (const char *)0x2000, "on_fail_cb", "", (const char *)0xffffffff /* KindOfUnknown: $t: Variant */, "", (const char *)0, "", (const char *)0, NULL,
+  NULL,
+  NULL,
+  NULL,
   (const char *)0x10006040, "getPrivData", "", (const char*)0, (const char*)0,
   "/**\n * ( excerpt from\n * http://php.net/manual/en/continuationwaithandle.getprivdata.php )\n *\n * Get user's private data\n *\n * @return     object  An Object with user's private data\n */",
   (const char *)0x40 /* KindOfObject */, NULL,
@@ -24881,6 +24911,12 @@ const char *g_class_map[] = {
   NULL,
   NULL,
   NULL,
+  (const char *)0x10016240, "setOnCreateCallback", "", (const char*)0, (const char*)0,
+  "/**\n * ( HipHop specific )\n *\n * Set callback for when a GenArrayWaitHandle is created\n *\n * @on_create_cb\n *             mixed   A Closure to be called on creation\n */",
+  (const char *)0x8 /* KindOfNull */, (const char *)0x2000, "on_create_cb", "", (const char *)0xffffffff /* KindOfUnknown: $t: Variant */, "", (const char *)0, "", (const char *)0, NULL,
+  NULL,
+  NULL,
+  NULL,
   NULL,
   NULL,
   NULL,
@@ -24900,6 +24936,12 @@ const char *g_class_map[] = {
   NULL,
   NULL,
   NULL,
+  (const char *)0x10016240, "setOnCreateCallback", "", (const char*)0, (const char*)0,
+  "/**\n * ( HipHop specific )\n *\n * Set callback for when a SetResultToRefWaitHandle is created\n *\n * @on_create_cb\n *             mixed   A Closure to be called on creation\n */",
+  (const char *)0x8 /* KindOfNull */, (const char *)0x2000, "on_create_cb", "", (const char *)0xffffffff /* KindOfUnknown: $t: Variant */, "", (const char *)0, "", (const char *)0, NULL,
+  NULL,
+  NULL,
+  NULL,
   NULL,
   NULL,
   NULL,