/** * @file llpumpio.h * @author Phoenix * @date 2004-11-19 * @brief Declaration of pump class which manages io chains. * * $LicenseInfo:firstyear=2004&license=viewergpl$ * * Copyright (c) 2004-2009, Linden Research, Inc. * * Second Life Viewer Source Code * The source code in this file ("Source Code") is provided by Linden Lab * to you under the terms of the GNU General Public License, version 2.0 * ("GPL"), unless you have obtained a separate licensing agreement * ("Other License"), formally executed by you and Linden Lab. Terms of * the GPL can be found in doc/GPL-license.txt in this distribution, or * online at http://secondlifegrid.net/programs/open_source/licensing/gplv2 * * There are special exceptions to the terms and conditions of the GPL as * it is applied to this Source Code. View the full text of the exception * in the file doc/FLOSS-exception.txt in this software distribution, or * online at * http://secondlifegrid.net/programs/open_source/licensing/flossexception * * By copying, modifying or distributing this software, you acknowledge * that you have read and understood your obligations described above, * and agree to abide by those obligations. * * ALL LINDEN LAB SOURCE CODE IS PROVIDED "AS IS." LINDEN LAB MAKES NO * WARRANTIES, EXPRESS, IMPLIED OR OTHERWISE, REGARDING ITS ACCURACY, * COMPLETENESS OR PERFORMANCE. * $/LicenseInfo$ */ #ifndef LL_LLPUMPIO_H #define LL_LLPUMPIO_H #include #include #if LL_LINUX // needed for PATH_MAX in APR. #include #endif #include "aiaprpool.h" #include "llbuffer.h" #include "llframetimer.h" #include "lliopipe.h" #include "llrun.h" // Define this to enable use with the APR thread library. //#define LL_THREADS_APR 1 // some simple constants to help with timeouts extern const F32 DEFAULT_CHAIN_EXPIRY_SECS; extern const F32 SHORT_CHAIN_EXPIRY_SECS; extern const F32 NEVER_CHAIN_EXPIRY_SECS; /** * @class LLPumpIO * @brief Class to manage sets of io chains. * * The pump class provides a thread abstraction for doing IO based * communication between two threads in a structured and optimized for * processor time. The primary usage is to create a pump, and call * pump() on a thread used for IO and call * respond() on a thread that is expected to do higher * level processing. You can call almost any other method from any * thread - see notes for each method for details. * * A pump instance manages much of the state for the pipe, including * the list of pipes in the chain, the channel for each element in the * chain, the buffer, and if any pipe has marked the stream or process * as done. Pipes can also set file descriptor based conditional * statements so that calls to process do not happen until data is * ready to be read or written. Pipes control execution of calls to * process by returning a status code such as STATUS_OK or * STATUS_BREAK. * One way to conceptualize the way IO will work is that a pump * combines the unit processing of pipes to behave like file pipes on * the unix command line. */ class LLPumpIO { public: /** * @brief Constructor. */ LLPumpIO(void); /** * @brief Destructor. */ ~LLPumpIO(); /** * @brief Typedef for having a chain of pipes. */ typedef std::vector chain_t; /** * @brief Add a chain to this pump and process in the next cycle. * * This method will automatically generate a buffer and assign * each link in the chain as if it were the consumer to the * previous. * @param chain The pipes for the chain * @param timeout The number of seconds in the future to * expire. Pass in 0.0f to never expire. * @return Returns true if anything was added to the pump. */ bool addChain(const chain_t& chain, F32 timeout); /** * @brief Struct to associate a pipe with it's buffer io indexes. */ struct LLLinkInfo { LLIOPipe::ptr_t mPipe; LLChannelDescriptors mChannels; }; /** * @brief Typedef for having a chain of LLLinkInfo * instances. */ typedef std::vector links_t; /** * @brief Add a chain to this pump and process in the next cycle. * * This method provides a slightly more sophisticated method for * adding a chain where the caller can specify which link elements * are on what channels. This method will fail if no buffer is * provided since any calls to generate new channels for the * buffers will cause unpredictable interleaving of data. * @param links The pipes and io indexes for the chain * @param data Shared pointer to data buffer * @param context Potentially undefined context meta-data for chain. * @param timeout The number of seconds in the future to * expire. Pass in 0.0f to never expire. * @return Returns true if anything was added to the pump. */ bool addChain( const links_t& links, LLIOPipe::buffer_ptr_t data, LLSD context, F32 timeout); /** * @brief Set or clear a timeout for the running chain * * @param timeout The number of seconds in the future to * expire. Pass in 0.0f to never expire. * @return Returns true if the timer was set. */ bool setTimeoutSeconds(F32 timeout); /** * @brief Adjust the timeout of the running chain. * * This method has no effect if there is no timeout on the chain. * @param delta The number of seconds to add to/remove from the timeout. */ void adjustTimeoutSeconds(F32 delta); /** * @brief Set up file descriptors for for the running chain. * @see rebuildPollset() * * There is currently a limit of one conditional per pipe. * *NOTE: The internal mechanism for building a pollset based on * pipe/pollfd/chain generates an epoll error on linux (and * probably behaves similarly on other platforms) because the * pollset rebuilder will add each apr_pollfd_t serially. This * does not matter for pipes on the same chain, since any * signalled pipe will eventually invoke a call to process(), but * is a problem if the same apr_pollfd_t is on different * chains. Once we have more than just network i/o on the pump, * this might matter. * *FIX: Given the structure of the pump and pipe relationship, * this should probably go through a different mechanism than the * pump. I think it would be best if the pipe had some kind of * controller which was passed into process() rather * than the pump which exposed this interface. * @param pipe The pipe which is setting a conditional * @param poll The entire socket and read/write condition - null to remove * @return Returns true if the poll state was set. */ bool setConditional(LLIOPipe* pipe, const apr_pollfd_t* poll); /** * @brief Lock the current chain. * @see sleepChain() since it relies on the implementation of this method. * * This locks the currently running chain so that no more calls to * process() until you call clearLock() * with the lock identifier. * *FIX: Given the structure of the pump and pipe relationship, * this should probably go through a different mechanism than the * pump. I think it would be best if the pipe had some kind of * controller which was passed into process() rather * than the pump which exposed this interface. * @return Returns the lock identifer to be used in * clearLock() or 0 on failure. */ S32 setLock(); /** * @brief Clears the identified lock. * * @param links A container for the links which will be appended */ void clearLock(S32 key); /** * @brief Stop processing a chain for a while. * @see setLock() * * This method will not update the timeout for this * chain, so it is possible to sleep the chain until it is * collected by the pump during a timeout cleanup. * @param seconds The number of seconds in the future to * resume processing. * @return Returns true if the */ bool sleepChain(F64 seconds); /** * @brief Copy the currently running chain link info * * *FIX: Given the structure of the pump and pipe relationship, * this should probably go through a different mechanism than the * pump. I think it would be best if the pipe had some kind of * controller which was passed into process() rather * than the pump which exposed this interface. * @param links A container for the links which will be appended * @return Returns true if the currently running chain was copied. */ bool copyCurrentLinkInfo(links_t& links) const; /** * @brief Call this method to call process on all running chains. * * This method iterates through the running chains, and if all * pipe on a chain are unconditionally ready or if any pipe has * any conditional processiong condition then process will be * called on every chain which has requested processing. that * chain has a file descriptor ready, process() will * be called for all pipes which have requested it. */ void pump(const S32& poll_timeout); void pump(); /** * @brief Add a chain to a special queue which will be called * during the next call to callback() and then * dropped from the queue. * * @param chain The IO chain that will get one process(). */ //void respond(const chain_t& pipes); /** * @brief Add pipe to a special queue which will be called * during the next call to callback() and then dropped * from the queue. * * This call will add a single pipe, with no buffer, context, or * channel information to the callback queue. It will be called * once, and then dropped. * @param pipe A single io pipe which will be called * @return Returns true if anything was added to the pump. */ bool respond(LLIOPipe* pipe); /** * @brief Add a chain to a special queue which will be called * during the next call to callback() and then * dropped from the queue. * * It is important to remember that you should not add a data * buffer or context which may still be in another chain - that * will almost certainly lead to a problems. Ensure that you are * done reading and writing to those parameters, have new * generated, or empty pointers. * @param links The pipes and io indexes for the chain * @param data Shared pointer to data buffer * @param context Potentially undefined context meta-data for chain. * @return Returns true if anything was added to the pump. */ bool respond( const links_t& links, LLIOPipe::buffer_ptr_t data, LLSD context); /** * @brief Run through the callback queue and call process(). * * This call will process all prending responses and call process * on each. This method will then drop all processed callback * requests which may lead to deleting the referenced objects. */ void callback(); /** * @brief Enumeration to send commands to the pump. */ enum EControl { PAUSE, RESUME, }; /** * @brief Send a command to the pump. * * @param op What control to send to the pump. */ void control(EControl op); protected: /** * @brief State of the pump */ enum EState { NORMAL, PAUSING, PAUSED }; // instance data EState mState; bool mRebuildPollset; apr_pollset_t* mPollset; S32 mPollsetClientID; S32 mNextLock; std::set mClearLocks; // This is the pump's runnable scheduler used for handling // expiring locks. LLRunner mRunner; // This structure is the stuff we track while running chains. struct LLChainInfo { // methods LLChainInfo(); void setTimeoutSeconds(F32 timeout); void adjustTimeoutSeconds(F32 delta); // basic member data bool mInit; S32 mLock; LLFrameTimer mTimer; links_t::iterator mHead; links_t mChainLinks; LLIOPipe::buffer_ptr_t mData; bool mEOS; LLSD mContext; // tracking inside the pump typedef std::pair pipe_conditional_t; typedef std::vector conditionals_t; conditionals_t mDescriptors; boost::shared_ptr mDescriptorsPool; }; // All the running chains & info typedef std::vector pending_chains_t; pending_chains_t mPendingChains; typedef std::list running_chains_t; running_chains_t mRunningChains; typedef running_chains_t::iterator current_chain_t; current_chain_t mCurrentChain; // structures necessary for doing callbacks // since the callbacks only get one chance to run, we do not have // to maintain a list. typedef std::vector callbacks_t; callbacks_t mPendingCallbacks; callbacks_t mCallbacks; // Memory pool for pollsets & mutexes. AIAPRPool mPool; AIAPRPool mCurrentPool; S32 mCurrentPoolReallocCount; #if LL_THREADS_APR apr_thread_mutex_t* mChainsMutex; apr_thread_mutex_t* mCallbackMutex; #else int* mChainsMutex; int* mCallbackMutex; #endif protected: void initialize(); /** * @brief Given the internal state of the chains, rebuild the pollset * @see setConditional() */ void rebuildPollset(); /** * @brief Process the chain passed in. * * This method will potentially modify the internals of the * chain. On end, the chain.mHead will equal * chain.mChainLinks.end(). * @param chain The LLChainInfo object to work on. */ void processChain(LLChainInfo& chain); /** * @brief Rewind through the chain to try to recover from an error. * * This method will potentially modify the internals of the * chain. * @param chain The LLChainInfo object to work on. * @return Retuns true if someone handled the error */ bool handleChainError(LLChainInfo& chain, LLIOPipe::EStatus error); public: /** * @brief Return number of running chains. * * *NOTE: This is only used in debugging and not considered * efficient or safe enough for production use. */ running_chains_t::size_type runningChains() const { return mRunningChains.size(); } }; #endif // LL_LLPUMPIO_H