<commit_before>/** post_auction_service.cc                                 -*- C++ -*-
    <NAME>, 18 Apr 2014
    Copyright (c) 2014 Datacratic.  All rights reserved.

    Implementation of the post auction service.

*/

#include "post_auction_service.h"
#include "simple_event_matcher.h"
#include "sharded_event_matcher.h"
#include "event_forwarder.h"
#include "rtbkit/common/messages.h"
#include "soa/service/rest_request_params.h"
#include "soa/service/rest_request_binding.h"
#include "rtbkit/common/analytics.h"

using namespace std;
using namespace Datacratic;
using namespace ML;

namespace RTBKIT {


/******************************************************************************/
/* POST AUCTION SERVICE                                                       */
/******************************************************************************/

Logging::Category PostAuctionService::print("PostAuctionService");
Logging::Category PostAuctionService::error("PostAuctionService Error", PostAuctionService::print);
Logging::Category PostAuctionService::trace("PostAuctionService Trace", PostAuctionService::print);


PostAuctionService::
PostAuctionService(
        std::shared_ptr<ServiceProxies> proxies, const std::string & serviceName)
    : ServiceBase(serviceName, proxies),

      auctionTimeout(EventMatcher::DefaultAuctionTimeout),
      winTimeout(EventMatcher::DefaultWinTimeout),
      winLossPipeTimeout(DefaultWinLossPipeTimeout),
      campaignEventPipeTimeout(DefaultCampaignEventPipeTimeout),

      loopMonitor(*this),
      configListener(getZmqContext()),
      monitorProviderClient(getZmqContext()),

      auctions(65536),
      events(65536),

      endpoint(getZmqContext()),
      bridge(getZmqContext()),
      router(!!getZmqContext()),

      totalEvents(0),
      orphanEvents(0),
      orphanRatios(30, 0)
{
    monitorProviderClient.addProvider(this);
}

PostAuctionService::
PostAuctionService(ServiceBase & parent, const std::string & serviceName)
    : ServiceBase(serviceName, parent),

      auctionTimeout(EventMatcher::DefaultAuctionTimeout),
      winTimeout(EventMatcher::DefaultWinTimeout),

      loopMonitor(*this),
      configListener(getZmqContext()),
      monitorProviderClient(getZmqContext()),

      auctions(65536),
      events(65536),

      endpoint(getZmqContext()),
      bridge(getZmqContext()),
      router(!!getZmqContext()),

