BIND 10 trac2875, updated. 90763b13076f29e3b35506e904724dce93edfe84 [2875] Launch the child processes

[BIND 10 source code commits]

The branch, trac2875 has been updated
       via  90763b13076f29e3b35506e904724dce93edfe84 (commit)
      from  ad25e8e1d467ce0398d8b868da979be581c83f43 (commit)

Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.

- Log -----------------------------------------------------------------
commit 90763b13076f29e3b35506e904724dce93edfe84
Author: Michal 'vorner' Vaner <michal.vaner at nic.cz>
Date:   Fri Jul 19 10:04:06 2013 +0200

    [2875] Launch the child processes
    
    Create the tree of child processes, waiting for them to launch and
    signal them to work or signal they finished.

-----------------------------------------------------------------------

Summary of changes:
 src/bin/resolver/bench/layers.cc |   91 ++++++++++++++++++++++++++++++++++++++
 src/bin/resolver/bench/layers.h  |    3 +-
 2 files changed, 93 insertions(+), 1 deletion(-)

-----------------------------------------------------------------------
diff --git a/src/bin/resolver/bench/layers.cc b/src/bin/resolver/bench/layers.cc
index 5088d4b..e0d805d 100644
--- a/src/bin/resolver/bench/layers.cc
+++ b/src/bin/resolver/bench/layers.cc
@@ -15,14 +15,105 @@
 #include <resolver/bench/layers.h>
 #include <resolver/bench/subprocess.h>
 
+#include <boost/bind.hpp>
+
+#include <sys/types.h>
+#include <sys/socket.h>
+
 namespace isc {
 namespace resolver {
 namespace bench {
 
+LayerResolver::LayerResolver(size_t count, size_t worker_count,
+                             size_t fanout) :
+    total_count_(count),
+    top(new Subprocess(boost::bind(&LayerResolver::spawn, this, count,
+                                   worker_count, fanout, _1)))
+{
+    // Wait for the subprocess to become ready
+    const std::string& ready = top->read(1);
+    assert(ready == "R");
+}
+
 LayerResolver::~LayerResolver() {
     delete top;
 }
 
+void
+LayerResolver::worker(size_t count, int channel) {
+    FakeInterface interface(count);
+    // We are ready
+    ssize_t result = send(channel, "R", 1, 0);
+    assert(result == 1);
+    // Wait for the run signal
+    char buffer;
+    result = recv(channel, &buffer, 1, 0);
+    assert(result == 1);
+    assert(buffer == 'R');
+
+    // TODO: Do the work
+    // Signal we are finished
+    result = send(channel, "F", 1, 0);
+    assert(result == 1);
+}
+
+namespace {
+
+// Single child subprocess. May be a smaller cache or a worker.
+// We don't really care about private/public here, this is to make
+// it hold together.
+class Child : boost::noncopyable {
+public:
+    Child(const boost::function<void(int)>& main) :
+        subprocess_(main)
+    {}
+    Subprocess subprocess_;
+};
+
+}
+
+void
+LayerResolver::spawn(size_t count, size_t worker_count, size_t fanout,
+                     int channel)
+{
+    if (worker_count == 1) {
+        // We are the single worker
+        worker(count, channel);
+    } else {
+        if (fanout > worker_count) {
+            // Just correction, so we don't spawn more children than needed
+            fanout = worker_count;
+        }
+        std::vector<Child*> children;
+        // How much of these was satisfied
+        size_t count_handled = 0;
+        size_t workers_handled = 0;
+        for (size_t i = 0; i < fanout; ++i) {
+            // How many to assign to this child
+            size_t count_local = (count * (i+1)) / fanout - count_handled;
+            size_t workers_local = (count * (i+1)) / fanout - workers_handled;
+            // Initialize the child and make sure it is ready
+            Child* child = new Child(boost::bind(&LayerResolver::spawn, this,
+                                                 count_local, workers_local,
+                                                 fanout, _1));
+            children.push_back(child);
+            const std::string& ready = child->subprocess_.read(1);
+            assert(ready == "R");
+
+            count_handled += count_local;
+            workers_handled += workers_local;
+        }
+        // All the children started, we are ready. Signal it and wait for
+        // signal to launch.
+        ssize_t result = send(channel, "R", 1, 0);
+        assert(result == 1);
+        char buffer;
+        result = recv(channel, &buffer, 1, 0);
+        assert(result == 1);
+        assert(buffer == 'R');
+    }
+}
+
 size_t
 LayerResolver::run() {
     // Send "Run" signal
diff --git a/src/bin/resolver/bench/layers.h b/src/bin/resolver/bench/layers.h
index 361900d..9afaf8a 100644
--- a/src/bin/resolver/bench/layers.h
+++ b/src/bin/resolver/bench/layers.h
@@ -32,8 +32,9 @@ public:
     size_t run();
 private:
     void spawn(size_t count, size_t worker_count, size_t fanout, int channel);
-    Subprocess* top;
+    void worker(size_t count, int channel);
     size_t total_count_;
+    Subprocess* top;
 };
 
 }