path: root/indra/llmessage/llpumpio.h

/**
 * @file llpumpio.h
 * @author Phoenix
 * @date 2004-11-19
 * @brief Declaration of pump class which manages io chains.
 *
 * $LicenseInfo:firstyear=2004&license=viewerlgpl$
 * Second Life Viewer Source Code
 * Copyright (C) 2010, Linden Research, Inc.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation;
 * version 2.1 of the License only.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
 *
 * Linden Research, Inc., 945 Battery Street, San Francisco, CA  94111  USA
 * $/LicenseInfo$
 */

#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_pools.h"
#include "llbuffer.h"
#include "llframetimer.h"
#include "lliopipe.h"
#include "llrun.h"

// 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
 * <code>pump()</code> on a thread used for IO and call
 * <code>respond()</code> 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. In order for the
 * threading abstraction to work, you need to call <code>prime()</code>
 * with a valid apr pool.
 * 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(apr_pool_t* pool);

    /**
     * @brief Destructor.
     */
    ~LLPumpIO();

    /**
     * @brief Prepare this pump for usage.
     *
     * If you fail to call this method prior to use, the pump will
     * try to work, but will not come with any thread locking
     * mechanisms.
     * @param pool The apr pool to use.
     * @return Returns true if the pump is primed.
     */
    bool prime(apr_pool_t* pool);

    /**
     * @brief Typedef for having a chain of pipes.
     */
    typedef std::vector<LLIOPipe::ptr_t> 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.
     * @param has_curl_request The chain contains LLURLRequest if true.
     * @return Returns true if anything was added to the pump.
     */
    bool addChain(const chain_t& chain, F32 timeout, bool has_curl_request = false);

    /**
     * @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 <code>LLLinkInfo</code>
     * instances.
     */
    typedef std::vector<LLLinkInfo> 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 <code>process()</code> 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
     * <code>process()</code> until you call <code>clearLock()</code>
     * 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 <code>process()</code> rather
     * than the pump which exposed this interface.
     * @return Returns the lock identifer to be used in
     * <code>clearLock()</code> 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 <em>not</em> 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 <code>process()</code> 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, <code>process()</code> 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 <code>callback()</code> and then
     * dropped from the queue.
     *
     * @param chain The IO chain that will get one <code>process()</code>.
     */
    //void respond(const chain_t& pipes);

    /**
     * @brief Add pipe to a special queue which will be called
     * during the next call to <code>callback()</code> 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 <code>callback()</code> 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 <code>process()</code>.
     *
     * 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<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);
        void adjustTimeoutSeconds(F32 delta);

        // basic member data
        bool mInit;
        bool mEOS;
        bool mHasCurlRequest;
        S32 mLock;
        LLFrameTimer mTimer;
        links_t::iterator mHead;
        links_t mChainLinks;
        LLIOPipe::buffer_ptr_t mData;
        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;

protected:
    void initialize(apr_pool_t* pool);
    void cleanup();
    current_chain_t removeRunningChain(current_chain_t& chain) ;
    /**
     * @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);

    //if the chain is expired, remove it
    bool isChainExpired(LLChainInfo& chain) ;

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