      totalEvents(0),
      orphanEvents(0),
      orphanRatios(30, 0)
{
    monitorProviderClient.addProvider(this);
}

PostAuctionService::~PostAuctionService()
{
    shutdown();
}

void
PostAuctionService::
bindTcp()
{
    if (analytics) analytics->bindTcp();
    endpoint.bindTcp(getServices()->ports->getRange("postAuctionLoop"));
    bridge.agents.bindTcp(getServices()->ports->getRange("postAuctionLoopAgents"));

    if (restEndpoint) {
        restEndpoint->bindTcp(
                getServices()->ports->getRange("postAuctionREST.zmq"),
                getServices()->ports->getRange("postAuctionREST.http"));
    }
}

void
PostAuctionService::
init(size_t externalShard, size_t internalShards)
{
    recordHit("up");

    // Loop monitor is purely for monitoring purposes. There's no message we can
    // just drop in the PAL to alleviate the load.
    loopMonitor.init();
    loopMonitor.addMessageLoop("postAuctionLoop", &loop);

    if(!bidder) {
        Json::Value json;
        json["type"] = "agents";
        initBidderInterface(json);
    }

    initMatcher(internalShards);
    initConnections(externalShard);
    initRestEndpoint();
    monitorProviderClient.init(getServices()->config);

    auto checkOrphans = [=] (double) {
        double ratio = 0;
        if (totalEvents > 0) ratio = double(orphanEvents) / double(totalEvents);
        orphanEvents = totalEvents = 0;

        orphanRatios.pop_back();
        orphanRatios.insert(orphanRatios.begin(), ratio);

        ratio = accumulate(orphanRatios.begin(), orphanRatios.end(), 0);
        ratio /= orphanRatios.size();

        ExcCheckLess(ratio, 0.1, "Excessive orphaned events detected");
    };
    loop.addPeriodic("PostAuctionService::checkOrphans", 60.0, checkOrphans);
}

void
PostAuctionService::
initBidderInterface(Json::Value const & json)
{
    bidder = BidderInterface::create(serviceName() + ".bidder", getServices(), json);
    bidder->init(&bridge);
}

void
PostAuctionService::
initMatcher(size_t shards)
{
    if (shards <= 1) {
        LOG(print) << "Creating SimpleEventMatcher" << endl;<commit_msg>increased orphaned events to 50%<commit_after>/** post_auction_service.cc                                 -*- C++ -*-
    <NAME>, 18 Apr 2014
    Copyright (c) 2014 Datacratic.  All rights reserved.

    Implementation of the post auction service.

*/

#include "post_auction_service.h"
#include "simple_event_matcher.h"
#include "sharded_event_matcher.h"
#include "event_forwarder.h"
#include "rtbkit/common/messages.h"
#include "soa/service/rest_request_params.h"
#include "soa/service/rest_request_binding.h"
#include "rtbkit/common/analytics.h"

using namespace std;
using namespace Datacratic;
using namespace ML;

namespace RTBKIT {


/******************************************************************************/
/* POST AUCTION SERVICE                                                       */
/******************************************************************************/

Logging::Category PostAuctionService::print("PostAuctionService");
Logging::Category PostAuctionService::error("PostAuctionService Error", PostAuctionService::print);
Logging::Category PostAuctionService::trace("PostAuctionService Trace", PostAuctionService::print);


PostAuctionService::
PostAuctionService(
        std::shared_ptr<ServiceProxies> proxies, const std::string & serviceName)
    : ServiceBase(serviceName, proxies),

      auctionTimeout(EventMatcher::DefaultAuctionTimeout),
      winTimeout(EventMatcher::DefaultWinTimeout),
      winLossPipeTimeout(DefaultWinLossPipeTimeout),
      campaignEventPipeTimeout(DefaultCampaignEventPipeTimeout),

      loopMonitor(*this),
      configListener(getZmqContext()),
      monitorProviderClient(getZmqContext()),

      auctions(65536),
      events(65536),

      endpoint(getZmqContext()),
      bridge(getZmqContext()),
      router(!!getZmqContext()),

      totalEvents(0),
      orphanEvents(0),
      orphanRatios(30, 0)
{
    monitorProviderClient.addProvider(this);
}

PostAuctionService::
PostAuctionService(ServiceBase & parent, const std::string & serviceName)
    : ServiceBase(serviceName, parent),

      auctionTimeout(EventMatcher::DefaultAuctionTimeout),
      winTimeout(EventMatcher::DefaultWinTimeout),

      loopMonitor(*this),
      configListener(getZmqContext()),
      monitorProviderClient(getZmqContext()),

      auctions(65536),
      events(65536),

      endpoint(getZmqContext()),
      bridge(getZmqContext()),
      router(!!getZmqContext()),

      totalEvents(0),
      orphanEvents(0),
      orphanRatios(30, 0)
{
    monitorProviderClient.addProvider(this);
}

PostAuctionService::~PostAuctionService()
{
    shutdown();
}

void
PostAuctionService::
bindTcp()
{
    if (analytics) analytics->bindTcp();
    endpoint.bindTcp(getServices()->ports->getRange("postAuctionLoop"));
    bridge.agents.bindTcp(getServices()->ports->getRange("postAuctionLoopAgents"));

    if (restEndpoint) {
        restEndpoint->bindTcp(
                getServices()->ports->getRange("postAuctionREST.zmq"),
                getServices()->ports->getRange("postAuctionREST.http"));
    }
}

void
PostAuctionService::
init(size_t externalShard, size_t internalShards)
{
    recordHit("up");

    // Loop monitor is purely for monitoring purposes. There's no message we can
    // just drop in the PAL to alleviate the load.
    loopMonitor.init();
    loopMonitor.addMessageLoop("postAuctionLoop", &loop);

    if(!bidder) {
        Json::Value json;
        json["type"] = "agents";
        initBidderInterface(json);
    }

    initMatcher(internalShards);
    initConnections(externalShard);
    initRestEndpoint();
    monitorProviderClient.init(getServices()->config);

    auto checkOrphans = [=] (double) {
        double ratio = 0;
        if (totalEvents > 0) ratio = double(orphanEvents) / double(totalEvents);
        orphanEvents = totalEvents = 0;

        orphanRatios.pop_back();
        orphanRatios.insert(orphanRatios.begin(), ratio);

        ratio = accumulate(orphanRatios.begin(), orphanRatios.end(), 0);
        ratio /= orphanRatios.size();

        ExcCheckLess(ratio, 0.5, "Excessive orphaned events detected");
    };
    loop.addPeriodic("PostAuctionService::checkOrphans", 60.0, checkOrphans);
}

void
PostAuctionService::
initBidderInterface(Json::Value const & json)
{
    bidder = BidderInterface::create(serviceName() + ".bidder", getServices(), json);
    bidder->init(&bridge);
}

void
PostAuctionService::
initMatcher(size_t shards)
{
    if (shards <= 1) {
        LOG(print) << "Creating SimpleEventMatcher" << endl;
{'fraction_non_alphanumeric': 0.15562950637825845, 'fraction_numerical': 0.011314475873544094, 'mean_word_length': 4.0172509738452975, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6533646', 'n_tokens_mistral': 2748, 'n_tokens_neox': 2566, 'n_words': 443}