llpumpio.h

Go to the documentation of this file.

#ifndef LL_LLPUMPIO_H
#define LL_LLPUMPIO_H

#include <set>
#if LL_LINUX  // needed for PATH_MAX in APR.
#include <sys/param.h>
#endif

#include "apr-1/apr_pools.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
{
public:
        LLPumpIO(apr_pool_t* pool);

        ~LLPumpIO();

        bool prime(apr_pool_t* pool);

        typedef std::vector<LLIOPipe::ptr_t> chain_t;

        bool addChain(const chain_t& chain, F32 timeout);
        
        struct LLLinkInfo
        {
                LLIOPipe::ptr_t mPipe;
                LLChannelDescriptors mChannels;
        };

        typedef std::vector<LLLinkInfo> links_t;

        bool addChain(
                const links_t& links,
                LLIOPipe::buffer_ptr_t data,
                LLSD context,
                F32 timeout);

        bool setTimeoutSeconds(F32 timeout);

        bool setConditional(LLIOPipe* pipe, const apr_pollfd_t* poll);

        S32 setLock();

        void clearLock(S32 key);

        bool sleepChain(F64 seconds);

        bool copyCurrentLinkInfo(links_t& links) const;

        void pump(const S32& poll_timeout);
        void pump();

        //void respond(const chain_t& pipes);

        bool respond(LLIOPipe* pipe);

        bool respond(
                const links_t& links,
                LLIOPipe::buffer_ptr_t data,
                LLSD context);

        void callback();

        enum EControl
        {
                PAUSE,
                RESUME,
        };
        
        void control(EControl op);

protected:
        enum EState
        {
                NORMAL,
                PAUSING,
                PAUSED
        };

        // instance data
        EState mState;
        bool mRebuildPollset;
        apr_pollset_t* mPollset;
        S32 mPollsetClientID;
        S32 mNextLock;
        std::set<S32> 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);

                // 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<LLIOPipe::ptr_t, apr_pollfd_t> pipe_conditional_t;
                typedef std::vector<pipe_conditional_t> conditionals_t;
                conditionals_t mDescriptors;
        };

        // All the running chains & info
        typedef std::vector<LLChainInfo> pending_chains_t;
        pending_chains_t mPendingChains;
        typedef std::list<LLChainInfo> 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<LLChainInfo> callbacks_t;
        callbacks_t mPendingCallbacks;
        callbacks_t mCallbacks;

        // memory allocator for pollsets & mutexes.
        apr_pool_t* mPool;
        apr_pool_t* 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(apr_pool_t* pool);
        void cleanup();

        void rebuildPollset();

        void processChain(LLChainInfo& chain);

        bool handleChainError(LLChainInfo& chain, LLIOPipe::EStatus error);
};


#endif // LL_LLPUMPIO_H

---