Http application layer benchmarks

rtbkit/core/banker/application_layer.cc

@@ -158,7 +158,8 @@ syncAccount(const ShadowAccount &account, const std::string &shadowStr,
 void
 HttpLayer::
 request(std::string method, const std::string &resource,
-        const RestParams &params, const std::string &content, OnRequestResult onResult)
+        const RestParams &params, const std::string &content,
+        const OnRequestResult & onResult)
 {
     std::transform(begin(method), end(method), begin(method), [](char c) { return ::tolower(c); });
 
@@ -315,7 +316,8 @@ syncAccount(const ShadowAccount &account, const std::string &shadowStr,
 void
 ZmqLayer::
 request(std::string method, const std::string &resource,
-        const RestParams &params, const std::string &content, OnRequestResult onResult)
+        const RestParams &params, const std::string &content,
+        const OnRequestResult & onResult)
 {
     proxy->push(onResult, method, resource, params, content);
 }

rtbkit/core/banker/application_layer.h

@@ -71,11 +71,10 @@ struct ApplicationLayer : public MessageLoop {
                                          Account &&)> onDone) = 0;
 
     /* CUSTOM */
-    virtual void request(
-                       std::string method, const std::string &resource,
-                       const RestParams &params,
-                       const std::string &content,
-                       OnRequestResult onResult) = 0;
+    virtual void request(std::string method, const std::string &resource,
+                         const RestParams &params,
+                         const std::string &content,
+                         const OnRequestResult & onResult) = 0;
 };
 
 /*****************************************************************************/
@@ -120,9 +119,9 @@ struct HttpLayer : public ApplicationLayer {
                                    Account &&)> onDone);
 
     void request(std::string method, const std::string &resource,
-               const RestParams &params,
-               const std::string &content,
-               OnRequestResult onResult);
+                 const RestParams &params,
+                 const std::string &content,
+                 const OnRequestResult & onResult);
 private:
     std::shared_ptr<HttpClient> httpClient;
 
@@ -172,9 +171,9 @@ struct ZmqLayer : public ApplicationLayer {
                      std::function<void (std::exception_ptr,
                                    Account &&)> onDone);
     void request(std::string method, const std::string &resource,
-               const RestParams &params,
-               const std::string &content,
-               OnRequestResult onResult);
+                 const RestParams &params,
+                 const std::string &content,
+                 const OnRequestResult & onResult);
 private:
     std::shared_ptr<RestProxy> proxy;
 

rtbkit/core/banker/slave_banker.cc

