Merge pull request #32 from logos-co/WSL-Update

WSL Update

README.md

@@ -44,8 +44,21 @@ These are arguments that can be modified:
 - _topology_file_: string. Default: **waku_test_topology_small.json**. If defines the network topology that will be created.
 - _simulation_time_: int. Default: **300**. Specifies the simulation time in seconds.
 - _message_rate_: int. Default: **25**. Specifies the message rate in packets per second.
-- _min_packet_size_: int. Default: **1**. Specifies the minimum size of the packet in bytes.
-- _min_packet_size_: int. Default: **1024**. Specifies the maximum size of the packet in bytes.
+- _min_packet_size_: int. Default: **1**. Specifies the minimum size of the packet in bytes. Must be an even number (Waku constrain).
+- _min_packet_size_: int. Default: **1024**. Specifies the maximum size of the packet in bytes. Must be an even number (Waku constrain).
+- _dist_type_: int. Default: **uniform**. Specifies the size distribution of the messages being injected into the network. Options are: **gaussian** and **uniform**
+- _emitters_fraction_: int. Default: **0.5**. Specifies the fraction of nodes that will be injecting traffic.
+- _inter_msg_type_: int. Default: **poisson**. Specifies the inter-message times. Options are: **poisson** and **uniform**
+
+dist_type : "gaussian"
+
+    # Fraction (of the total number of nodes) that inject traffic
+    # Values: [0., 1.]
+    emitters_fraction : 0.5
+
+    # Inter-message times
+    # Values: uniform and gaussian
+    inter_msg_type : "uniform"
 
 #### What will happen
 

config/config.json

@@ -1,8 +1,11 @@
 {
     "same_toml_configuration": true,
     "topology_file": "waku_test_topology_small.json",
-    "simulation_time": 1,
-    "message_rate": 50,
-    "min_packet_size": 8,
-    "max_packet_size": 8
+    "simulation_time": 60,
+    "message_rate": 10,
+    "min_packet_size": 2,
+    "max_packet_size": 1024,
+    "inter_msg_type": "poisson",
+    "dist_type": "gaussian",
+    "emitters_fraction": 1.0
 }

main.star

@@ -38,4 +38,4 @@ def run(args):
     # waku.get_waku_peers(waku_topology.keys()[1])
 
     # Setup WSL & Start the Simulation
-    wsl_service = wsl.set_up_wsl(waku_services,  config['simulation_time'], config['message_rate'], config['min_packet_size'], config['max_packet_size'])
+    wsl_service = wsl.set_up_wsl(waku_services,  config['simulation_time'], config['message_rate'], config['min_packet_size'], config['max_packet_size'], config['inter_msg_type'], config['dist_type'], config['emitters_fraction'])

src/wsl.star

