hphp/runtime/ext/asio/asio_session.cpp

   1 /*
   2    +----------------------------------------------------------------------+
   3    | HipHop for PHP                                                       |
   4    +----------------------------------------------------------------------+
   5    | Copyright (c) 2010-2013 Facebook, Inc. (http://www.facebook.com)     |
   6    | Copyright (c) 1997-2010 The PHP Group                                |
   7    +----------------------------------------------------------------------+
   8    | This source file is subject to version 3.01 of the PHP license,      |
   9    | that is bundled with this package in the file LICENSE, and is        |
  10    | available through the world-wide-web at the following url:           |
  11    | http://www.php.net/license/3_01.txt                                  |
  12    | If you did not receive a copy of the PHP license and are unable to   |
  13    | obtain it through the world-wide-web, please send a note to          |
  14    | license@php.net so we can mail you a copy immediately.               |
  15    +----------------------------------------------------------------------+
  16 */
  17
  18 #include "hphp/runtime/ext/asio/asio_session.h"
  19 #include "hphp/runtime/ext/ext_asio.h"
  20 #include "hphp/system/lib/systemlib.h"
  21
  22 namespace HPHP {
  23 ///////////////////////////////////////////////////////////////////////////////
  24
  25 IMPLEMENT_THREAD_LOCAL_PROXY(AsioSession, false, AsioSession::s_current);
  26
  27 namespace {
  28   const context_idx_t MAX_CONTEXT_DEPTH = std::numeric_limits<context_idx_t>::max();
  29 }
  30
  31 void AsioSession::Init() {
  32   s_current.set(new AsioSession());
  33 }
  34
  35 AsioSession::AsioSession()
  36     : m_contexts(), m_readyExternalThreadEvents(nullptr),
  37       m_readyExternalThreadEventsMutex(),
  38       m_readyExternalThreadEventsCondition() {
  39 }
  40
  41 void AsioSession::enterContext() {
  42   assert(!isInContext() || getCurrentContext()->isRunning());
  43
  44   if (UNLIKELY(getCurrentContextIdx() >= MAX_CONTEXT_DEPTH)) {
  45     Object e(SystemLib::AllocInvalidOperationExceptionObject(
  46       "Unable to enter asio context: too many contexts open"));
  47     throw e;
  48   }
  49
  50   m_contexts.push_back(new AsioContext());
  51
  52   assert(static_cast<context_idx_t>(m_contexts.size()) == m_contexts.size());
  53   assert(isInContext());
  54   assert(!getCurrentContext()->isRunning());
  55 }
  56
  57 void AsioSession::exitContext() {
  58   assert(isInContext());
  59   assert(!getCurrentContext()->isRunning());
  60
  61   m_contexts.back()->exit(m_contexts.size());
  62   delete m_contexts.back();
  63   m_contexts.pop_back();
  64
  65   assert(!isInContext() || getCurrentContext()->isRunning());
  66 }
  67
  68 uint16_t AsioSession::getCurrentWaitHandleDepth() {
  69   assert(!isInContext() || getCurrentContext()->isRunning());
  70   return isInContext() ? getCurrentWaitHandle()->getDepth() : 0;
  71 }
  72
  73 c_ExternalThreadEventWaitHandle* AsioSession::waitForExternalThreadEvents() {
  74   // try check for ready external thread events without grabbing lock
  75   auto ready = m_readyExternalThreadEvents.exchange(nullptr);
  76   if (ready != nullptr) {
  77     assert(ready != k_waitingForExternalThreadEvents);
  78     return ready;
  79   }
  80
  81   // no ready external thread events available, synchronization needed
  82   std::unique_lock<std::mutex> lock(m_readyExternalThreadEventsMutex);
  83
  84   // transition from empty to WAITING
  85   if (m_readyExternalThreadEvents.compare_exchange_strong(ready, k_waitingForExternalThreadEvents)) {
  86     // wait for transition from WAITING to non-empty
  87     do {
  88       m_readyExternalThreadEventsCondition.wait(lock);
  89     } while (m_readyExternalThreadEvents.load() == k_waitingForExternalThreadEvents);
  90   } else  {
  91     // external thread transitioned from empty to non-empty while grabbing lock
  92   }
  93
  94   ready = m_readyExternalThreadEvents.exchange(nullptr);
  95   assert(ready != nullptr);
  96   assert(ready != k_waitingForExternalThreadEvents);
  97   return ready;
  98 }
  99
 100 void AsioSession::enqueueExternalThreadEvent(c_ExternalThreadEventWaitHandle* wait_handle) {
 101   auto next = m_readyExternalThreadEvents.load();
 102   while (true) {
 103     while (next != k_waitingForExternalThreadEvents) {
 104       wait_handle->setNextToProcess(next);
 105       if (m_readyExternalThreadEvents.compare_exchange_weak(next, wait_handle)) {
 106         return;
 107       }
 108     }
 109
 110     // try to transition from WAITING to non-empty
 111     wait_handle->setNextToProcess(nullptr);
 112     if (m_readyExternalThreadEvents.compare_exchange_weak(next, wait_handle)) {
 113       // succeeded, notify condition
 114       std::unique_lock<std::mutex> lock(m_readyExternalThreadEventsMutex);
 115       m_readyExternalThreadEventsCondition.notify_one();
 116       return;
 117     }
 118   }
 119 }
 120
 121 void AsioSession::initAbruptInterruptException() {
 122   assert(!hasAbruptInterruptException());
 123   m_abruptInterruptException = SystemLib::AllocInvalidOperationExceptionObject(
 124     "The request was abruptly interrupted.");
 125 }
 126
 127 void AsioSession::onFailed(CObjRef exception) {
 128   if (m_onFailedCallback.get()) {
 129     try {
 130       vm_call_user_func(m_onFailedCallback, Array::Create(exception));
 131     } catch (const Object& callback_exception) {
 132       raise_warning("[asio] Ignoring exception thrown by onFailed callback");
 133     }
 134   }
 135 }
 136
 137 void AsioSession::onContinuationCreate(c_ContinuationWaitHandle* cont) {
 138   assert(m_onContinuationCreateCallback.get());
 139   try {
 140     vm_call_user_func(
 141       m_onContinuationCreateCallback,
 142       Array::Create(cont));
 143   } catch (const Object& callback_exception) {
 144     raise_warning("[asio] Ignoring exception thrown by ContinuationWaitHandle::onCreate callback");
 145   }
 146 }
 147
 148 void AsioSession::onContinuationYield(c_ContinuationWaitHandle* cont, c_WaitHandle* child) {
 149   assert(m_onContinuationYieldCallback.get());
 150   try {
 151     vm_call_user_func(
 152       m_onContinuationYieldCallback,
 153       CREATE_VECTOR2(cont, child));
 154   } catch (const Object& callback_exception) {
 155     raise_warning("[asio] Ignoring exception thrown by ContinuationWaitHandle::onYield callback");
 156   }
 157 }
 158
 159 void AsioSession::onContinuationSuccess(c_ContinuationWaitHandle* cont, CVarRef result) {
 160   assert(m_onContinuationSuccessCallback.get());
 161   try {
 162     vm_call_user_func(
 163       m_onContinuationSuccessCallback,
 164       CREATE_VECTOR2(cont, result));
 165   } catch (const Object& callback_exception) {
 166     raise_warning("[asio] Ignoring exception thrown by ContinuationWaitHandle::onSuccess callback");
 167   }
 168 }
 169
 170 void AsioSession::onContinuationFail(c_ContinuationWaitHandle* cont, CObjRef exception) {
 171   assert(m_onContinuationFailCallback.get());
 172   try {
 173     vm_call_user_func(
 174       m_onContinuationFailCallback,
 175       CREATE_VECTOR2(cont, exception));
 176   } catch (const Object& callback_exception) {
 177     raise_warning("[asio] Ignoring exception thrown by ContinuationWaitHandle::onFail callback");
 178   }
 179 }
 180
 181 void AsioSession::onJoin(c_WaitHandle* wait_handle) {
 182   assert(m_onJoinCallback.get());
 183   try {
 184     vm_call_user_func(m_onJoinCallback, Array::Create(wait_handle));
 185   } catch (const Object& callback_exception) {
 186     raise_warning("[asio] Ignoring exception thrown by WaitHandle::onJoin callback");
 187   }
 188 }
 189
 190 void AsioSession::onGenArrayCreate(c_GenArrayWaitHandle* wait_handle, CVarRef dependencies) {
 191   assert(m_onGenArrayCreateCallback.get());
 192   try {
 193     vm_call_user_func(
 194       m_onGenArrayCreateCallback,
 195       CREATE_VECTOR2(wait_handle, dependencies));
 196   } catch (const Object& callback_exception) {
 197     raise_warning("[asio] Ignoring exception thrown by GenArrayWaitHandle::onCreate callback");
 198   }
 199 }
 200
 201 void AsioSession::onGenVectorCreate(c_GenVectorWaitHandle* wait_handle, CVarRef dependencies) {
 202   assert(m_onGenVectorCreateCallback.get());
 203   try {
 204     vm_call_user_func(
 205       m_onGenVectorCreateCallback,
 206       CREATE_VECTOR2(wait_handle, dependencies));
 207   } catch (const Object& callback_exception) {
 208     raise_warning("[asio] Ignoring exception thrown by GenVectorWaitHandle::onCreate callback");
 209   }
 210 }
 211
 212 void AsioSession::onSetResultToRefCreate(c_SetResultToRefWaitHandle* wait_handle, CObjRef child) {
 213   assert(m_onSetResultToRefCreateCallback.get());
 214   try {
 215     vm_call_user_func(
 216       m_onSetResultToRefCreateCallback,
 217       CREATE_VECTOR2(wait_handle, child));
 218   } catch (const Object& callback_exception) {
 219     raise_warning("[asio] Ignoring exception thrown by SetResultToRefWaitHandle::onCreate callback");
 220   }
 221 }
 222
 223 ///////////////////////////////////////////////////////////////////////////////
 224 }