@@ -159,7 +159,7 @@ init(const std::string & accountSuffix, CurrencyPool spendRate)
                           std::placeholders::_1),
                 true /* single threaded */);
     addPeriodic("SlaveBanker::reauthorizeBudget", 1.0,
-                std::bind(&SlaveBanker::reauthorizeBudget,
+                std::bind(&SlaveBanker::reauthorizeBudgetPeriodic,
                           this,
                           std::placeholders::_1),
                 true /* single threaded */);
@@ -408,56 +408,78 @@ reportSpend(uint64_t numTimeoutsExpired)
 
 void
 SlaveBanker::
-reauthorizeBudget(uint64_t numTimeoutsExpired)
+reauthorizeBudgetPeriodic(uint64_t numTimeoutsExpired)
 {
     if (numTimeoutsExpired > 1) {
         cerr << "warning: slave banker missed " << numTimeoutsExpired
              << " timeouts" << endl;
     }
 
-    //std::unique_lock<Lock> guard(lock);
     if (reauthorizing) {
-        cerr << "warning: reauthorize budget still in progress" << endl;
+        cerr << "warning: reauthorize budget still in progress (skipping)\n";
         return;
     }
 
-    accountsLeft = 0;
+    auto onReauthorizedDone = [&] () {
+        reauthorizing = false;
+        numReauthorized++;
+    };
+    reauthorizing = true;
+    reauthorizeBudget(onReauthorizedDone);
+}
 
-    // For each of our accounts, we report back what has been spent
-    // and re-up to our desired float
-    auto onAccount = [&] (const AccountKey & key,
-                          const ShadowAccount & account)
-        {
-            Json::Value payload = spendRate.toJson();
 
-            auto onDone = std::bind(&SlaveBanker::onReauthorizeBudgetMessage, this,
-                                    key,
-                                    std::placeholders::_1, std::placeholders::_2,
-                                    std::placeholders::_3);
+void
+SlaveBanker::
+reauthorizeBudget(const OnReauthorizeBudgetDone & onDone)
+{
+    //std::unique_lock<Lock> guard(lock);
 
-            accountsLeft++;
+    // For each of our accounts, we report back what has been spent
+    // and re-up to our desired float
 
-            // Finally, send it out
-            applicationLayer->request(
-                            "POST", "/v1/accounts/" + getShadowAccountStr(key) + "/balance",
-                            { { "accountType", "spend" } },
-                            payload.toString(),
-                            onDone);
+    auto reauthorizeOp = make_shared<ReauthorizeOp>();
+    reauthorizeOp->start = Date::now();
+    reauthorizeOp->numAccounts = 0;
+    reauthorizeOp->pending = 0;
+    reauthorizeOp->onDone = onDone;
+
+    auto onAccount
+        = [&] (const AccountKey & key, const ShadowAccount & account) {
+        reauthorizeOp->numAccounts++;
+        reauthorizeOp->pending++;
+
+        // Finally, send it out
+        Json::Value payload = spendRate.toJson();
+        auto onAccountDone = [=] (exception_ptr exc, int responseCode,
+                                  const string & payload) {
+            this->onReauthorizeBudgetMessage(key, exc, responseCode, payload);
+            reauthorizeOp->pending--;
+            if (reauthorizeOp->pending == 0) {
+                Date now = Date::now();
+                if (reauthorizeOp->onDone) {
+                    reauthorizeOp->onDone();
+                }
+            }
         };
+
+        applicationLayer->request("POST", "/v1/accounts/" + getShadowAccountStr(key) + "/balance",
+                                  { { "accountType", "spend" } },
+                                  payload.toString(),
+                                  onAccountDone);
+    };
     accounts.forEachInitializedAccount(onAccount);
-
-    if (accountsLeft > 0) {
-        reauthorizing = true;
-        reauthorizeDate = Date::now();
+    if (reauthorizeOp->numAccounts == 0) {
+        if (reauthorizeOp->onDone) {
+            reauthorizeOp->onDone();
+        }
     }
 }
 
 void
 SlaveBanker::
-onReauthorizeBudgetMessage(const AccountKey & accountKey,
-                           std::exception_ptr exc,
-                           int responseCode,
-                           const std::string & payload)
+onReauthorizeBudgetMessage(const AccountKey & accountKey, exception_ptr exc,
+                           int responseCode, const string & payload)
 {
     if (exc) {
         cerr << "reauthorize budget got exception" << payload << endl;
@@ -469,13 +491,6 @@ onReauthorizeBudgetMessage(const AccountKey & accountKey,
         Account masterAccount = Account::fromJson(Json::parse(payload));
         accounts.syncFromMaster(accountKey, masterAccount);
     }
-    reauthorizeBudgetSent = Date();
-    accountsLeft--;
-    if (accountsLeft == 0) {
-        lastReauthorizeDelay = Date::now() - reauthorizeDate;
-        numReauthorized++;
-        reauthorizing = false;
-    }
 }
 
 void

rtbkit/core/banker/slave_banker.h

@@ -19,7 +19,6 @@
 
 namespace RTBKIT {
 
-
 /*****************************************************************************/
 /* SLAVE BUDGET CONTROLLER                                                   */
 /*****************************************************************************/
@@ -81,6 +80,7 @@ struct SlaveBudgetController
     budgetResultCallback(const SlaveBudgetController::OnBudgetResult & onResult);
 private:
     std::shared_ptr<ApplicationLayer> applicationLayer;
+
     //std::shared_ptr<HttpClient> httpClient;
 };
 
@@ -212,18 +212,11 @@ struct SlaveBanker : public Banker, public MessageLoop {
 
     void waitReauthorized() const;
 
-    size_t getNumReauthorized()
-        const
+    int getNumReauthorized() const
     {
         return numReauthorized;
     }
 
-    double getLastReauthorizeDelay()
-        const
-    {
-        return lastReauthorizeDelay;
-    }
-
     /* Logging */
     virtual void logBidEvents(const Datacratic::EventRecorder & eventRecorder)
     {
@@ -246,16 +239,26 @@ struct SlaveBanker : public Banker, public MessageLoop {
     mutable Lock syncLock;
     Datacratic::Date lastSync;
 
-
     /** Periodically we report spend to the banker.*/
     void reportSpend(uint64_t numTimeoutsExpired);
     Date reportSpendSent;
 
     /** Periodically we ask the banker to re-authorize our budget. */
-    void reauthorizeBudget(uint64_t numTimeoutsExpired);
-    Date reauthorizeBudgetSent;
-    CurrencyPool spendRate;
+    void reauthorizeBudgetPeriodic(uint64_t numTimeoutsExpired);
+
+    typedef std::function<void()> OnReauthorizeBudgetDone;
+    void reauthorizeBudget(const OnReauthorizeBudgetDone & onDone = nullptr);
+
+    struct ReauthorizeOp {
+        Date start;
 
+        int numAccounts;
+        std::atomic<int> pending;
+
+        OnReauthorizeBudgetDone onDone;
+    };
+
+    CurrencyPool spendRate;
 
     /// Called when we get an account status back from the master banker
     /// after a synchrnonization
@@ -288,11 +291,8 @@ struct SlaveBanker : public Banker, public MessageLoop {
 
     std::atomic<bool> shutdown_;
 
-    std::atomic<bool> reauthorizing;
-    Date reauthorizeDate;
-    double lastReauthorizeDelay;
-    size_t numReauthorized;
-    size_t accountsLeft;
+    bool reauthorizing;
+    int numReauthorized;
 };
 
 } // naemspace RTBKIT