@@ -4,9 +4,10 @@ system_variables = import_module("github.com/logos-co/wakurtosis/src/system_vari
 # Module Imports
 files = import_module(system_variables.FILE_HELPERS_MODULE)
 
-def create_wsl_config(simulation_time=300, message_rate=50, min_packet_size=1, max_packet_size=1024, dist_type='uniform', emitters_fraction=0.5):
+def create_wsl_config(simulation_time=300, message_rate=50, min_packet_size=1, max_packet_size=1024, inter_msg_type='uniform', dist_type='uniform', emitters_fraction=0.5):
     
-    template_data = {"simulation_time": simulation_time, "message_rate" : message_rate, "min_packet_size" : min_packet_size, "max_packet_size" : max_packet_size, "dist_type" : dist_type, "emitters_fraction" : emitters_fraction}
+    template_data = {"simulation_time": simulation_time, "message_rate" : message_rate, "min_packet_size" : min_packet_size, 
+                    "max_packet_size" : max_packet_size, "dist_type" : dist_type, "emitters_fraction" : emitters_fraction, "inter_msg_type" : inter_msg_type}
 
     # Traffic simulation parameters
     wsl_yml_template = """
@@ -35,6 +36,10 @@ def create_wsl_config(simulation_time=300, message_rate=50, min_packet_size=1, m
             # Fraction (of the total number of nodes) that inject traffic
             # Values: [0., 1.]
             emitters_fraction : {{.emitters_fraction}}
+
+            # Inter-message times
+            # Values: uniform and poisson
+            inter_msg_type : {{.inter_msg_type}}
     """
 
     artifact_id = render_templates(
@@ -69,10 +74,10 @@ def create_wsl_targets(services):
 
     return artifact_id
 
-def set_up_wsl(services, simulation_time, message_rate, min_packet_size, max_packet_size):
+def set_up_wsl(services, simulation_time, message_rate, min_packet_size, max_packet_size, inter_msg_type, dist_type, emitters_fraction):
     
     # Generate simulation config
-    wsl_config = create_wsl_config(simulation_time, message_rate, min_packet_size, max_packet_size)
+    wsl_config = create_wsl_config(simulation_time, message_rate, min_packet_size, max_packet_size, inter_msg_type, dist_type, emitters_fraction)
 
     # Create targets.json
     wsl_targets = create_wsl_targets(services)
@@ -92,6 +97,6 @@ def set_up_wsl(services, simulation_time, message_rate, min_packet_size, max_pac
         )
     )
 
-    print('kurtosis service logs -f wakurtosis SERVICE-GUID')
+    print('kurtosis service logs wakurtosis SERVICE-GUID')
     
     return wsl_service

wsl-module/build.sh

@@ -1,2 +1,3 @@
 #!/bin/sh
+# pip freeze > requirements.txt
 docker image build --progress=plain  -t wsl:0.0.1 ./

wsl-module/config/wsl.yml

@@ -23,4 +23,8 @@ general:
     # Fraction (of the total number of nodes) that inject traffic
     # Values: [0., 1.]
     emitters_fraction : 0.5
+
+    # Inter-message times
+    # Values: uniform and poisson
+    inter_msg_type : "uniform"
     

wsl-module/dockerfile

@@ -2,5 +2,6 @@ FROM python:3.11.0
 LABEL Maintainer="Daimakaimura"
 WORKDIR /wsl
 COPY wsl.py .
+COPY rtnorm.py .
 COPY requirements.txt ./
 RUN pip install -r requirements.txt

wsl-module/requirements.txt

@@ -16,6 +16,7 @@ pyparsing==3.0.9
 python-dateutil==2.8.2
 PyYAML==6.0
 requests==2.28.1
+scipy==1.10.0
 six==1.16.0
 tqdm==4.64.1
 urllib3==1.26.13

wsl-module/wsl.py

@@ -97,28 +97,25 @@ def send_waku_msg(node_address, topic, payload, nonce=1):
 def poisson_interval(rate):
     return random.expovariate(rate)
 
-def make_payload(size, rnd=True):
-
-    # Size in bytes (2 hex digits per byte)
-    if rnd:
-        payload = ''.join(random.choices('0123456789abcdef', k=int( 2 * size - 1)))
-    else:
-        payload = ''.join('00' * int(size))
-
-    payload = '0x%s' %payload 
-
+def make_payload(size):
+    payload = hex(random.getrandbits(4*size))     
     G_LOGGER.debug('Payload of size %d bytes: %s' %(size, payload))
-
     return payload
 
 def make_payload_dist(dist_type, min_size, max_size):
 
+    # Check if min and max packet sizes are the same
+    if min_size == max_size:
+        G_LOGGER.warning('Packet size is constant: min_size=max_size=%d' %min_size)
+        return make_payload(min_size)
+
     # Payload sizes are even integers uniformly distributed in [min_size, max_size] 
     if dist_type == 'uniform':
-        size = random.uniform(min_size, max_size)
-        # Make sure we only sample even sizes
+        size = int(random.uniform(min_size, max_size))
+        
+        # Reject non even sizes
         while(size % 2) != 0:
-            size = random.uniform(min_size, max_size)
+            size = int(random.uniform(min_size, max_size))
             
         return make_payload(size)
 
@@ -129,10 +126,10 @@ def make_payload_dist(dist_type, min_size, max_size):
         size = int(rtnorm.rtnorm(min_size, max_size, sigma=σ, mu=μ, size=1))
         
         # Reject non even sizes
-        while size % 2 != 0.0:
+        while(size % 2) != 0:
             size = int(rtnorm.rtnorm(min_size, max_size, sigma=σ, mu=μ, size=1))
 
-        return size
+        return make_payload(size)
 
     G_LOGGER.error('Unknown distribution type %s')
 
@@ -156,7 +153,18 @@ def parse_targets(enclave_dump_path, waku_port=8545):
 
     return targets
 
-def main():
+def get_next_time_to_msg(inter_msg_type, msg_rate, simulation_time):
+    
+    if inter_msg_type == 'poisson':
+        return poisson_interval(msg_rate) 
+    
+    if inter_msg_type == 'uniform':
+        return simulation_time / msg_rate
+        
+    G_LOGGER.error('%s is not a valid inter_msg_type. Aborting.' %inter_msg_type)
+    sys.exit()
+
+def main(): 
 
     global G_LOGGER
     
@@ -223,7 +231,7 @@ def main():
 
     emitters = random.sample(targets, num_emitters)
     G_LOGGER.info('Selected %d emitters out of %d total nodes' %(len(emitters), len(targets)))
-    
+
     """ Start simulation """
     stats = {}
     msg_cnt = 0
@@ -239,7 +247,7 @@ def main():
         # Check end condition
         elapsed_s = time.time() - s_time
         if  elapsed_s >= config['general']['simulation_time']:
-            G_LOGGER.info('Simulation ended. Sent %d messages (%d bytes) in %d at an avg. bandwitdth of %d Bps' %(msg_cnt, bytes_cnt, elapsed_s, bytes_cnt / elapsed_s))
+            G_LOGGER.info('Simulation ended. Sent %d messages (%d bytes) in %ds.' %(msg_cnt, bytes_cnt, elapsed_s))
             break
 
         # Send message
@@ -247,6 +255,8 @@ def main():
         msg_elapsed = time.time() - last_msg_time
         if msg_elapsed <= next_time_to_msg:
             continue
+
+        G_LOGGER.debug('Time Δ: %.6f ms.' %((msg_elapsed - next_time_to_msg) * 1000.0))
         
         # Reference: https://rfc.vac.dev/spec/16/#get_waku_v2_relay_v1_messages
         node_address = 'http://%s/' %random.choice(emitters)
@@ -268,12 +278,13 @@ def main():
         # else:
         #     G_LOGGER.error('RPC Message failed to node_address')
 
-        # Sampling inter-message times from a Poisson distribution)
-        next_time_to_msg = poisson_interval(config['general']['msg_rate'])
+        # Compute the time to next message
+        next_time_to_msg = get_next_time_to_msg(config['general']['inter_msg_type'], config['general']['msg_rate'], config['general']['simulation_time']) 
+        G_LOGGER.debug('Next message will happen in %d ms.' %(next_time_to_msg * 1000.0))
+        
         last_msg_time = time.time()
         
-        msg_cnt += 1
-        bytes_cnt += len(payload) / 2 - 2
+        msg_cnt += 1 
         
     # Pull messages 
     # get_waku_v2_relay_v1_messagesget_waku_v2_relay_v1_messages