Gennet + Starlark: specifying inter-subnetwork QoS requirements

config/config.json

@@ -13,15 +13,21 @@
     "fanout": 13,
     "num_topics": 1,
     "num_partitions": 1,
-    "num_subnets": 1,
+    "num_subnets": 7,
     "container_size": "1",
     "node_type_distribution": {
       "nwaku:relay:rpc:metrics": 100,
       "gowaku:rln:dnsdisc:dns": 0
     },
     "network_type": "newmanwattsstrogatz",
     "output_dir": "network_data",
-    "benchmark": "False"
+    "benchmark": "False",
+    "inter_subnet_qos_distribution": {
+      "None": 60,
+      "Block": 10,
+      "80:Uniform:10": 10,
+      "2.0:Normal:5000:30:10": 20
+    }
   },
   "wls": {
     "debug_level": "DEBUG",

monitoring/dstats/dstats.sh

@@ -51,7 +51,7 @@ docker exec $wait_cid touch /wls/start.signal
 
 echo "dstats: waiting for WLS to finish : dstats $dstats_pid is running"
 docker container wait $wait_cid
-sleep 10        # make sure you collect the stats until last messages settle down
+sleep 30        # collect as much stats possible as the simulation is winding down
 
 echo "dstats: WLS finished: stopping the docker monitor $dstats_pid"
 kill -15 $dstats_pid

monitoring/host-proc/host-proc-helper.sh

@@ -35,7 +35,7 @@ procfs_pid=$!
 
 echo "host-proc: waiting for WLS to finish: procfs $procfs_pid running"
 docker container wait $WAIT_CID   # now wait for the wakurtosis to finish
-sleep 60        # make sure you collect the stats until last messages settle down
+sleep 30        # collect as much stats possible as the simulation is winding down
 
 echo "host-proc: stopping the helper $procfs_pid"
 kill -15  $procfs_pid   # procfs-stats is a su process

run.sh

@@ -50,7 +50,7 @@ kurtosis_inspect="kurtosis_inspect.log"
 usr=`id -u`
 grp=`id -g`
 stats_dir=stats
-signal_fifo=/tmp/hostproc-signal.fifo   # do not create fifo under ./stats, or inside the repo
+signal_fifo=/tmp/host-proc-signal.fifo   # do not create fifo under ./stats, or inside the repo
 ##################### MONITORING MODULE PROLOGUES
 if [ "$metrics_infra" = "cadvisor" ]; then #CADVISOR
     # prepare the enclave
@@ -89,7 +89,8 @@ fi
 jobs=$(cat config/${wakurtosis_config_file} | jq -r ".kurtosis.jobs")
 echo -e "\nSetting up the enclave: $enclave_name"
 
-kurtosis_cmd="kurtosis --cli-log-level \"$loglevel\" run --full-uuids --enclave ${enclave_name} . '{\"wakurtosis_config_file\" : \"config/${wakurtosis_config_file}\"}' --parallelism ${jobs} > $kurtosis_run 2>&1"
+#kurtosis_cmd="kurtosis --cli-log-level \"$loglevel\" run --full-uuids --enclave ${enclave_name} . '{\"wakurtosis_config_file\" : \"config/${wakurtosis_config_file}\"}' --parallelism ${jobs} > $kurtosis_run 2>&1"
+kurtosis_cmd="kurtosis --cli-log-level \"$loglevel\" run --full-uuids --enclave ${enclave_name} . '{\"wakurtosis_config_file\" : \"config/${wakurtosis_config_file}\"}' --with-subnetworks --parallelism ${jobs} > $kurtosis_run 2>&1"
 
 START=$(date +%s)
   eval $kurtosis_cmd
@@ -165,8 +166,8 @@ echo "Simulation took $DIFF1 + $DIFF2 = $(( $END2 - $START)) secs"
 
 
 ##################### GATHER CONFIG, LOGS & METRICS
-# give time for the messages to settle down before we collect the logs
-sleep 60
+# give time for the logs and metrics collection to settle down
+sleep 45 # 30 + 15
 
 # dump logs
 echo "Dumping Kurtosis logs"
@@ -182,6 +183,7 @@ if [ "$metrics_infra" = "dstats" ]; then
     echo "dstats: copying the dstats data"
     cp -r ./monitoring/dstats/stats  ${enclave_name}_logs/dstats-data
 elif [ "$metrics_infra" = "host-proc" ]; then
+    rm -f $signal_fifo
     echo "Copying the host-proc data"
     cp -r ./monitoring/host-proc/stats  ${enclave_name}_logs/host-proc-data
 elif [ "$metrics_infra" = "container-proc" ]; then
@@ -206,13 +208,13 @@ if jq -e ."plotting" >/dev/null 2>&1 "./config/${wakurtosis_config_file}"; then
         docker run --name "dstats" --network "host" -v "$(pwd)/wakurtosis_logs:/simulation_data/" --add-host=host.docker.internal:host-gateway analysis src/hproc.py dstats /simulation_data/ --config-file /simulation_data/config/config.json  >/dev/null 2>&1  
         docker cp dstats:/analysis/plots/ wakurtosis_logs/dstats-plots
         cd wakurtosis_logs
-        ln -s dstats-plots/output-dstats-compare.pdf analysis.pdf
+        cp dstats-plots/output-dstats-compare.pdf analysis.pdf
         cd ..
     elif [ "$metrics_infra" = "host-proc" ]; then
         docker run --name "host-proc" --network "host" -v "$(pwd)/wakurtosis_logs:/simulation_data/" --add-host=host.docker.internal:host-gateway analysis src/hproc.py host-proc /simulation_data/ --config-file /simulation_data/config/config.json  >/dev/null 2>&1
         docker cp host-proc:/analysis/plots/ wakurtosis_logs/host-proc-plots
         cd wakurtosis_logs
-        ln -s host-proc-plots/output-host-proc-compare.pdf analysis.pdf
+        cp host-proc-plots/output-host-proc-compare.pdf analysis.pdf
         cd ..
     elif [ "$metrics_infra" = "container-proc" ]; then
         docker run --network "host" -v "$(pwd)/wakurtosis_logs:/simulation_data/" --add-host=host.docker.internal:host-gateway analysis src/main.py -i container-proc >/dev/null 2>&1

src/node_builders/dispatchers.star

@@ -38,4 +38,9 @@ ports_dispatcher = {
     vars.GENNET_GOWAKU_IMAGE_VALUE: waku_builder.add_waku_ports_info_to_topology,
     vars.GENNET_NWAKU_IMAGE_VALUE: waku_builder.add_waku_ports_info_to_topology,
     vars.GENNET_NOMOS_IMAGE_VALUE: nomos_builder.add_nomos_ports_info_to_topology
-}
+}
+
+#QoS_distribution_dispatcher = {
+#    vars.GENNET_SUBNETS_QOS_UNIFORM: UniformPacketDelayDistribution,
+#    vars.GENNET_SUBNETS_QOS_NORMAL: NormalPacketDelayDistribution
+#}

