core/router/router.h

/* rtb_router.h                                                   -*- C++ -*-
   Jeremy Barnes, 13 March 2011
   Copyright (c) 2011 Datacratic.  All rights reserved.

   Router class for RTB.
*/

#ifndef __rtb__router_h__
#define __rtb__router_h__

#include "soa/service/zmq.hpp"
#include <unordered_map>
#include <boost/thread/thread.hpp>
#include <boost/scoped_ptr.hpp>
#include "jml/utils/filter_streams.h"
#include "soa/service/zmq_named_pub_sub.h"
#include "soa/service/socket_per_thread.h"
#include "soa/service/timeout_map.h"
#include "soa/service/pending_list.h"
#include "augmentation_loop.h"
#include "router_types.h"
#include "soa/gc/gc_lock.h"
#include "jml/utils/ring_buffer.h"
#include "jml/arch/wakeup_fd.h"
#include "jml/utils/smart_ptr_utils.h"
#include <unordered_set>
#include <thread>
#include "rtbkit/common/exchange_connector.h"
#include "rtbkit/core/agent_configuration/blacklist.h"
#include "rtbkit/core/agent_configuration/agent_configuration_listener.h"
#include "rtbkit/core/agent_configuration/agent_config.h"
#include "rtbkit/core/monitor/monitor_provider.h"
#include "rtbkit/core/monitor/monitor_client.h"

namespace RTBKIT {

struct Banker;
struct BudgetController;
struct Accountant;


/*****************************************************************************/
/* AGENT INFO                                                                */
/*****************************************************************************/

struct AgentInfoEntry {
    std::string name;
    std::shared_ptr<const AgentConfig> config;
    std::shared_ptr<const AgentStatus> status;
    std::shared_ptr<AgentStats> stats;

    bool valid() const { return config && stats; }

    Json::Value toJson() const;
};


struct AllAgentInfo : public std::vector<AgentInfoEntry> {
    std::unordered_map<std::string, int> agentIndex;
    std::unordered_map<AccountKey, std::vector<int> > accountIndex;
};

/*****************************************************************************/
/* DEBUG INFO                                                                */
/*****************************************************************************/

struct AuctionDebugInfo {
    void addAuctionEvent(Date date, std::string type,
                         const std::vector<std::string> & args);
    void addSpotEvent(const Id & spot, Date date, std::string type,
                      const std::vector<std::string> & args);
    void dumpAuction() const;
    void dumpSpot(Id spot) const;

    struct Message {
        Date timestamp;
        Id spot;
        std::string type;
        std::vector<std::string> args;
    };

