hub/sweep_service: optimize bundle

hub/db/connection.h

@@ -268,8 +268,7 @@ class Connection {
   /// @param[in] requiredAmount - the required total amount to obtain
   /// @param[in] olderThan - the cutoff point in time
   /// @return std::vector - a list of TransferInput
-  virtual std::vector<TransferInput> getHubInputsForSweep(
-      uint64_t requiredAmount,
+  virtual std::vector<TransferInput> availableHubInputs(
       const std::chrono::system_clock::time_point& olderThan) = 0;
 
   /// @param[in] sweepId - the sweepId if the creation results from a sweep

hub/db/connection_impl.h

@@ -233,11 +233,9 @@ class ConnectionImpl : public Connection {
     return db::helper<Conn>::getDepositsForSweep(*_conn, max, olderThan);
   };
 
-  std::vector<TransferInput> getHubInputsForSweep(
-      uint64_t requiredAmount,
+  std::vector<TransferInput> availableHubInputs(
       const std::chrono::system_clock::time_point& olderThan) override {
-    return db::helper<Conn>::getHubInputsForSweep(*_conn, requiredAmount,
-                                                  olderThan);
+    return db::helper<Conn>::availableHubInputs(*_conn, olderThan);
   }
 
   bool isSweepConfirmed(uint64_t sweepId) override {

hub/db/helper.cc

@@ -657,9 +657,8 @@ std::vector<TransferInput> helper<C>::getDepositsForSweep(
 }
 
 template <typename C>
-std::vector<TransferInput> helper<C>::getHubInputsForSweep(
-    C& connection, uint64_t requiredAmount,
-    const std::chrono::system_clock::time_point& olderThan) {
+std::vector<TransferInput> helper<C>::availableHubInputs(
+    C& connection, const std::chrono::system_clock::time_point& olderThan) {
   db::sql::HubAddress add;
   db::sql::HubAddressBalance bal;
   db::sql::Sweep swp;
@@ -714,17 +713,7 @@ std::vector<TransferInput> helper<C>::getHubInputsForSweep(
   std::sort(availableInputs.begin(), availableInputs.end(),
             [](const auto& a, const auto& b) { return a.amount < b.amount; });
 
-  std::vector<TransferInput> selectedInputs;
-  auto it = availableInputs.begin();
-
-  uint64_t total = 0;
-  while (it != std::end(availableInputs) && total < requiredAmount) {
-    total += it->amount;
-    selectedInputs.emplace_back(std::move(*it));
-    it++;
-  }
-
-  return selectedInputs;
+  return availableInputs;
 }
 
 template <typename C>

hub/db/helper.h

@@ -125,9 +125,8 @@ struct helper {
       C& connection, size_t max,
       const std::chrono::system_clock::time_point& olderThan);
 
-  static std::vector<TransferInput> getHubInputsForSweep(
-      C& connection, uint64_t requiredAmount,
-      const std::chrono::system_clock::time_point& olderThan);
+  static std::vector<TransferInput> availableHubInputs(
+      C& connection, const std::chrono::system_clock::time_point& olderThan);
 
   static nonstd::optional<AddressInfo> getAddressInfo(
       C& connection, const common::crypto::Address& address);

hub/service/sweep_service.cc

@@ -7,6 +7,7 @@
 
 #include "hub/service/sweep_service.h"
 
+#include <algorithm>
 #include <chrono>
 #include <numeric>
 #include <sstream>
@@ -46,10 +47,16 @@ DEFINE_uint32(sweep_max_deposit, 5,
 namespace hub {
 namespace service {
 
-db::TransferOutput SweepService::getHubOutput(uint64_t remainder) {
+db::TransferOutput SweepService::getHubOutput(
+    uint64_t remainder, nonstd::optional<db::TransferInput> optionalHubOutput) {
+  if (optionalHubOutput.has_value()) {
+    return {optionalHubOutput.value().addressId,
+            remainder,
+            {},
+            std::move(optionalHubOutput.value().address)};
+  }
   auto& dbConnection = db::DBManager::get().connection();
   auto& cryptoProvider = common::crypto::CryptoManager::get().provider();
-
   common::crypto::UUID hubOutputUUID;
   auto address = cryptoProvider.getAddressForUUID(hubOutputUUID).value();
 
@@ -265,12 +272,14 @@ bool SweepService::doTick() {
     std::vector<db::TransferInput> hubInputs;
     uint64_t hubInputTotal = 0;
 
+    nonstd::optional<db::TransferInput> optionalOutputInput;
+
     if (depositsTotal < requiredOutput) {
       auto missing = requiredOutput - depositsTotal;
 
       LOG(INFO) << "Need to fill " << missing << " via Hub addresses.";
 
-      hubInputs = dbConnection.getHubInputsForSweep(missing, sweepStart);
+      hubInputs = dbConnection.availableHubInputs(sweepStart);
 
       hubInputTotal = std::accumulate(
           hubInputs.cbegin(), hubInputs.cend(), 0uLL,
@@ -286,6 +295,33 @@ bool SweepService::doTick() {
         hubInputs.clear();
         requiredOutput = 0;
         withdrawals.clear();
+      } else {
+        std::vector<db::TransferInput> minimalVecOfInputs;
+        getVecOfMinSizeWithSumNotLessThan(missing, hubInputs,
+                                          minimalVecOfInputs);
+
+        // Last element in vec is the smallest, but it's good to sort
+        // in case someone ever changes that behavior
+        std::sort(
+            minimalVecOfInputs.begin(), minimalVecOfInputs.end(),
+            [](const auto& a, const auto& b) { return a.amount < b.amount; });
+
+        // First element in vec is the smallest, but it's good to sort
+        // in case someone ever changes db query that gets that vec
+        std::sort(
+            hubInputs.begin(), hubInputs.end(),
+            [](const auto& a, const auto& b) { return a.amount < b.amount; });
+
+        // if poorest hub address was not used
+        if (!hubInputs.empty() && hubInputs.front().addressId !=
+                                      minimalVecOfInputs.front().addressId) {
+          optionalOutputInput = hubInputs.front();
+        }
+
+        hubInputs.swap(minimalVecOfInputs);
+        hubInputTotal = std::accumulate(
+            hubInputs.cbegin(), hubInputs.cend(), 0uLL,
+            [](uint64_t a, const auto& b) { return a + b.amount; });
       }
     }
 
@@ -302,7 +338,7 @@ bool SweepService::doTick() {
 
     // 4. Determine Hub output address
     auto remainder = (hubInputTotal + depositsTotal) - requiredOutput;
-    auto hubOutput = getHubOutput(remainder);
+    auto hubOutput = getHubOutput(remainder, optionalOutputInput);
 
     LOG(INFO) << "Will move " << remainder
               << " into new Hub address: " << hubOutput.payoutAddress.str();
@@ -328,5 +364,27 @@ bool SweepService::doTick() {
 
   return true;
 }
+
+void SweepService::getVecOfMinSizeWithSumNotLessThan(
+    uint64_t amount, const std::vector<db::TransferInput>& hubInputs,
+    std::vector<db::TransferInput>& minVecHubInputs) {
+  auto inputs = hubInputs;
+  // Sort lowest to highest - input should already be sorted, but just in case
+  std::sort(inputs.begin(), inputs.end(),
+            [](const auto& a, const auto& b) { return a.amount < b.amount; });
+  uint64_t missing = amount;
+
+  while (missing && !inputs.empty()) {
+    auto low = std::lower_bound(
+        inputs.begin(), inputs.end(), missing,
+        [](const auto& inp, uint64_t missing) { return inp.amount < missing; });
+    if (low == inputs.end()) {
+      --low;
+    }
+    minVecHubInputs.push_back(*low);
+    inputs.erase(low);
+    missing = low->amount > missing ? 0 : missing - low->amount;
+  }
+}
 }  // namespace service
 }  // namespace hub

hub/service/sweep_service.h

@@ -41,6 +41,10 @@ class SweepService : public ScheduledService {
   /// @return string - the descriptive name of the service
   const std::string name() const override { return "SweepService"; }
 
+  static void getVecOfMinSizeWithSumNotLessThan(
+      uint64_t amount, const std::vector<db::TransferInput>& hubInputs,
+      std::vector<db::TransferInput>& minVecHubInputs);
+
  protected:
   /// Called by tick() by default. Override in subclasses
   /// @return false if it wants to stop.
@@ -53,7 +57,10 @@ class SweepService : public ScheduledService {
   /// - the new hub address
   /// - the id of the new hub address
   /// - the remainder
-  db::TransferOutput getHubOutput(uint64_t remainder);
+  /// - an optional hub input that will use as change address
+  db::TransferOutput getHubOutput(
+      uint64_t remainder,
+      nonstd::optional<db::TransferInput> optionalHubOutput);
 
   /// Compute and serialize a bundle made up of all the deposits and withdrawals
   /// identified during the sweep.

hub/service/tests/sweep_service.cc

@@ -0,0 +1,94 @@
+// Copyright 2018 IOTA Foundation
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include <chrono>
+#include <thread>
+#include <vector>
+#include <numeric>
+
+
+#include "hub/service/sweep_service.h"
+
+using namespace testing;
+
+using namespace hub;
+using namespace hub::service;
+
+namespace {
+
+class SweepServiceTest : public ::testing::Test {};
+
+class MockService : public SweepService {
+ public:
+  using SweepService::SweepService;
+
+  const std::string name() const override { return "MockService"; }
+  MOCK_METHOD0(doTick, bool());
+};
+
+TEST_F(SweepServiceTest, getVecOfMinSizeWithSumNotLessThan) {
+
+    std::vector<db::TransferInput> inputs ;
+    std::vector<db::TransferInput> res ;
+
+    common::crypto::UUID uuid;
+    common::crypto::Address address(
+            "999999999999999999999999999999999999999999999999999999999999999999999999"
+            "999999999");
+
+    //{1, 4, 45, 6, 0, 19};
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 0});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 1});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 4});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 6});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 19});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 45});
+
+    SweepService::getVecOfMinSizeWithSumNotLessThan(51,inputs,res);
+
+    auto sum = std::accumulate(res.cbegin(), res.cend(), 0uLL,
+    [](uint64_t a, const auto& b) { return a + b.amount; });
+
+    EXPECT_EQ(res.size(),2);
+    EXPECT_EQ(sum,51);/*45+6*/
+
+    inputs.clear();
+    res.clear();
+    //{1, 11, 100, 1, 0, 200, 3, 2, 1, 250}
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 0});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 1});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 1});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 1});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 2});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 3});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 11});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 100});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 200});
+    inputs.emplace_back(hub::db::TransferInput{0,0, address, uuid, 250});
+
+    SweepService::getVecOfMinSizeWithSumNotLessThan(280,inputs,res);
+
+    sum = std::accumulate(res.cbegin(), res.cend(), 0uLL,
+                               [](uint64_t a, const auto& b) { return a + b.amount; });
+
+    EXPECT_EQ(res.size(),2);
+    EXPECT_EQ(sum,350);
+
+    //check bigger sum than inputs have
+    res.clear();
+    SweepService::getVecOfMinSizeWithSumNotLessThan(9999,inputs,res);
+    EXPECT_EQ(res.size(),inputs.size());
+
+    //check nothing crashes
+    inputs.clear();
+    res.clear();
+
+    EXPECT_EQ(res.size(),0);
+
+
+
+}
+
+};  // namespace