src/node_builders/node_builders.star

@@ -50,6 +50,15 @@ def instantiate_services(plan, network_topology, testing):
         configs=all_services_configuration
     )
 
+    # Deploy the subnetworks
+    if network_topology[vars.GENNET_SUBNETS_KEY]:
+      plan.print("Greetings")
+      for src in network_topology[vars.GENNET_SUBNETS_KEY].keys():
+          for dst in network_topology[vars.GENNET_SUBNETS_KEY][src].keys():
+              QoS = network_topology[vars.GENNET_SUBNETS_KEY][dst][src]
+              connection_config = get_connection_config(QoS, plan)
+              plan.print(src + "-" + dst + " = " + str(connection_config))
+              plan.set_connection(subnetworks = (src, dst), config = connection_config)
     _add_service_info_to_topology(plan, all_services_information, network_topology)
 
 
@@ -94,4 +103,37 @@ def _add_service_info_to_topology(plan, all_services_information, network_topolo
         ip = network_topology[vars.GENNET_NODES_KEY][nodes[0]][vars.IP_KEY]
         network_topology[vars.GENNET_ALL_CONTAINERS_KEY][container_id] = {}
         network_topology[vars.GENNET_ALL_CONTAINERS_KEY][container_id][vars.GENNET_NODES_KEY] = nodes
-        network_topology[vars.GENNET_ALL_CONTAINERS_KEY][container_id][vars.KURTOSIS_IP_KEY] = ip
+        network_topology[vars.GENNET_ALL_CONTAINERS_KEY][container_id][vars.KURTOSIS_IP_KEY] = ip
+
+
+def get_connection_config(QoS, plan):
+    QoS_lst = QoS.split(":")
+    n = len(QoS_lst)
+    if n == 1:
+      if QoS_lst[0] == "None":
+         return kurtosis.connection.ALLOWED
+      elif QoS_lst[0]  == "Block":
+          return kurtosis.connection.BLOCKED
+      else:
+          plan.print("Invalid QoS atom \"" + QoS_lst[0] + "\"")
+          plan.exit()
+    elif n == 3:
+        packet_loss_perc, dist, delay = float(QoS_lst[0]), QoS_lst[1], int(QoS_lst[2])
+        if dist == "Uniform":
+            return ConnectionConfig(packet_loss_perc, UniformPacketDelayDistribution(ms=delay))
+        else:
+            plan.print("Invalid QoS atom \"" + dist + "\"")
+            plan.exit()
+    elif n == 5:
+        packet_loss_perc, dist, mean = float(QoS_lst[0]), QoS_lst[1], int(QoS_lst[2])
+        jitter, corr = int(QoS_lst[3]), float(QoS_lst[4])
+        if dist == "Normal":
+            distri = NormalPacketDelayDistribution(
+                        mean_ms=mean, std_dev_ms=jitter, correlation=corr)
+            return ConnectionConfig(packet_loss_perc, distri)
+        else:
+            plan.print("Invalid QoS atom \"" + dist + "\"")
+            plan.exit()
+    else:
+        plan.print("Invalid QoS spec \"" + QoS + "\"")
+        plan.exit()

src/node_builders/types/waku_builder.star

@@ -47,4 +47,4 @@ def _add_waku_port(network_topology, all_services_information, node_id, node_inf
         (all_services_information[node_info[vars.GENNET_NODE_CONTAINER_KEY]].ports[
              port_id].number,
          all_services_information[node_info[vars.GENNET_NODE_CONTAINER_KEY]].ports[
-             port_id].transport_protocol)
+             port_id].transport_protocol)

src/system_variables.star

@@ -86,6 +86,10 @@ GENNET_STATIC_NODES_KEY = "static_nodes"
 GENNET_GOWAKU_IMAGE_VALUE = "go-waku"
 GENNET_NWAKU_IMAGE_VALUE = "nim-waku"
 GENNET_NOMOS_IMAGE_VALUE = "nomos"
+GENNET_SUBNETS_KEY = "subnetworks"
+
+GENNET_SUBNETS_QOS_UNIFORM = "Uniform"
+GENNET_SUBNETS_QOS_NORMAL = "Normal"
 
 PEER_ID_KEY = "peer_id"
 IP_KEY = "ip_address"