//------------------------------------------------------------------------------
/*
    Copyright (c) 2011-2013, OpenCoin, Inc.
*/
//==============================================================================

SETUP_LOG (LoadManager)

//------------------------------------------------------------------------------

class LoadManager
    : public ILoadManager
    , public beast::InterruptibleThread::EntryPoint
{
private:
    /*  Entry mapping utilization to cost.

        The cost is expressed as a unitless relative quantity. These
        mappings are statically loaded at startup with heuristic values.
    */
    class Cost
    {
    public:
        // VFALCO TODO Eliminate this default ctor somehow
        Cost ()
            : m_loadType ()
            , m_cost (0)
            , m_resourceFlags (0)
        {
        }
        
        Cost (LoadType loadType, int cost, int resourceFlags)
            : m_loadType (loadType)
            , m_cost (cost)
            , m_resourceFlags (resourceFlags)
        {
        }

        LoadType getLoadType () const
        {
            return m_loadType;
        }

        int getCost () const
        {
            return m_cost;
        }

        int getResourceFlags () const
        {
            return m_resourceFlags;
        }

    public:
        // VFALCO TODO Make these private and const
        LoadType    m_loadType;
        int         m_cost;
        int         m_resourceFlags;
    };

public:
    LoadManager ()
        : m_thread ("loadmgr")
        , m_logThread ("loadmgr_log")
        , mCreditRate (100)
        , mCreditLimit (500)
        , mDebitWarn (-500)
        , mDebitLimit (-1000)
        , mArmed (false)
        , mDeadLock (0)
        , mCosts (LT_MAX)
    {
        m_logThread.start ();

        /** Flags indicating the type of load.

            Utilization may include any combination of:

            - CPU
            - Storage space
            - Network transfer
        */
        // VFALCO NOTE These flags are not used...
        enum
        {
            flagDisk = 1,
            flagCpu  = 2,
            flagNet  = 4
        };

        // VFALCO TODO Replace this with a function that uses a simple switch statement...
        //
        addCost (Cost (LT_InvalidRequest,     -10,   flagCpu | flagNet));
        addCost (Cost (LT_RequestNoReply,      -1,   flagCpu | flagDisk));
        addCost (Cost (LT_InvalidSignature,  -100,   flagCpu));
        addCost (Cost (LT_UnwantedData,        -5,   flagCpu | flagNet));
        addCost (Cost (LT_BadData,            -20,   flagCpu));

        addCost (Cost (LT_RPCInvalid,         -10,   flagCpu | flagNet));
        addCost (Cost (LT_RPCReference,       -10,   flagCpu | flagNet));
        addCost (Cost (LT_RPCException,       -20,   flagCpu | flagNet));
        addCost (Cost (LT_RPCBurden,          -50,   flagCpu | flagNet));

        // VFALCO NOTE Why do these supposedly "good" load types still have a negative cost?
        //
        addCost (Cost (LT_NewTrusted,         -10,   0));
        addCost (Cost (LT_NewTransaction,      -2,   0));
        addCost (Cost (LT_NeededData,         -10,   0));

        addCost (Cost (LT_RequestData,         -5,   flagDisk | flagNet));
        addCost (Cost (LT_CheapQuery,          -1,   flagCpu));

        UptimeTimer::getInstance ().beginManualUpdates ();
    }

private:
    ~LoadManager ()
    {
        UptimeTimer::getInstance ().endManualUpdates ();

        m_thread.interrupt ();
    }

    void startThread ()
    {
        m_thread.start (this);
    }

    void canonicalize (LoadSource& source, int now) const
    {
        if (source.mLastUpdate != now)
        {
            if (source.mLastUpdate < now)
            {
                source.mBalance += mCreditRate * (now - source.mLastUpdate);

                if (source.mBalance > mCreditLimit)
                {
                    source.mBalance = mCreditLimit;
                    source.mLogged = false;
                }
            }

            source.mLastUpdate = now;
        }
    }

    bool shouldWarn (LoadSource& source) const
    {
        {
            boost::mutex::scoped_lock sl (mLock);

            int now = UptimeTimer::getInstance ().getElapsedSeconds ();
            canonicalize (source, now);

            if (source.isPrivileged () || (source.mBalance > mDebitWarn) || (source.mLastWarning == now))
                return false;

            source.mLastWarning = now;
        }
        logWarning (source.getName ());
        return true;
    }

    bool shouldCutoff (LoadSource& source) const
    {
        {
            boost::mutex::scoped_lock sl (mLock);
            int now = UptimeTimer::getInstance ().getElapsedSeconds ();
            canonicalize (source, now);

            if (source.isPrivileged () || (source.mBalance > mDebitLimit))
                return false;

            if (source.mLogged)
                return true;

            source.mLogged = true;
        }
        logDisconnect (source.getName ());
        return true;
    }

    bool applyLoadCharge (LoadSource& source, LoadType loadType) const
    {
        // FIXME: Scale by category
        Cost cost = mCosts[static_cast<int> (loadType)];
 
        return adjust (source, cost.m_cost);
    }

    bool adjust (LoadSource& source, int credits) const
    {
        // return: true = need to warn/cutoff

        // We do it this way in case we want to add exponential decay later
        int now = UptimeTimer::getInstance ().getElapsedSeconds ();

        boost::mutex::scoped_lock sl (mLock);
        canonicalize (source, now);
        source.mBalance += credits;

        if (source.mBalance > mCreditLimit)
            source.mBalance = mCreditLimit;

        if (source.isPrivileged ()) // privileged sources never warn/cutoff
            return false;

        if ( (source.mBalance >= mDebitWarn) ||
            ((source.mBalance >= mDebitLimit) && (source.mLastWarning == now)))
            return false;

        return true;
    }

    void logWarning (const std::string& source) const
    {
        if (source.empty ())
            WriteLog (lsDEBUG, LoadManager) << "Load warning from empty source";
        else
            WriteLog (lsINFO, LoadManager) << "Load warning: " << source;
    }

    void logDisconnect (const std::string& source) const
    {
        if (source.empty ())
            WriteLog (lsINFO, LoadManager) << "Disconnect for empty source";
        else
            WriteLog (lsWARNING, LoadManager) << "Disconnect for: " << source;
    }

    // VFALCO TODO Implement this and stop accessing the vector directly
    //Cost const& getCost (LoadType loadType) const;
    int getCost (LoadType t) const
    {
        return mCosts [static_cast <int> (t)].getCost ();
    }

    void resetDeadlockDetector ()
    {
        boost::mutex::scoped_lock sl (mLock);
        mDeadLock = UptimeTimer::getInstance ().getElapsedSeconds ();
    }

    void activateDeadlockDetector ()
    {
        mArmed = true;
    }

    static void logDeadlock (int dlTime)
    {
        WriteLog (lsWARNING, LoadManager) << "Server stalled for " << dlTime << " seconds.";

        char const* fileName = getApp ().getMasterLock ().getFileName ();
        int lineNumber = getApp ().getMasterLock ().getLineNumber ();

        WriteLog (lsWARNING, LoadManager) << "Master lock owned by " << File (fileName).getFileName ().toStdString () << ", line " << lineNumber;
    }

private:
    // VFALCO TODO These used to be public but are apparently not used. Find out why.
    /*
    int getCreditRate () const
    {
        boost::mutex::scoped_lock sl (mLock);
        return mCreditRate;
    }

    int getCreditLimit () const
    {
        boost::mutex::scoped_lock sl (mLock);
        return mCreditLimit;
    }

    int getDebitWarn () const
    {
        boost::mutex::scoped_lock sl (mLock);
        return mDebitWarn;
    }

    int getDebitLimit () const
    {
        boost::mutex::scoped_lock sl (mLock);
        return mDebitLimit;
    }

    void setCreditRate (int r)
    {
        boost::mutex::scoped_lock sl (mLock);
        mCreditRate = r;
    }

    void setCreditLimit (int r)
    {
        boost::mutex::scoped_lock sl (mLock);
        mCreditLimit = r;
    }

    void setDebitWarn (int r)
    {
        boost::mutex::scoped_lock sl (mLock);
        mDebitWarn = r;
    }

    void setDebitLimit (int r)
    {
        boost::mutex::scoped_lock sl (mLock);
        mDebitLimit = r;
    }
    */

private:
    void addCost (const Cost& c)
    {
        mCosts [static_cast <int> (c.getLoadType ())] = c;
    }

    // VFALCO NOTE Where's the thread object? It's not a data member...
    //
    void threadRun ()
    {
        // VFALCO TODO replace this with a beast Time object?
        //
        // Initialize the clock to the current time.
        boost::posix_time::ptime t = boost::posix_time::microsec_clock::universal_time ();

        while (! m_thread.interruptionPoint ())
        {
            {
                // VFALCO NOTE What is this lock protecting?
                boost::mutex::scoped_lock sl (mLock);

                // VFALCO NOTE I think this is to reduce calls to the operating system
                //             for retrieving the current time.
                //
                //        TODO Instead of incrementing can't we just retrieve the current
                //             time again?
                //
                // Manually update the timer.
                UptimeTimer::getInstance ().incrementElapsedTime ();

                // Measure the amount of time we have been deadlocked, in seconds.
                //
                // VFALCO NOTE mDeadLock is a canary for detecting the condition.
                int const timeSpentDeadlocked = UptimeTimer::getInstance ().getElapsedSeconds () - mDeadLock;

                // VFALCO NOTE I think that "armed" refers to the deadlock detector
                //
                if (mArmed && (timeSpentDeadlocked >= 10))
                {
                    // Report the deadlocked condition every 10 seconds
                    if ((timeSpentDeadlocked % 10) == 0)
                    {
                        // VFALCO TODO Replace this with a dedicated thread with call queue.
                        //
                        m_logThread.call (&logDeadlock, timeSpentDeadlocked);
                    }

                    // If we go over 500 seconds spent deadlocked, it means that the
                    // deadlock resolution code has failed, which qualifies as undefined
                    // behavior.
                    //
                    assert (timeSpentDeadlocked < 500);
                }
            }

            bool change;

            // VFALCO TODO Eliminate the dependence on the Application object.
            //             Choices include constructing with the job queue / feetracker.
            //             Another option is using an observer pattern to invert the dependency.
            if (getApp().getJobQueue ().isOverloaded ())
            {
                WriteLog (lsINFO, LoadManager) << getApp().getJobQueue ().getJson (0);
                change = getApp().getFeeTrack ().raiseLocalFee ();
            }
            else
            {
                change = getApp().getFeeTrack ().lowerLocalFee ();
            }

            if (change)
            {
                // VFALCO TODO replace this with a Listener / observer and subscribe in NetworkOPs or Application
                getApp().getOPs ().reportFeeChange ();
            }

            t += boost::posix_time::seconds (1);
            boost::posix_time::time_duration when = t - boost::posix_time::microsec_clock::universal_time ();

            if ((when.is_negative ()) || (when.total_seconds () > 1))
            {
                WriteLog (lsWARNING, LoadManager) << "time jump";
                t = boost::posix_time::microsec_clock::universal_time ();
            }
            else
            {
                boost::this_thread::sleep (when);
            }
        }
    }

private:
    beast::InterruptibleThread m_thread;
    beast::ThreadWithCallQueue m_logThread;

    int mCreditRate;            // credits gained/lost per second
    int mCreditLimit;           // the most credits a source can have
    int mDebitWarn;             // when a source drops below this, we warn
    int mDebitLimit;            // when a source drops below this, we cut it off (should be negative)

    bool mArmed;

    int mDeadLock;              // Detect server deadlocks

    mutable boost::mutex mLock; // VFALCO TODO Replace with juce::Mutex and remove the mutable attribute

    std::vector <Cost> mCosts;
};

//------------------------------------------------------------------------------

ILoadManager* ILoadManager::New ()
{
    return new LoadManager;
}
