Parallelised merge, distribution phase.erlmerge_dist

1 files changed, 51 insertions, 24 deletions

diff --git a/tools/compileconfig.py b/tools/compileconfig.py
index b5e5053..aea7ace 100755
--- a/tools/compileconfig.py
+++ b/tools/compileconfig.py
@@ -8,6 +8,7 @@ import sys
 import readconfig
 import re
 import base64
+from datetime import datetime
 
 class Symbol(str):
     pass
@@ -131,24 +132,24 @@ def gen_http_servers(nodetype, nodeconfig, bind_addresses, bind_publicaddress, b
     return (http_servers,
             https_servers)
 
-def allowed_clients_frontend(mergenodenames, primarymergenode):
+def allowed_clients_frontend(mergenodenames, primarymergenodename):
     return [
         ("/plop/v1/frontend/sendentry", mergenodenames),
         ("/plop/v1/frontend/sendlog", mergenodenames),
-        ("/plop/v1/frontend/publish-sth", [primarymergenode]),
-        ("/plop/v1/frontend/verify-entries", [primarymergenode]),
+        ("/plop/v1/frontend/publish-sth", [primarymergenodename]),
+        ("/plop/v1/frontend/verify-entries", [primarymergenodename]),
         ("/plop/v1/frontend/currentposition", mergenodenames),
         ("/plop/v1/frontend/missingentries", mergenodenames),
     ]
 
-def allowed_clients_mergesecondary(primarymergenode):
+def allowed_clients_mergesecondary(primarymergenodename):
     return [
-        ("/plop/v1/merge/sendentry", [primarymergenode]),
-        ("/plop/v1/merge/sendlog", [primarymergenode]),
-        ("/plop/v1/merge/verifyroot", [primarymergenode]),
-        ("/plop/v1/merge/verifiedsize", [primarymergenode]),
-        ("/plop/v1/merge/setverifiedsize", [primarymergenode]),
-        ("/plop/v1/merge/missingentries", [primarymergenode]),
+        ("/plop/v1/merge/sendentry", [primarymergenodename]),
+        ("/plop/v1/merge/sendlog", [primarymergenodename]),
+        ("/plop/v1/merge/verifyroot", [primarymergenodename]),
+        ("/plop/v1/merge/verifiedsize", [primarymergenodename]),
+        ("/plop/v1/merge/setverifiedsize", [primarymergenodename]),
+        ("/plop/v1/merge/missingentries", [primarymergenodename]),
     ]
 
 def allowed_clients_public():
@@ -164,10 +165,10 @@ def allowed_clients_public():
         ("/ct/v1/get-roots", noauth),
     ]
 
-def allowed_clients_signing(frontendnodenames, primarymergenode):
+def allowed_clients_signing(frontendnodenames, primarymergenodename):
     return [
         ("/plop/v1/signing/sct", frontendnodenames),
-        ("/plop/v1/signing/sth", [primarymergenode]),
+        ("/plop/v1/signing/sth", [primarymergenodename]),
     ]
 
 def allowed_clients_storage(frontendnodenames, mergenodenames):
@@ -185,6 +186,14 @@ def allowed_servers_frontend(signingnodenames, storagenodenames):
         ("/plop/v1/signing/sct", signingnodenames),
     ]
 
+def allowed_servers_primarymerge(frontendnodenames):
+    return [
+        ("/plop/v1/frontend/verify-entries", frontendnodenames),
+        ("/plop/v1/frontend/sendlog", frontendnodenames),
+        ("/plop/v1/frontend/sendentry", frontendnodenames),
+        ("/plop/v1/frontend/publish-sth", frontendnodenames),
+    ]
+
 def parse_ratelimit_expression(expression):
     if expression == "none":
         return Symbol("none")
@@ -228,7 +237,10 @@ def gen_config(nodename, config, localconfig):
     print >>plopconfigfile, "%% plop configuration file (-*- erlang -*-)"
 
     (nodetype, nodeconfig) = get_node_config(nodename, config)
-    (http_servers, https_servers) = gen_http_servers(nodetype, nodeconfig, bind_addresses, bind_publicaddress, bind_publichttpaddress=bind_publichttpaddress)
+    if nodename == config["primarymergenode"]:
+        (http_servers, https_servers) = [], []
+    else:
+        (http_servers, https_servers) = gen_http_servers(nodetype, nodeconfig, bind_addresses, bind_publicaddress, bind_publichttpaddress=bind_publichttpaddress)
 
     catlfishconfig = []
     plopconfig = []
@@ -248,8 +260,6 @@ def gen_config(nodename, config, localconfig):
         (Symbol("http_servers"), http_servers),
         (Symbol("https_certfile"), paths["https_certfile"]),
         (Symbol("https_keyfile"), paths["https_keyfile"]),
-        (Symbol("https_cacertfile"), paths["https_cacertfile"]),
-        (Symbol("https_cacert_fingerprint"), Binary(base64.b16decode(config["cafingerprint"]))),
     ]
 
     catlfishconfig.append((Symbol("mmd"), config["mmd"]))
@@ -263,6 +273,11 @@ def gen_config(nodename, config, localconfig):
         ])
     ]
 
+    plopconfig += [
+        (Symbol("https_cacertfile"), paths["https_cacertfile"]),
+        (Symbol("https_cacert_fingerprint"), Binary(base64.b16decode(config["cafingerprint"]))),
+    ]
+
     if "dbbackend" in localconfig:
         dbbackend = localconfig["dbbackend"]
         if dbbackend not in ("fsdb", "permdb"):
@@ -275,7 +290,7 @@ def gen_config(nodename, config, localconfig):
             print >>sys.stderr, "When using permdb, all services have to be in the same node"
             sys.exit(1)
 