    std::vector<Message> messages;
};

/*****************************************************************************/
/* ROUTER                                                                    */
/*****************************************************************************/

struct Router : public ServiceBase,
                public MonitorProvider
{
    Router(ServiceBase & parent,
           const std::string & serviceName = "router",
           double secondsUntilLossAssumed = 2.0,
           bool connectPostAuctionLoop = true);

    Router(std::shared_ptr<ServiceProxies> services,
           const std::string & serviceName = "router",
           double secondsUntilLossAssumed = 2.0,
           bool connectPostAuctionLoop = true);

    ~Router();

    double secondsUntilLossAssumed() const { return secondsUntilLossAssumed_; }

    void setSecondsUntilLossAssumed(double newValue)
    {
        if (newValue < 0.0)
            throw ML::Exception("invalid seconds until loss assumed");
        secondsUntilLossAssumed_ = newValue;
    }

    std::shared_ptr<Banker> getBanker() const;
    void setBanker(const std::shared_ptr<Banker> & newBanker);

    void init();

    void bindTcp();

    void bindAgents(std::string agentUri);

    void bindAugmentors(const std::string & uri);

    void unsafeDisableMonitor();

    virtual void
    start(boost::function<void ()> onStop = boost::function<void ()>());

    virtual void sleepUntilIdle();

    void issueTimestamp();

    virtual size_t numNonIdle() const;
    
    virtual void shutdown();

    template<typename F>
    void forAllExchanges(F functor) {
        for(auto & item : exchanges)
            functor(item);
    }

    void connectExchange(ExchangeConnector & exchange)
    {
        exchange.onNewAuction  = [=] (std::shared_ptr<Auction> a) { this->injectAuction(a, secondsUntilLossAssumed_); };
        exchange.onAuctionDone = [=] (std::shared_ptr<Auction> a) { this->onAuctionDone(a); };
    }

    void addExchange(ExchangeConnector * exchange)
    {
        Guard guard(lock);
        exchanges.push_back(std::shared_ptr<ExchangeConnector>(exchange));
        connectExchange(*exchange);
    }

    void addExchange(ExchangeConnector & exchange)
    {
        Guard guard(lock);
        exchanges.emplace_back(ML::make_unowned_std_sp(exchange));
        connectExchange(exchange);
    }
    
    void addExchange(std::unique_ptr<ExchangeConnector> && exchange)
    {
        addExchange(exchange.release());
    }

    void startExchange(const std::string & exchangeType,
                       const Json::Value & exchangeConfig);

    void startExchange(const Json::Value & exchangeConfig);

    void injectAuction(std::shared_ptr<Auction> auction,
                       double lossTime = INFINITY);
    
    std::shared_ptr<Auction>
    injectAuction(Auction::HandleAuction onAuctionFinished,
                  std::shared_ptr<BidRequest> request,
                  const std::string & requestStr,
                  const std::string & requestStrFormat,
                  double startTime = 0.0,
                  double expiryTime = 0.0,
                  double lossTime = INFINITY);

    void notifyFinishedAuction(const Id & auctionId);

    int numAuctionsInProgress() const;

    int numAuctionsAwaitingResult() const;

    Json::Value getStats() const;

    Json::Value getAgentInfo(const std::string & agent) const;

    Json::Value getAllAgentInfo() const;

    Json::Value getAccountInfo(const AccountKey & account) const;
    
    void setGlobalBidProbability(double val) { globalBidProbability = val; }
    
    void setBidsErrorRate(double val) { bidsErrorRate = val; }

    void setBudgetErrorRate(double val) { budgetErrorRate = val; }

    boost::function<void (double)> acceptAuctionProbabilityFn;

    void setAcceptAuctionProbability(double val)
    {
        if (acceptAuctionProbabilityFn)
            acceptAuctionProbabilityFn(val);
        else if (val != 1.0)
            std::cerr << "warning: no way to change accept auction probability" << std::endl;
    }

    virtual Json::Value getServiceStatus() const;

    std::function<void (std::shared_ptr<Auction>, Id, Auction::Response)>
        onSubmittedAuction;

    virtual void submitToPostAuctionService(std::shared_ptr<Auction> auction,
                                            Id auctionId,
                                            const Auction::Response & bid);

protected:
    // This thread contains the main router loop
    boost::scoped_ptr<boost::thread> runThread;

    // This thread wakes up every now and again to run the destructors
    // on auction objects, which are expensive to destroy, so that they
    // don't have to run in the main loop
    boost::scoped_ptr<boost::thread> cleanupThread;

    typedef std::recursive_mutex Lock;
    typedef std::unique_lock<Lock> Guard;

    int shutdown_;

public:
    // Connection to the agents
    ZmqNamedClientBus agentEndpoint;

    // Connection to the post auction loop
    ZmqNamedProxy postAuctionEndpoint;

    void updateAllAgents();

    typedef std::map<std::string, AgentInfo> Agents;
    Agents agents;

    ML::RingBufferSRMW<std::pair<std::string, std::shared_ptr<const AgentConfig> > > configBuffer;
    ML::RingBufferSRMW<std::shared_ptr<AugmentationInfo> > startBiddingBuffer;
    ML::RingBufferSRMW<std::shared_ptr<Auction> > submittedBuffer;
    ML::RingBufferSWMR<std::shared_ptr<Auction> > auctionGraveyard;

    ML::Wakeup_Fd wakeupMainLoop;

    AugmentationLoop augmentationLoop;
    Blacklist blacklist;

    typedef TimeoutMap<Id, AuctionInfo> InFlight;
    InFlight inFlight;

    AuctionInfo &
    addAuction(std::shared_ptr<Auction> auction, Date timeout);

    DutyCycleEntry dutyCycleCurrent;
    std::vector<DutyCycleEntry> dutyCycleHistory;

    void run();

    void handleAgentMessage(const std::vector<std::string> & message);

    void checkDeadAgents();

    void checkExpiredAuctions();

    void returnErrorResponse(const std::vector<std::string> & message,
                             const std::string & error);

    void doShutdown();

    std::shared_ptr<AugmentationInfo>
    preprocessAuction(const std::shared_ptr<Auction> & auction);

    void augmentAuction(const std::shared_ptr<AugmentationInfo> & info);

    void doStartBidding(const std::vector<std::string> & message);

    void doStartBidding(const std::shared_ptr<AugmentationInfo> & augInfo);

    void doSubmitted(std::shared_ptr<Auction> auction);

    //std::unordered_set<Id> recentlySubmitted;  // DEBUG

    void doBid(const std::vector<std::string> & message);

    void doPong(int level, const std::vector<std::string> & message);

    void sendPings();

    void doStats(const std::vector<std::string> & message);

    void onNewAuction(std::shared_ptr<Auction> auction);

    void onAuctionDone(std::shared_ptr<Auction> auction);

    void doConfig(const std::string & agent,
                  std::shared_ptr<const AgentConfig> config);

    void unconfigure(const std::string & agent, const AgentConfig & config);

    void configure(const std::string & agent, AgentConfig & config);

    template<typename... Args>
    void sendAgentMessage(const std::string & agent,
                          const std::string & messageType,
                          const Date & date,
                          Args... args)
    {
        agentEndpoint.sendMessage(agent, messageType, date, args...);
    }

    void sendBidResponse(const std::string & agent,
                         const AgentInfo & info,
                         BidStatus status,
                         Date timestamp,
                         const std::string & message,
                         const Id & auctionId,
                         int spotNum = -1,
                         Amount price = Amount(),
                         const Auction * auction = 0,
                         const std::string & bidData = "",
                         const Json::Value & metadata = Json::Value(),
                         const std::string & augmentationsStr = "");
                         

    mutable Lock lock;

    std::shared_ptr<Banker> banker;

    double secondsUntilLossAssumed_;
    double globalBidProbability;
    double bidsErrorRate;
    double budgetErrorRate;
    bool connectPostAuctionLoop;


    /*************************************************************************/
    /* AGENT INTERACTIONS                                                    */
    /*************************************************************************/

    AllAgentInfo * allAgents;

    mutable GcLock allAgentsGc;

    typedef std::function<void (const AgentInfoEntry & info)> OnAgentFn;
    void forEachAgent(const OnAgentFn & onAgent) const;

    void forEachAccountAgent(const AccountKey & account,
                             const OnAgentFn & onAgent) const;

    AgentInfoEntry getAgentEntry(const std::string & agent) const;

    AgentConfigurationListener configListener;

    bool initialized;

    std::vector<std::shared_ptr<ExchangeConnector> > exchanges;

    /*************************************************************************/
    /* EXCEPTIONS                                                            */
    /*************************************************************************/

    void throwException(const std::string & key, const std::string & fmt,
                        ...) __attribute__((__noreturn__));


    /*************************************************************************/
    /* SYSTEM LOGGING                                                        */
    /*************************************************************************/

    template<typename... Args>
    void logRouterError(const std::string & function,
                        const std::string & exception,
                        Args... args)
    {
        logger.publish("ROUTERERROR", Date::now().print(5),
                       function, exception, args...);
        recordHit("error.%s", function);
    }


    /*************************************************************************/
    /* DATA LOGGING                                                          */
    /*************************************************************************/

    template<typename... Args>
    void logMessage(const std::string & channel, Args... args)
    {
        using namespace std;
        //cerr << "********* logging message to " << channel << endl;
        logger.publish(channel, Date::now().print(5), args...);
    }

    template<typename... Args>
    void logMessageNoTimestamp(const std::string & channel, Args... args)
    {
        using namespace std;
        //cerr << "********* logging message to " << channel << endl;
        logger.publish(channel, args...);
    }

    /*************************************************************************/
    /* DEBUGGING                                                             */
    /*************************************************************************/

    void debugAuction(const Id & auction, const std::string & type,
                      const std::vector<std::string> & args
                      = std::vector<std::string>())
    {
        if (JML_LIKELY(!doDebug)) return;
        debugAuctionImpl(auction, type, args);
    }

    void debugAuctionImpl(const Id & auction, const std::string & type,
                          const std::vector<std::string> & args);

    void debugSpot(const Id & auction,
                   const Id & spot,
                   const std::string & type,
                   const std::vector<std::string> & args
                       = std::vector<std::string>())
    {
        if (JML_LIKELY(!doDebug)) return;
        debugSpotImpl(auction, spot, type, args);
    }

    void debugSpotImpl(const Id & auction,
                       const Id & spot,
                       const std::string & type,
                       const std::vector<std::string> & args);

    void expireDebugInfo();

    void dumpAuction(const Id & auction) const;
    void dumpSpot(const Id & auction, const Id & spot) const;

    Date getCurrentTime() const { return Date::now(); }

    ZmqNamedPublisher logger;

    bool doDebug;

    mutable ML::Spinlock debugLock;
    TimeoutMap<Id, AuctionDebugInfo> debugInfo;

    uint64_t numAuctions;
    uint64_t numBids;
    uint64_t numNonEmptyBids;
    uint64_t numAuctionsWithBid;
    uint64_t numNoPotentialBidders;
    uint64_t numNoBidders;

    /* Client connection to the Monitor, determines if we can process bid
       requests */
    MonitorClient monitorClient;
    Date slowModeLastAuction;
    int slowModeCount;

    /* MONITOR PROVIDER */
    /* Post service health status to Monitor */
    MonitorProviderClient monitorProviderClient;

    /* MonitorProvider interface */
    std::string getProviderName() const;
    Json::Value getProviderIndicators() const;
};


} // namespace RTBKIT


#endif /* __rtb__router_h__ */