开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 游戏 > 游戏引擎 > 查看源码
llpumpio.cpp
资源名称：snowglobe-src-viewer-2.0.0-r0.tar.gz　[点击查看]
上传用户：king477883
上传日期：2021-03-01
资源大小：9553k
文件大小：28k
源码类别：

游戏引擎

开发平台：

C++ Builder

llpumpio.cpp：源码内容
  1. /** 
  2.  * @file llpumpio.cpp
  3.  * @author Phoenix
  4.  * @date 2004-11-21
  5.  * @brief Implementation of the i/o pump and related functions.
  6.  *
  7.  * $LicenseInfo:firstyear=2004&license=viewergpl$
  8.  * 
  9.  * Copyright (c) 2004-2010, Linden Research, Inc.
  10.  * 
  11.  * Second Life Viewer Source Code
  12.  * The source code in this file ("Source Code") is provided by Linden Lab
  13.  * to you under the terms of the GNU General Public License, version 2.0
  14.  * ("GPL"), unless you have obtained a separate licensing agreement
  15.  * ("Other License"), formally executed by you and Linden Lab.  Terms of
  16.  * the GPL can be found in doc/GPL-license.txt in this distribution, or
  17.  * online at http://secondlifegrid.net/programs/open_source/licensing/gplv2
  18.  * 
  19.  * There are special exceptions to the terms and conditions of the GPL as
  20.  * it is applied to this Source Code. View the full text of the exception
  21.  * in the file doc/FLOSS-exception.txt in this software distribution, or
  22.  * online at
  23.  * http://secondlifegrid.net/programs/open_source/licensing/flossexception
  24.  * 
  25.  * By copying, modifying or distributing this software, you acknowledge
  26.  * that you have read and understood your obligations described above,
  27.  * and agree to abide by those obligations.
  28.  * 
  29.  * ALL LINDEN LAB SOURCE CODE IS PROVIDED "AS IS." LINDEN LAB MAKES NO
  30.  * WARRANTIES, EXPRESS, IMPLIED OR OTHERWISE, REGARDING ITS ACCURACY,
  31.  * COMPLETENESS OR PERFORMANCE.
  32.  * $/LicenseInfo$
  33.  */
  35. #include "linden_common.h"
  36. #include "llpumpio.h"
  38. #include <map>
  39. #include <set>
  40. #include "apr_poll.h"
  42. #include "llapr.h"
  43. #include "llmemtype.h"
  44. #include "llstl.h"
  45. #include "llstat.h"
  47. // These should not be enabled in production, but they can be
  48. // intensely useful during development for finding certain kinds of
  49. // bugs.
  50. #if LL_LINUX
  51. //#define LL_DEBUG_PIPE_TYPE_IN_PUMP 1
  52. //#define LL_DEBUG_POLL_FILE_DESCRIPTORS 1
  53. #if LL_DEBUG_POLL_FILE_DESCRIPTORS
  54. #include "apr_portable.h"
  55. #endif
  56. #endif
  58. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  59. #include <typeinfo>
  60. #endif
  62. // constants for poll timeout. if we are threading, we want to have a
  63. // longer poll timeout.
  64. #if LL_THREADS_APR
  65. static const S32 DEFAULT_POLL_TIMEOUT = 1000;
  66. #else
  67. static const S32 DEFAULT_POLL_TIMEOUT = 0;
  68. #endif
  70. // The default (and fallback) expiration time for chains
  71. const F32 DEFAULT_CHAIN_EXPIRY_SECS = 30.0f;
  72. extern const F32 SHORT_CHAIN_EXPIRY_SECS = 1.0f;
  73. extern const F32 NEVER_CHAIN_EXPIRY_SECS = 0.0f;
  75. // sorta spammy debug modes.
  76. //#define LL_DEBUG_SPEW_BUFFER_CHANNEL_IN_ON_ERROR 1
  77. //#define LL_DEBUG_PROCESS_LINK 1
  78. //#define LL_DEBUG_PROCESS_RETURN_VALUE 1
  80. // Super spammy debug mode.
  81. //#define LL_DEBUG_SPEW_BUFFER_CHANNEL_IN 1
  82. //#define LL_DEBUG_SPEW_BUFFER_CHANNEL_OUT 1
  84. //
  85. // local functions
  86. //
  87. void ll_debug_poll_fd(const char* msg, const apr_pollfd_t* poll)
  88. {
  89. #if LL_DEBUG_POLL_FILE_DESCRIPTORS
  90. if(!poll)
  91. {
  92. lldebugs << "Poll -- " << (msg?msg:"") << ": no pollfd." << llendl;
  93. return;
  94. }
  95. if(poll->desc.s)
  96. {
  97. apr_os_sock_t os_sock;
  98. if(APR_SUCCESS == apr_os_sock_get(&os_sock, poll->desc.s))
  99. {
  100. lldebugs << "Poll -- " << (msg?msg:"") << " on fd " << os_sock
  101.  << " at " << poll->desc.s << llendl;
  102. }
  103. else
  104. {
  105. lldebugs << "Poll -- " << (msg?msg:"") << " no fd "
  106.  << " at " << poll->desc.s << llendl;
  107. }
  108. }
  109. else if(poll->desc.f)
  110. {
  111. apr_os_file_t os_file;
  112. if(APR_SUCCESS == apr_os_file_get(&os_file, poll->desc.f))
  113. {
  114. lldebugs << "Poll -- " << (msg?msg:"") << " on fd " << os_file
  115.  << " at " << poll->desc.f << llendl;
  116. }
  117. else
  118. {
  119. lldebugs << "Poll -- " << (msg?msg:"") << " no fd "
  120.  << " at " << poll->desc.f << llendl;
  121. }
  122. }
  123. else
  124. {
  125. lldebugs << "Poll -- " << (msg?msg:"") << ": no descriptor." << llendl;
  126. }
  127. #endif
  128. }
  130. /**
  131.  * @class
  132.  */
  133. class LLChainSleeper : public LLRunnable
  134. {
  135. public:
  136. static LLRunner::run_ptr_t build(LLPumpIO* pump, S32 key)
  137. {
  138. return LLRunner::run_ptr_t(new LLChainSleeper(pump, key));
  139. }
  141. virtual void run(LLRunner* runner, S64 handle)
  142. {
  143. mPump->clearLock(mKey);
  144. }
  146. protected:
  147. LLChainSleeper(LLPumpIO* pump, S32 key) : mPump(pump), mKey(key) {}
  148. LLPumpIO* mPump;
  149. S32 mKey;
  150. };
  153. /**
  154.  * @struct ll_delete_apr_pollset_fd_client_data
  155.  * @brief This is a simple helper class to clean up our client data.
  156.  */
  157. struct ll_delete_apr_pollset_fd_client_data
  158. {
  159. typedef std::pair<LLIOPipe::ptr_t, apr_pollfd_t> pipe_conditional_t;
  160. void operator()(const pipe_conditional_t& conditional)
  161. {
  162. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  163. S32* client_id = (S32*)conditional.second.client_data;
  164. delete client_id;
  165. }
  166. };
  168. /**
  169.  * LLPumpIO
  170.  */
  171. LLPumpIO::LLPumpIO(apr_pool_t* pool) :
  172. mState(LLPumpIO::NORMAL),
  173. mRebuildPollset(false),
  174. mPollset(NULL),
  175. mPollsetClientID(0),
  176. mNextLock(0),
  177. mPool(NULL),
  178. mCurrentPool(NULL),
  179. mCurrentPoolReallocCount(0),
  180. mChainsMutex(NULL),
  181. mCallbackMutex(NULL),
  182. mCurrentChain(mRunningChains.end())
  183. {
  184. mCurrentChain = mRunningChains.end();
  186. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  187. initialize(pool);
  188. }
  190. LLPumpIO::~LLPumpIO()
  191. {
  192. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  193. cleanup();
  194. }
  196. bool LLPumpIO::prime(apr_pool_t* pool)
  197. {
  198. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  199. cleanup();
  200. initialize(pool);
  201. return ((pool == NULL) ? false : true);
  202. }
  204. bool LLPumpIO::addChain(const chain_t& chain, F32 timeout)
  205. {
  206. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  207. if(chain.empty()) return false;
  209. #if LL_THREADS_APR
  210. LLScopedLock lock(mChainsMutex);
  211. #endif
  212. LLChainInfo info;
  213. info.setTimeoutSeconds(timeout);
  214. info.mData = LLIOPipe::buffer_ptr_t(new LLBufferArray);
  215. LLLinkInfo link;
  216. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  217. lldebugs << "LLPumpIO::addChain() " << chain[0] << " '"
  218. << typeid(*(chain[0])).name() << "'" << llendl;
  219. #else
  220. lldebugs << "LLPumpIO::addChain() " << chain[0] <<llendl;
  221. #endif
  222. chain_t::const_iterator it = chain.begin();
  223. chain_t::const_iterator end = chain.end();
  224. for(; it != end; ++it)
  225. {
  226. link.mPipe = (*it);
  227. link.mChannels = info.mData->nextChannel();
  228. info.mChainLinks.push_back(link);
  229. }
  230. mPendingChains.push_back(info);
  231. return true;
  232. }
  234. bool LLPumpIO::addChain(
  235. const LLPumpIO::links_t& links,
  236. LLIOPipe::buffer_ptr_t data,
  237. LLSD context,
  238. F32 timeout)
  239. {
  240. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  242. // remember that if the caller is providing a full link
  243. // description, we need to have that description matched to a
  244. // particular buffer.
  245. if(!data) return false;
  246. if(links.empty()) return false;
  248. #if LL_THREADS_APR
  249. LLScopedLock lock(mChainsMutex);
  250. #endif
  251. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  252. lldebugs << "LLPumpIO::addChain() " << links[0].mPipe << " '"
  253. << typeid(*(links[0].mPipe)).name() << "'" << llendl;
  254. #else
  255. lldebugs << "LLPumpIO::addChain() " << links[0].mPipe << llendl;
  256. #endif
  257. LLChainInfo info;
  258. info.setTimeoutSeconds(timeout);
  259. info.mChainLinks = links;
  260. info.mData = data;
  261. info.mContext = context;
  262. mPendingChains.push_back(info);
  263. return true;
  264. }
  266. bool LLPumpIO::setTimeoutSeconds(F32 timeout)
  267. {
  268. // If no chain is running, return failure.
  269. if(mRunningChains.end() == mCurrentChain)
  270. {
  271. return false;
  272. }
  273. (*mCurrentChain).setTimeoutSeconds(timeout);
  274. return true;
  275. }
  277. void LLPumpIO::adjustTimeoutSeconds(F32 delta)
  278. {
  279. // Ensure a chain is running
  280. if(mRunningChains.end() != mCurrentChain)
  281. {
  282. (*mCurrentChain).adjustTimeoutSeconds(delta);
  283. }
  284. }
  286. static std::string events_2_string(apr_int16_t events)
  287. {
  288. std::ostringstream ostr;
  289. if(events & APR_POLLIN)
  290. {
  291. ostr << "read,";
  292. }
  293. if(events & APR_POLLPRI)
  294. {
  295. ostr << "priority,";
  296. }
  297. if(events & APR_POLLOUT)
  298. {
  299. ostr << "write,";
  300. }
  301. if(events & APR_POLLERR)
  302. {
  303. ostr << "error,";
  304. }
  305. if(events & APR_POLLHUP)
  306. {
  307. ostr << "hangup,";
  308. }
  309. if(events & APR_POLLNVAL)
  310. {
  311. ostr << "invalid,";
  312. }
  313. return chop_tail_copy(ostr.str(), 1);
  314. }
  316. bool LLPumpIO::setConditional(LLIOPipe* pipe, const apr_pollfd_t* poll)
  317. {
  318. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  319. if(!pipe) return false;
  320. ll_debug_poll_fd("Set conditional", poll);
  322. lldebugs << "Setting conditionals (" << (poll ? events_2_string(poll->reqevents) :"null")
  323.  << ") "
  324. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  325.  << "on pipe " << typeid(*pipe).name() 
  326. #endif
  327.  << " at " << pipe << llendl;
  329. // remove any matching poll file descriptors for this pipe.
  330. LLIOPipe::ptr_t pipe_ptr(pipe);
  331. LLChainInfo::conditionals_t::iterator it;
  332. it = (*mCurrentChain).mDescriptors.begin();
  333. while(it != (*mCurrentChain).mDescriptors.end())
  334. {
  335. LLChainInfo::pipe_conditional_t& value = (*it);
  336. if(pipe_ptr == value.first)
  337. {
  338. ll_delete_apr_pollset_fd_client_data()(value);
  339. it = (*mCurrentChain).mDescriptors.erase(it);
  340. mRebuildPollset = true;
  341. }
  342. else
  343. {
  344. ++it;
  345. }
  346. }
  348. if(!poll)
  349. {
  350. mRebuildPollset = true;
  351. return true;
  352. }
  353. LLChainInfo::pipe_conditional_t value;
  354. value.first = pipe_ptr;
  355. value.second = *poll;
  356. value.second.rtnevents = 0;
  357. if(!poll->p)
  358. {
  359. // each fd needs a pool to work with, so if one was
  360. // not specified, use this pool.
  361. // *FIX: Should it always be this pool?
  362. value.second.p = mPool;
  363. }
  364. value.second.client_data = new S32(++mPollsetClientID);
  365. (*mCurrentChain).mDescriptors.push_back(value);
  366. mRebuildPollset = true;
  367. return true;
  368. }
  370. S32 LLPumpIO::setLock()
  371. {
  372. // *NOTE: I do not think it is necessary to acquire a mutex here
  373. // since this should only be called during the pump(), and should
  374. // only change the running chain. Any other use of this method is
  375. // incorrect usage. If it becomes necessary to acquire a lock
  376. // here, be sure to lock here and call a protected method to get
  377. // the lock, and sleepChain() should probably acquire the same
  378. // lock while and calling the same protected implementation to
  379. // lock the runner at the same time.
  381. // If no chain is running, return failure.
  382. if(mRunningChains.end() == mCurrentChain)
  383. {
  384. return 0;
  385. }
  387. // deal with wrap.
  388. if(++mNextLock <= 0)
  389. {
  390. mNextLock = 1;
  391. }
  393. // set the lock
  394. (*mCurrentChain).mLock = mNextLock;
  395. return mNextLock;
  396. }
  398. void LLPumpIO::clearLock(S32 key)
  399. {
  400. // We need to lock it here since we do not want to be iterating
  401. // over the chains twice. We can safely call process() while this
  402. // is happening since we should not be erasing a locked pipe, and
  403. // therefore won't be treading into deleted memory. I think we can
  404. // also clear the lock on the chain safely since the pump only
  405. // reads that value.
  406. #if LL_THREADS_APR
  407. LLScopedLock lock(mChainsMutex);
  408. #endif
  409. mClearLocks.insert(key);
  410. }
  412. bool LLPumpIO::sleepChain(F64 seconds)
  413. {
  414. // Much like the call to setLock(), this should only be called
  415. // from one chain during processing, so there is no need to
  416. // acquire a mutex.
  417. if(seconds <= 0.0) return false;
  418. S32 key = setLock();
  419. if(!key) return false;
  420. LLRunner::run_handle_t handle = mRunner.addRunnable(
  421. LLChainSleeper::build(this, key),
  422. LLRunner::RUN_IN,
  423. seconds);
  424. if(0 == handle) return false;
  425. return true;
  426. }
  428. bool LLPumpIO::copyCurrentLinkInfo(links_t& links) const
  429. {
  430. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  431. if(mRunningChains.end() == mCurrentChain)
  432. {
  433. return false;
  434. }
  435. std::copy(
  436. (*mCurrentChain).mChainLinks.begin(),
  437. (*mCurrentChain).mChainLinks.end(),
  438. std::back_insert_iterator<links_t>(links));
  439. return true;
  440. }
  442. void LLPumpIO::pump()
  443. {
  444. pump(DEFAULT_POLL_TIMEOUT);
  445. }
  447. static LLFastTimer::DeclareTimer FTM_PUMP("Pump");
  449. //timeout is in microseconds
  450. void LLPumpIO::pump(const S32& poll_timeout)
  451. {
  452. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  453. LLFastTimer t1(FTM_PUMP);
  454. //llinfos << "LLPumpIO::pump()" << llendl;
  456. // Run any pending runners.
  457. mRunner.run();
  459. // We need to move all of the pending heads over to the running
  460. // chains.
  461. PUMP_DEBUG;
  462. if(true)
  463. {
  464. #if LL_THREADS_APR
  465. LLScopedLock lock(mChainsMutex);
  466. #endif
  467. // bail if this pump is paused.
  468. if(PAUSING == mState)
  469. {
  470. mState = PAUSED;
  471. }
  472. if(PAUSED == mState)
  473. {
  474. return;
  475. }
  477. PUMP_DEBUG;
  478. // Move the pending chains over to the running chaings
  479. if(!mPendingChains.empty())
  480. {
  481. PUMP_DEBUG;
  482. //lldebugs << "Pushing " << mPendingChains.size() << "." << llendl;
  483. std::copy(
  484. mPendingChains.begin(),
  485. mPendingChains.end(),
  486. std::back_insert_iterator<running_chains_t>(mRunningChains));
  487. mPendingChains.clear();
  488. PUMP_DEBUG;
  489. }
  491. // Clear any locks. This needs to be done here so that we do
  492. // not clash during a call to clearLock().
  493. if(!mClearLocks.empty())
  494. {
  495. PUMP_DEBUG;
  496. running_chains_t::iterator it = mRunningChains.begin();
  497. running_chains_t::iterator end = mRunningChains.end();
  498. std::set<S32>::iterator not_cleared = mClearLocks.end();
  499. for(; it != end; ++it)
  500. {
  501. if((*it).mLock && mClearLocks.find((*it).mLock) != not_cleared)
  502. {
  503. (*it).mLock = 0;
  504. }
  505. }
  506. PUMP_DEBUG;
  507. mClearLocks.clear();
  508. }
  509. }
  511. PUMP_DEBUG;
  512. // rebuild the pollset if necessary
  513. if(mRebuildPollset)
  514. {
  515. PUMP_DEBUG;
  516. rebuildPollset();
  517. mRebuildPollset = false;
  518. }
  520. // Poll based on the last known pollset
  521. // *TODO: may want to pass in a poll timeout so it works correctly
  522. // in single and multi threaded processes.
  523. PUMP_DEBUG;
  524. typedef std::map<S32, S32> signal_client_t;
  525. signal_client_t signalled_client;
  526. const apr_pollfd_t* poll_fd = NULL;
  527. if(mPollset)
  528. {
  529. PUMP_DEBUG;
  530. //llinfos << "polling" << llendl;
  531. S32 count = 0;
  532. S32 client_id = 0;
  533.         {
  534.             LLPerfBlock polltime("pump_poll");
  535.             apr_pollset_poll(mPollset, poll_timeout, &count, &poll_fd);
  536.         }
  537. PUMP_DEBUG;
  538. for(S32 ii = 0; ii < count; ++ii)
  539. {
  540. ll_debug_poll_fd("Signalled pipe", &poll_fd[ii]);
  541. client_id = *((S32*)poll_fd[ii].client_data);
  542. signalled_client[client_id] = ii;
  543. }
  544. PUMP_DEBUG;
  545. }
  547. PUMP_DEBUG;
  548. // set up for a check to see if each one was signalled
  549. signal_client_t::iterator not_signalled = signalled_client.end();
  551. // Process everything as appropriate
  552. //lldebugs << "Running chain count: " << mRunningChains.size() << llendl;
  553. running_chains_t::iterator run_chain = mRunningChains.begin();
  554. bool process_this_chain = false;
  555. while( run_chain != mRunningChains.end() )
  556. {
  557. PUMP_DEBUG;
  558. if((*run_chain).mInit
  559.    && (*run_chain).mTimer.getStarted()
  560.    && (*run_chain).mTimer.hasExpired())
  561. {
  562. PUMP_DEBUG;
  563. if(handleChainError(*run_chain, LLIOPipe::STATUS_EXPIRED))
  564. {
  565. // the pipe probably handled the error. If the handler
  566. // forgot to reset the expiration then we need to do
  567. // that here.
  568. if((*run_chain).mTimer.getStarted()
  569.    && (*run_chain).mTimer.hasExpired())
  570. {
  571. PUMP_DEBUG;
  572. llinfos << "Error handler forgot to reset timeout. "
  573. << "Resetting to " << DEFAULT_CHAIN_EXPIRY_SECS
  574. << " seconds." << llendl;
  575. (*run_chain).setTimeoutSeconds(DEFAULT_CHAIN_EXPIRY_SECS);
  576. }
  577. }
  578. else
  579. {
  580. PUMP_DEBUG;
  581. // it timed out and no one handled it, so we need to
  582. // retire the chain
  583. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  584. lldebugs << "Removing chain "
  585. << (*run_chain).mChainLinks[0].mPipe
  586. << " '"
  587. << typeid(*((*run_chain).mChainLinks[0].mPipe)).name()
  588. << "' because it timed out." << llendl;
  589. #else
  590. // lldebugs << "Removing chain "
  591. // << (*run_chain).mChainLinks[0].mPipe
  592. // << " because we reached the end." << llendl;
  593. #endif
  594. run_chain = mRunningChains.erase(run_chain);
  595. continue;
  596. }
  597. }
  598. PUMP_DEBUG;
  599. if((*run_chain).mLock)
  600. {
  601. ++run_chain;
  602. continue;
  603. }
  604. PUMP_DEBUG;
  605. mCurrentChain = run_chain;
  607. if((*run_chain).mDescriptors.empty())
  608. {
  609. // if there are no conditionals, just process this chain.
  610. process_this_chain = true;
  611. //lldebugs << "no conditionals - processing" << llendl;
  612. }
  613. else
  614. {
  615. PUMP_DEBUG;
  616. //lldebugs << "checking conditionals" << llendl;
  617. // Check if this run chain was signalled. If any file
  618. // descriptor is ready for something, then go ahead and
  619. // process this chian.
  620. process_this_chain = false;
  621. if(!signalled_client.empty())
  622. {
  623. PUMP_DEBUG;
  624. LLChainInfo::conditionals_t::iterator it;
  625. it = (*run_chain).mDescriptors.begin();
  626. LLChainInfo::conditionals_t::iterator end;
  627. end = (*run_chain).mDescriptors.end();
  628. S32 client_id = 0;
  629. signal_client_t::iterator signal;
  630. for(; it != end; ++it)
  631. {
  632. PUMP_DEBUG;
  633. client_id = *((S32*)((*it).second.client_data));
  634. signal = signalled_client.find(client_id);
  635. if (signal == not_signalled) continue;
  636. static const apr_int16_t POLL_CHAIN_ERROR =
  637. APR_POLLHUP | APR_POLLNVAL | APR_POLLERR;
  638. const apr_pollfd_t* poll = &(poll_fd[(*signal).second]);
  639. if(poll->rtnevents & POLL_CHAIN_ERROR)
  640. {
  641. // Potential eror condition has been
  642. // returned. If HUP was one of them, we pass
  643. // that as the error even though there may be
  644. // more. If there are in fact more errors,
  645. // we'll just wait for that detection until
  646. // the next pump() cycle to catch it so that
  647. // the logic here gets no more strained than
  648. // it already is.
  649. LLIOPipe::EStatus error_status;
  650. if(poll->rtnevents & APR_POLLHUP)
  651. error_status = LLIOPipe::STATUS_LOST_CONNECTION;
  652. else
  653. error_status = LLIOPipe::STATUS_ERROR;
  654. if(handleChainError(*run_chain, error_status)) break;
  655. ll_debug_poll_fd("Removing pipe", poll);
  656. llwarns << "Removing pipe "
  657. << (*run_chain).mChainLinks[0].mPipe
  658. << " '"
  659. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  660. << typeid(
  661. *((*run_chain).mChainLinks[0].mPipe)).name()
  662. #endif
  663. << "' because: "
  664. << events_2_string(poll->rtnevents)
  665. << llendl;
  666. (*run_chain).mHead = (*run_chain).mChainLinks.end();
  667. break;
  668. }
  670. // at least 1 fd got signalled, and there were no
  671. // errors. That means we process this chain.
  672. process_this_chain = true;
  673. break;
  674. }
  675. }
  676. }
  677. if(process_this_chain)
  678. {
  679. PUMP_DEBUG;
  680. if(!((*run_chain).mInit))
  681. {
  682. (*run_chain).mHead = (*run_chain).mChainLinks.begin();
  683. (*run_chain).mInit = true;
  684. }
  685. PUMP_DEBUG;
  686. processChain(*run_chain);
  687. }
  689. PUMP_DEBUG;
  690. if((*run_chain).mHead == (*run_chain).mChainLinks.end())
  691. {
  692. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  693. lldebugs << "Removing chain " << (*run_chain).mChainLinks[0].mPipe
  694. << " '"
  695. << typeid(*((*run_chain).mChainLinks[0].mPipe)).name()
  696. << "' because we reached the end." << llendl;
  697. #else
  698. // lldebugs << "Removing chain " << (*run_chain).mChainLinks[0].mPipe
  699. // << " because we reached the end." << llendl;
  700. #endif
  702. PUMP_DEBUG;
  703. // This chain is done. Clean up any allocated memory and
  704. // erase the chain info.
  705. std::for_each(
  706. (*run_chain).mDescriptors.begin(),
  707. (*run_chain).mDescriptors.end(),
  708. ll_delete_apr_pollset_fd_client_data());
  709. run_chain = mRunningChains.erase(run_chain);
  711. // *NOTE: may not always need to rebuild the pollset.
  712. mRebuildPollset = true;
  713. }
  714. else
  715. {
  716. PUMP_DEBUG;
  717. // this chain needs more processing - just go to the next
  718. // chain.
  719. ++run_chain;
  720. }
  721. }
  723. PUMP_DEBUG;
  724. // null out the chain
  725. mCurrentChain = mRunningChains.end();
  726. END_PUMP_DEBUG;
  727. }
  729. //bool LLPumpIO::respond(const chain_t& pipes)
  730. //{
  731. //#if LL_THREADS_APR
  732. // LLScopedLock lock(mCallbackMutex);
  733. //#endif
  734. // LLChainInfo info;
  735. // links_t links;
  736. //
  737. // mPendingCallbacks.push_back(info);
  738. // return true;
  739. //}
  741. bool LLPumpIO::respond(LLIOPipe* pipe)
  742. {
  743. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  744. if(NULL == pipe) return false;
  746. #if LL_THREADS_APR
  747. LLScopedLock lock(mCallbackMutex);
  748. #endif
  749. LLChainInfo info;
  750. LLLinkInfo link;
  751. link.mPipe = pipe;
  752. info.mChainLinks.push_back(link);
  753. mPendingCallbacks.push_back(info);
  754. return true;
  755. }
  757. bool LLPumpIO::respond(
  758. const links_t& links,
  759. LLIOPipe::buffer_ptr_t data,
  760. LLSD context)
  761. {
  762. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  763. // if the caller is providing a full link description, we need to
  764. // have that description matched to a particular buffer.
  765. if(!data) return false;
  766. if(links.empty()) return false;
  768. #if LL_THREADS_APR
  769. LLScopedLock lock(mCallbackMutex);
  770. #endif
  772. // Add the callback response
  773. LLChainInfo info;
  774. info.mChainLinks = links;
  775. info.mData = data;
  776. info.mContext = context;
  777. mPendingCallbacks.push_back(info);
  778. return true;
  779. }
  781. void LLPumpIO::callback()
  782. {
  783. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  784. //llinfos << "LLPumpIO::callback()" << llendl;
  785. if(true)
  786. {
  787. #if LL_THREADS_APR
  788. LLScopedLock lock(mCallbackMutex);
  789. #endif
  790. std::copy(
  791. mPendingCallbacks.begin(),
  792. mPendingCallbacks.end(),
  793. std::back_insert_iterator<callbacks_t>(mCallbacks));
  794. mPendingCallbacks.clear();
  795. }
  796. if(!mCallbacks.empty())
  797. {
  798. callbacks_t::iterator it = mCallbacks.begin();
  799. callbacks_t::iterator end = mCallbacks.end();
  800. for(; it != end; ++it)
  801. {
  802. // it's always the first and last time for respone chains
  803. (*it).mHead = (*it).mChainLinks.begin();
  804. (*it).mInit = true;
  805. (*it).mEOS = true;
  806. processChain(*it);
  807. }
  808. mCallbacks.clear();
  809. }
  810. }
  812. void LLPumpIO::control(LLPumpIO::EControl op)
  813. {
  814. #if LL_THREADS_APR
  815. LLScopedLock lock(mChainsMutex);
  816. #endif
  817. switch(op)
  818. {
  819. case PAUSE:
  820. mState = PAUSING;
  821. break;
  822. case RESUME:
  823. mState = NORMAL;
  824. break;
  825. default:
  826. // no-op
  827. break;
  828. }
  829. }
  831. void LLPumpIO::initialize(apr_pool_t* pool)
  832. {
  833. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  834. if(!pool) return;
  835. #if LL_THREADS_APR
  836. // SJB: Windows defaults to NESTED and OSX defaults to UNNESTED, so use UNNESTED explicitly.
  837. apr_thread_mutex_create(&mChainsMutex, APR_THREAD_MUTEX_UNNESTED, pool);
  838. apr_thread_mutex_create(&mCallbackMutex, APR_THREAD_MUTEX_UNNESTED, pool);
  839. #endif
  840. mPool = pool;
  841. }
  843. void LLPumpIO::cleanup()
  844. {
  845. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  846. #if LL_THREADS_APR
  847. if(mChainsMutex) apr_thread_mutex_destroy(mChainsMutex);
  848. if(mCallbackMutex) apr_thread_mutex_destroy(mCallbackMutex);
  849. #endif
  850. mChainsMutex = NULL;
  851. mCallbackMutex = NULL;
  852. if(mPollset)
  853. {
  854. // lldebugs << "cleaning up pollset" << llendl;
  855. apr_pollset_destroy(mPollset);
  856. mPollset = NULL;
  857. }
  858. if(mCurrentPool)
  859. {
  860. apr_pool_destroy(mCurrentPool);
  861. mCurrentPool = NULL;
  862. }
  863. mPool = NULL;
  864. }
  866. void LLPumpIO::rebuildPollset()
  867. {
  868. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  869. // lldebugs << "LLPumpIO::rebuildPollset()" << llendl;
  870. if(mPollset)
  871. {
  872. //lldebugs << "destroying pollset" << llendl;
  873. apr_pollset_destroy(mPollset);
  874. mPollset = NULL;
  875. }
  876. U32 size = 0;
  877. running_chains_t::iterator run_it = mRunningChains.begin();
  878. running_chains_t::iterator run_end = mRunningChains.end();
  879. for(; run_it != run_end; ++run_it)
  880. {
  881. size += (*run_it).mDescriptors.size();
  882. }
  883. //lldebugs << "found " << size << " descriptors." << llendl;
  884. if(size)
  885. {
  886. // Recycle the memory pool
  887. const S32 POLLSET_POOL_RECYCLE_COUNT = 100;
  888. if(mCurrentPool
  889.    && (0 == (++mCurrentPoolReallocCount % POLLSET_POOL_RECYCLE_COUNT)))
  890. {
  891. apr_pool_destroy(mCurrentPool);
  892. mCurrentPool = NULL;
  893. mCurrentPoolReallocCount = 0;
  894. }
  895. if(!mCurrentPool)
  896. {
  897. apr_status_t status = apr_pool_create(&mCurrentPool, mPool);
  898. (void)ll_apr_warn_status(status);
  899. }
  901. // add all of the file descriptors
  902. run_it = mRunningChains.begin();
  903. LLChainInfo::conditionals_t::iterator fd_it;
  904. LLChainInfo::conditionals_t::iterator fd_end;
  905. apr_pollset_create(&mPollset, size, mCurrentPool, 0);
  906. for(; run_it != run_end; ++run_it)
  907. {
  908. fd_it = (*run_it).mDescriptors.begin();
  909. fd_end = (*run_it).mDescriptors.end();
  910. for(; fd_it != fd_end; ++fd_it)
  911. {
  912. apr_pollset_add(mPollset, &((*fd_it).second));
  913. }
  914. }
  915. }
  916. }
  918. void LLPumpIO::processChain(LLChainInfo& chain)
  919. {
  920. PUMP_DEBUG;
  921. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  922. LLIOPipe::EStatus status = LLIOPipe::STATUS_OK;
  923. links_t::iterator it = chain.mHead;
  924. links_t::iterator end = chain.mChainLinks.end();
  925. bool need_process_signaled = false;
  926. bool keep_going = true;
  927. do
  928. {
  929. #if LL_DEBUG_PROCESS_LINK
  930. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  931. llinfos << "Processing " << typeid(*((*it).mPipe)).name() << "."
  932. << llendl;
  933. #else
  934. llinfos << "Processing link " << (*it).mPipe << "." << llendl;
  935. #endif
  936. #endif
  937. #if LL_DEBUG_SPEW_BUFFER_CHANNEL_IN
  938. if(chain.mData)
  939. {
  940. char* buf = NULL;
  941. S32 bytes = chain.mData->countAfter((*it).mChannels.in(), NULL);
  942. if(bytes)
  943. {
  944. buf = new char[bytes + 1];
  945. chain.mData->readAfter(
  946. (*it).mChannels.in(),
  947. NULL,
  948. (U8*)buf,
  949. bytes);
  950. buf[bytes] = '';
  951. llinfos << "CHANNEL IN(" << (*it).mChannels.in() << "): "
  952. << buf << llendl;
  953. delete[] buf;
  954. buf = NULL;
  955. }
  956. else
  957. {
  958. llinfos << "CHANNEL IN(" << (*it).mChannels.in()<< "): (null)"
  959. << llendl;
  960. }
  961. }
  962. #endif
  963. PUMP_DEBUG;
  964. status = (*it).mPipe->process(
  965. (*it).mChannels,
  966. chain.mData,
  967. chain.mEOS,
  968. chain.mContext,
  969. this);
  970. #if LL_DEBUG_SPEW_BUFFER_CHANNEL_OUT
  971. if(chain.mData)
  972. {
  973. char* buf = NULL;
  974. S32 bytes = chain.mData->countAfter((*it).mChannels.out(), NULL);
  975. if(bytes)
  976. {
  977. buf = new char[bytes + 1];
  978. chain.mData->readAfter(
  979. (*it).mChannels.out(),
  980. NULL,
  981. (U8*)buf,
  982. bytes);
  983. buf[bytes] = '';
  984. llinfos << "CHANNEL OUT(" << (*it).mChannels.out()<< "): "
  985. << buf << llendl;
  986. delete[] buf;
  987. buf = NULL;
  988. }
  989. else
  990. {
  991. llinfos << "CHANNEL OUT(" << (*it).mChannels.out()<< "): (null)"
  992. << llendl;
  993. }
  994. }
  995. #endif
  997. #if LL_DEBUG_PROCESS_RETURN_VALUE
  998. // Only bother with the success codes - error codes are logged
  999. // below.
  1000. if(LLIOPipe::isSuccess(status))
  1001. {
  1002. llinfos << "Pipe returned: '"
  1003. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  1004. << typeid(*((*it).mPipe)).name() << "':'"
  1005. #endif
  1006. << LLIOPipe::lookupStatusString(status) << "'" << llendl;
  1007. }
  1008. #endif
  1010. PUMP_DEBUG;
  1011. switch(status)
  1012. {
  1013. case LLIOPipe::STATUS_OK:
  1014. // no-op
  1015. break;
  1016. case LLIOPipe::STATUS_STOP:
  1017. PUMP_DEBUG;
  1018. status = LLIOPipe::STATUS_OK;
  1019. chain.mHead = end;
  1020. keep_going = false;
  1021. break;
  1022. case LLIOPipe::STATUS_DONE:
  1023. PUMP_DEBUG;
  1024. status = LLIOPipe::STATUS_OK;
  1025. chain.mHead = (it + 1);
  1026. chain.mEOS = true;
  1027. break;
  1028. case LLIOPipe::STATUS_BREAK:
  1029. PUMP_DEBUG;
  1030. status = LLIOPipe::STATUS_OK;
  1031. keep_going = false;
  1032. break;
  1033. case LLIOPipe::STATUS_NEED_PROCESS:
  1034. PUMP_DEBUG;
  1035. status = LLIOPipe::STATUS_OK;
  1036. if(!need_process_signaled)
  1037. {
  1038. need_process_signaled = true;
  1039. chain.mHead = it;
  1040. }
  1041. break;
  1042. default:
  1043. PUMP_DEBUG;
  1044. if(LLIOPipe::isError(status))
  1045. {
  1046. llinfos << "Pump generated pipe err: '"
  1047. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  1048. << typeid(*((*it).mPipe)).name() << "':'"
  1049. #endif
  1050. << LLIOPipe::lookupStatusString(status)
  1051. << "'" << llendl;
  1052. #if LL_DEBUG_SPEW_BUFFER_CHANNEL_IN_ON_ERROR
  1053. if(chain.mData)
  1054. {
  1055. char* buf = NULL;
  1056. S32 bytes = chain.mData->countAfter(
  1057. (*it).mChannels.in(),
  1058. NULL);
  1059. if(bytes)
  1060. {
  1061. buf = new char[bytes + 1];
  1062. chain.mData->readAfter(
  1063. (*it).mChannels.in(),
  1064. NULL,
  1065. (U8*)buf,
  1066. bytes);
  1067. buf[bytes] = '';
  1068. llinfos << "Input After Error: " << buf << llendl;
  1069. delete[] buf;
  1070. buf = NULL;
  1071. }
  1072. else
  1073. {
  1074. llinfos << "Input After Error: (null)" << llendl;
  1075. }
  1076. }
  1077. else
  1078. {
  1079. llinfos << "Input After Error: (null)" << llendl;
  1080. }
  1081. #endif
  1082. keep_going = false;
  1083. chain.mHead  = it;
  1084. if(!handleChainError(chain, status))
  1085. {
  1086. chain.mHead = end;
  1087. }
  1088. }
  1089. else
  1090. {
  1091. llinfos << "Unhandled status code: " << status << ":"
  1092. << LLIOPipe::lookupStatusString(status) << llendl;
  1093. }
  1094. break;
  1095. }
  1096. PUMP_DEBUG;
  1097. } while(keep_going && (++it != end));
  1098. PUMP_DEBUG;
  1099. }
  1101. bool LLPumpIO::handleChainError(
  1102. LLChainInfo& chain,
  1103. LLIOPipe::EStatus error)
  1104. {
  1105. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  1106. links_t::reverse_iterator rit;
  1107. if(chain.mHead == chain.mChainLinks.end())
  1108. {
  1109. rit = links_t::reverse_iterator(chain.mHead);
  1110. }
  1111. else
  1112. {
  1113. rit = links_t::reverse_iterator(chain.mHead + 1);
  1114. }
  1116. links_t::reverse_iterator rend = chain.mChainLinks.rend();
  1117. bool handled = false;
  1118. bool keep_going = true;
  1119. do
  1120. {
  1121. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  1122. lldebugs << "Passing error to " << typeid(*((*rit).mPipe)).name()
  1123.  << "." << llendl;
  1124. #endif
  1125. error = (*rit).mPipe->handleError(error, this);
  1126. switch(error)
  1127. {
  1128. case LLIOPipe::STATUS_OK:
  1129. handled = true;
  1130. chain.mHead = rit.base();
  1131. break;
  1132. case LLIOPipe::STATUS_STOP:
  1133. case LLIOPipe::STATUS_DONE:
  1134. case LLIOPipe::STATUS_BREAK:
  1135. case LLIOPipe::STATUS_NEED_PROCESS:
  1136. #if LL_DEBUG_PIPE_TYPE_IN_PUMP
  1137. lldebugs << "Pipe " << typeid(*((*rit).mPipe)).name()
  1138.  << " returned code to stop error handler." << llendl;
  1139. #endif
  1140. keep_going = false;
  1141. break;
  1142. default:
  1143. if(LLIOPipe::isSuccess(error))
  1144. {
  1145. llinfos << "Unhandled status code: " << error << ":"
  1146. << LLIOPipe::lookupStatusString(error) << llendl;
  1147. error = LLIOPipe::STATUS_ERROR;
  1148. keep_going = false;
  1149. }
  1150. break;
  1151. }
  1152. } while(keep_going && !handled && (++rit != rend));
  1153. return handled;
  1154. }
  1156. /**
  1157.  * LLPumpIO::LLChainInfo
  1158.  */
  1160. LLPumpIO::LLChainInfo::LLChainInfo() :
  1161. mInit(false),
  1162. mLock(0),
  1163. mEOS(false)
  1164. {
  1165. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  1166. mTimer.setTimerExpirySec(DEFAULT_CHAIN_EXPIRY_SECS);
  1167. }
  1169. void LLPumpIO::LLChainInfo::setTimeoutSeconds(F32 timeout)
  1170. {
  1171. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  1172. if(timeout > 0.0f)
  1173. {
  1174. mTimer.start();
  1175. mTimer.reset();
  1176. mTimer.setTimerExpirySec(timeout);
  1177. }
  1178. else
  1179. {
  1180. mTimer.stop();
  1181. }
  1182. }
  1184. void LLPumpIO::LLChainInfo::adjustTimeoutSeconds(F32 delta)
  1185. {
  1186. LLMemType m1(LLMemType::MTYPE_IO_PUMP);
  1187. if(mTimer.getStarted())
  1188. {
  1189. F64 expiry = mTimer.expiresAt();
  1190. expiry += delta;
  1191. mTimer.setExpiryAt(expiry);
  1192. }
  1193. }