API: colummn_family.hh yield in map_reduce_column_families_locally

map_reduce_column_families_locally iterate over all tables (column
family) in a shard.

If the number of tables is big it can cause latency spikes.

This patch replaces the current loop with a do_for_each allowing
preepmtion inside the loop.

Fixes scylladb#3886

Signed-off-by: Amnon Heiman <amnon@scylladb.com>
Message-Id: <20181115154825.23430-1-amnon@scylladb.com>

api/column_family.hh

@@ -24,6 +24,7 @@
 #include "api.hh"
 #include "api/api-doc/column_family.json.hh"
 #include "database.hh"
+#include "core/future-util.hh"
 #include <any>
 
 namespace api {
@@ -71,12 +72,13 @@ struct map_reduce_column_families_locally {
     std::any init;
     std::function<std::any (column_family&)> mapper;
     std::function<std::any (std::any, std::any)> reducer;
-    std::any operator()(database& db) const {
-        auto res = init;
-        for (auto i : db.get_column_families()) {
-            res = reducer(res, mapper(*i.second.get()));
-        }
-        return res;
+    future<std::any> operator()(database& db) const {
+        auto res = seastar::make_lw_shared<std::any>(init);
+        return do_for_each(db.get_column_families(), [res, this](const std::pair<utils::UUID, seastar::lw_shared_ptr<table>>& i) {
+            *res = reducer(*res.get(), mapper(*i.second.get()));
+        }).then([res] {
+            return *res;
+        });
     }
 };