-    print "nodetype", ", ".join(nodetype)
+    #print "nodetype", ", ".join(nodetype)
     if nodetype & set(["frontendnodes", "storagenodes"]):
         plopconfig += [
             (Symbol("entry_root_path"), paths["db"] + "certentries"),
@@ -310,9 +325,10 @@ def gen_config(nodename, config, localconfig):
     signingnodes = config["signingnodes"]
     signingnodeaddresses = ["https://%s/plop/v1/signing/" % node["address"] for node in config["signingnodes"]]
     mergenodenames = [node["name"] for node in config["mergenodes"]]
-    primarymergenode = config["primarymergenode"]
+    primarymergenodename = config["primarymergenode"]
     storagenodeaddresses = ["https://%s/plop/v1/storage/" % node["address"] for node in config["storagenodes"]]
     frontendnodenames = [node["name"] for node in config["frontendnodes"]]
+    frontendnodeaddresses = ["https://%s/plop/v1/frontend/" % node["address"] for node in config["frontendnodes"]]
 
     allowed_clients = []
     allowed_servers = []
@@ -323,20 +339,30 @@ def gen_config(nodename, config, localconfig):
         reloadableplopconfig.append((Symbol("storage_nodes"), storagenodeaddresses))
         reloadableplopconfig.append((Symbol("storage_nodes_quorum"), config["storage-quorum-size"]))
         services.add(Symbol("ht"))
-        allowed_clients += allowed_clients_frontend(mergenodenames, primarymergenode)
+        allowed_clients += allowed_clients_frontend(mergenodenames, primarymergenodename)
         allowed_clients += allowed_clients_public()
         allowed_servers += allowed_servers_frontend([node["name"] for node in signingnodes], storagenodenames)
     if "storagenodes" in nodetype:
         allowed_clients += allowed_clients_storage(frontendnodenames, mergenodenames)
     if "signingnodes" in nodetype:
-        allowed_clients += allowed_clients_signing(frontendnodenames, primarymergenode)
+        allowed_clients += allowed_clients_signing(frontendnodenames, primarymergenodename)
         services = [Symbol("sign")]
     if "mergenodes" in nodetype:
-        storagenodenames = [node["name"] for node in config["storagenodes"]]
         reloadableplopconfig.append((Symbol("storage_nodes"), storagenodeaddresses))
         reloadableplopconfig.append((Symbol("storage_nodes_quorum"), config["storage-quorum-size"]))
         services.add(Symbol("ht"))
-        allowed_clients += allowed_clients_mergesecondary(primarymergenode)
+        if nodename == primarymergenodename:
+            merge = localconfig["merge"]
+            plopconfig.append((Symbol("db_backend_opt"), [(Symbol("write_flag"), Symbol("read"))]))
+            plopconfig.append((Symbol("merge_delay"), merge["min-delay"]))
+            plopconfig.append((Symbol("merge_dist_winsize"), merge["dist-window-size"]))
+            plopconfig.append((Symbol("merge_dist_sendlog_chunksize"), merge["dist-sendlog-chunksize"]))
+            plopconfig.append((Symbol("merge_dist_sendentries_chunksize"), merge["dist-sendentries-chunksize"]))
+            plopconfig.append((Symbol("frontend_nodes"), frontendnodeaddresses))
+            plopconfig.append((Symbol("sth_path"), paths["mergedb"] + "/sth"))
+            allowed_servers += allowed_servers_primarymerge(frontendnodenames)
+        else:
+            allowed_clients += allowed_clients_mergesecondary(primarymergenodename)
 
     plopconfig += [
         (Symbol("services"), list(services)),
@@ -383,12 +409,13 @@ def gen_config(nodename, config, localconfig):
 
 def gen_testmakefile(config, testmakefile, machines, shellvars=False):
     configfile = open(testmakefile, "w")
+    print >>configfile, "#", testmakefile, "generated by", sys.argv[0], datetime.now()
+
     frontendnodenames = set([node["name"] for node in config["frontendnodes"]])
     storagenodenames = set([node["name"] for node in config["storagenodes"]])
     signingnodenames = set([node["name"] for node in config["signingnodes"]])
     mergenodenames = set([node["name"] for node in config["mergenodes"]])
-    erlangnodenames = frontendnodenames | storagenodenames | signingnodenames | \
-      set(filter(lambda name: name != config["primarymergenode"], mergenodenames))
+    erlangnodenames_and_apps = ['%s:%s' % (nn, 'catlfish' if nn != config["primarymergenode"] else "merge") for nn in frontendnodenames | storagenodenames | signingnodenames | mergenodenames]
 
     frontendnodeaddresses = [node["publicaddress"] for node in config["frontendnodes"]]
     storagenodeaddresses = [node["address"] for node in config["storagenodes"]]
@@ -398,7 +425,7 @@ def gen_testmakefile(config, testmakefile, machines, shellvars=False):
     delimiter = '"' if shellvars else ''
     
     print >>configfile, "NODES=" + delimiter + " ".join(frontendnodenames|storagenodenames|signingnodenames|mergenodenames) + delimiter
-    print >>configfile, "ERLANGNODES=" + delimiter + " ".join(erlangnodenames)  + delimiter
+    print >>configfile, "ERLANGNODES=" + delimiter + " ".join(erlangnodenames_and_apps) + delimiter
     print >>configfile, "MACHINES=" + delimiter + " ".join([str(e) for e in range(1, machines+1)]) + delimiter
     print >>configfile, "TESTURLS=" + delimiter + " ".join(frontendnodeaddresses+storagenodeaddresses+signingnodeaddresses+mergenodeaddresses) + delimiter
     print >>configfile, "BASEURL=" + delimiter + config["baseurl"] + delimiter