Allowing to break KV foreach loops (#221)

* Allowing to break KV foreach loops

src/ds/champmap.h

@@ -133,13 +133,15 @@ namespace champ
     }
 
     template <class F>
-    void foreach(F&& f) const
+    bool foreach(F&& f) const
     {
       for (const auto& bin : bins)
       {
         for (const auto& entry : bin)
-          f(entry->key, entry->value);
+          if (!f(entry->key, entry->value))
+            return false;
       }
+      return true;
     }
   };
 
@@ -274,21 +276,30 @@ namespace champ
     }
 
     template <class F>
-    void foreach(SmallIndex depth, F&& f) const
+    bool foreach(SmallIndex depth, F&& f) const
     {
       const auto entries = data_map.pop();
       for (SmallIndex i = 0; i < entries; ++i)
       {
         const auto& entry = node_as<Entry<K, V>>(i);
-        f(entry->key, entry->value);
+        if (!f(entry->key, entry->value))
+          return false;
       }
       for (size_t i = entries; i < nodes.size(); ++i)
       {
         if (depth == (collision_depth - 1))
-          node_as<Collisions<K, V, H>>(i)->foreach(std::forward<F>(f));
+        {
+          if (!node_as<Collisions<K, V, H>>(i)->foreach(std::forward<F>(f)))
+            return false;
+        }
         else
-          node_as<SubNodes<K, V, H>>(i)->foreach(depth + 1, std::forward<F>(f));
+        {
+          if (!node_as<SubNodes<K, V, H>>(i)->foreach(
+                depth + 1, std::forward<F>(f)))
+            return false;
+        }
       }
+      return true;
     }
 
   private:
@@ -340,9 +351,9 @@ namespace champ
     }
 
     template <class F>
-    void foreach(F&& f) const
+    bool foreach(F&& f) const
     {
-      root->foreach(0, std::forward<F>(f));
+      return root->foreach(0, std::forward<F>(f));
     }
   };
 }

src/ds/test/map_bench.cpp

@@ -75,8 +75,10 @@ static void benchmark_foreach(picobench::state& s)
   for (auto _ : s)
   {
     (void)_;
-    map.foreach(
-      [&v, s, map](const auto& key, const auto& value) { v += value; });
+    map.foreach([&v, s, map](const auto& key, const auto& value) {
+      v += value;
+      return true;
+    });
     clobber_memory();
   }
   s.stop_timer();

src/ds/test/map_test.cpp

@@ -114,6 +114,7 @@ TEST_CASE("persistent map operations")
         auto p = rb_new.get(k);
         REQUIRE(p.has_value());
         REQUIRE(p.value() == v);
+        return true;
       });
       REQUIRE(n == champ_new.size());
     }
@@ -126,6 +127,7 @@ TEST_CASE("persistent map operations")
         auto p = rb.get(k);
         REQUIRE(p.has_value());
         REQUIRE(p.value() == v);
+        return true;
       });
       REQUIRE(n == champ.size());
     }

src/genesisgen/genesisgen.h

@@ -80,8 +80,8 @@ public:
 
     nodes_view->foreach(
       [&start_node](const ccf::NodeId& id, const ccf::NodeInfo& v) {
-        if (start_node == ccf::INVALID_ID)
-          start_node = id;
+        start_node = id;
+        return false;
       });
 
     rpc.params.id = start_node;

src/kv/kv.h

@@ -195,34 +195,36 @@ namespace kv
         return false;
 
       size_t count = 0;
-      state2.state.foreach(
-        [&count](const K& k, const VersionV& v) { count++; });
+      state2.state.foreach([&count](const K& k, const VersionV& v) {
+        count++;
+        return true;
+      });
 
       size_t i = 0;
-      bool ok = true;
+      bool ok =
+        state1.state.foreach([&state2, &i](const K& k, const VersionV& v) {
+          auto search = state2.state.get(k);
 
-      state1.state.foreach([&state2, &ok, &i](const K& k, const VersionV& v) {
-        auto search = state2.state.get(k);
-
-        if (search.has_value())
-        {
-          auto& found = search.value();
-          if (found.version != v.version)
+          if (search.has_value())
           {
-            ok = false;
+            auto& found = search.value();
+            if (found.version != v.version)
+            {
+              return false;
+            }
+            else if (Check::ne(found.value, v.value))
+            {
+              return false;
+            }
           }
-          else if (Check::ne(found.value, v.value))
+          else
           {
-            ok = false;
+            return false;
           }
-        }
-        else
-        {
-          ok = false;
-        }
 
-        i++;
-      });
+          i++;
+          return true;
+        });
 
       if (i != count)
         ok = false;
@@ -430,13 +432,14 @@ namespace kv
 
       /** Iterate over all entries in the map
        *
-       * @param F functor, taking a key and a value, return value is ignored
+       * @param F functor, taking a key and a value, return value determines
+       * whether the iteration should continue (true) or stop (false)
        */
       template <class F>
-      void foreach(F&& f)
+      bool foreach(F&& f)
       {
         if (commit_version != NoVersion)
-          return;
+          return false;
 
         // Record a global read dependency.
         read_version = start_version;
@@ -446,14 +449,17 @@ namespace kv
           auto write = w.find(k);
 
           if ((write == w.end()) && !deleted(v.version))
-            f(k, v.value);
+            return f(k, v.value);
+          return true;
         });
 
         for (auto write = writes.begin(); write != writes.end(); ++write)
         {
           if (!deleted(write->second.version))
-            f(write->first, write->second.value);
+            if (!f(write->first, write->second.value))
+              return false;
         }
+        return true;
       }
 
       Version start_order()

src/kv/test/kv_test.cpp

@@ -182,6 +182,20 @@ TEST_CASE("Reads/writes and deletions")
     auto vc = view3->get(k);
     REQUIRE(!vc.has_value());
   }
+
+  INFO("Test early temination of KV foreach");
+  {
+    Store::Tx tx;
+    auto view = tx.get_view(map);
+    view->put("key1", "value1");
+    view->put("key2", "value2");
+    size_t ctr = 0;
+    view->foreach([&ctr](const auto& key, const auto& value) {
+      ++ctr;
+      return false;
+    });
+    REQUIRE(ctr == 1);
+  }
 }
 
 TEST_CASE("Rollback and compact")

src/luainterp/luakv.h

@@ -141,6 +141,7 @@ namespace ccf
 
           // Call the lua functor. This pops the args and functor-copy
           lua_pcall(l, ifunc, 0, 0);
+          return true;
         });
 
         return 0;

src/node/nodestate.h

@@ -635,6 +635,7 @@ namespace ccf
 #endif
           result.quotes.push_back(quote);
         }
+        return true;
       });
     };
 
@@ -655,6 +656,7 @@ namespace ccf
           // Only retire nodes that have not already been retired
           if (ni.status != ccf::NodeStatus::RETIRED)
             nodes_to_delete[nid] = ni;
+          return true;
         });
       for (auto [nid, ni] : nodes_to_delete)
       {
@@ -666,6 +668,7 @@ namespace ccf
       certs_view->foreach(
         [&certs_to_delete](const Cert& cstr, const NodeId& _) {
           certs_to_delete.push_back(cstr);
+          return true;
         });
       for (Cert& cstr : certs_to_delete)
       {
@@ -796,6 +799,7 @@ namespace ccf
         [&new_followers, this](const NodeId& nid, const NodeInfo& ni) {
           if (ni.status != ccf::NodeStatus::RETIRED && nid != self)
             new_followers[nid] = ni;
+          return true;
         });
 
       // For all nodes in the new network, write all past network secrets to the
@@ -1108,6 +1112,7 @@ namespace ccf
           s.foreach([&](NodeId node_id, const Nodes::VersionV& v) {
             if (v.value.status != NodeStatus::RETIRED)
               configuration.insert(node_id);
+            return true;
           });
           raft->add_configuration(version, move(configuration));
         }

src/node/rpc/frontend.h

@@ -262,6 +262,7 @@ namespace ccf
             {
               out.nodes.push_back({nid, ni.pubhost, ni.tlsport});
             }
+            return true;
           });
 
           return jsonrpc::success(out);

src/node/rpc/memberfrontend.h

@@ -397,14 +397,14 @@ namespace ccf
 #endif
         auto nodes_view = args.tx.get_view(this->network.nodes);
         NodeId duplicate_node_id = NoNode;
-        // TODO(#api): foreach should check the callback's value and be able to
-        // stop once the returned value is false
         nodes_view->foreach([&new_node, &duplicate_node_id](
                               const NodeId& nid, const NodeInfo& ni) {
-          if (
-            duplicate_node_id == NoNode &&
-            (new_node.tlsport == ni.tlsport && new_node.host == ni.host))
+          if (new_node.tlsport == ni.tlsport && new_node.host == ni.host)
+          {
             duplicate_node_id = nid;
+            return false;
+          }
+          return true;
         });
         if (duplicate_node_id != NoNode)
           return jsonrpc::error(