Fix multi-threading issues in routes object and add test case

Author: bitterpanda63

aikido_zen/background_process/routes/__init__.py

@@ -41,18 +41,19 @@ def increment_route(self, route_metadata):
             self.initialize_route(route_metadata)
         # Add hits to
         route = self.routes.get(key)
-        route["hits"] += 1
-        route["hits_delta_since_sync"] += 1
+        if route:
+            route["hits"] += 1
+            route["hits_delta_since_sync"] += 1
 
     def update_route_with_apispec(self, route_metadata, apispec):
         """
         Updates apispec of a given route (or creates it).
         route_metadata object includes route, url and method
         """
         key = route_to_key(route_metadata)
-        if not self.routes.get(key):
-            return
-        update_route_info(apispec, self.routes[key])
+        route = self.routes.get(key)
+        if route:
+            update_route_info(apispec, route)
 
     def get(self, route_metadata):
         """Gets you the route entry if it exists using route metadata"""
@@ -61,8 +62,8 @@ def get(self, route_metadata):
 
     def get_routes_with_hits(self):
         """Gets you all routes with a positive hits delta"""
-        result = dict()
-        for key, route in self.routes.items():
+        result = {}
+        for key, route in list(self.routes.items()):
             if route["hits_delta_since_sync"] <= 0:
                 continue  # do not add routes without a hit delta
             result[key] = route
@@ -78,11 +79,13 @@ def manage_routes_size(self):
         """
         if len(self.routes) >= self.max_size:
             least_used = [None, float("inf")]
-            for key, route in self.routes.items():
+            for key, route in list(self.routes.items()):
                 if route.get("hits") < least_used[1]:
                     least_used = [key, route.get("hits")]
             if least_used[0]:
-                del self.routes[least_used[0]]
+                delete_route = self.routes[least_used[0]]
+                if delete_route:
+                    del delete_route
 
     def __iter__(self):
         return iter(self.routes.values())

aikido_zen/background_process/routes/routes_lock_test.py

@@ -0,0 +1,85 @@
+import os
+import random
+import sys
+import threading
+import time
+
+from aikido_zen.background_process.routes import Routes
+from aikido_zen.helpers.logging import logger
+
+
+def add_routes(routes, num_routes, error_count):
+    """Function to add routes to the Routes object."""
+    try:
+        for _ in range(num_routes):
+            route_metadata = {
+                "method": random.choice(["GET", "POST", "PUT", "DELETE"]),
+                "route": f"/api/test/{random.randint(1, 10000)}",
+            }
+            routes.increment_route(route_metadata)
+    except Exception:
+        error_count["errors_add"] += 1
+
+
+def export_routes(routes, error_count):
+    """Function to export routes with hits."""
+    while True:
+        try:
+            routes.get_routes_with_hits()
+        except Exception:
+            error_count["errors_export"] += 1
+
+
+def runtime_error_without_data_lock():
+    """Test function to check for runtime errors without data locking."""
+    routes = Routes(max_size=5000)
+
+    # Create threads for adding routes and exporting routes
+    num_add_threads = 30
+    num_routes_per_thread = 3000
+    num_export_threads = 20
+    error_count = {
+        "errors_add": 0,
+        "errors_export": 0,
+    }
+
+    add_threads = []
+    for i in range(num_add_threads):
+        print(f"Adding thread {i} started")
+        thread = threading.Thread(
+            target=add_routes,
+            args=(routes, num_routes_per_thread, error_count),
+            daemon=True,
+        )
+        add_threads.append(thread)
+        thread.start()
+
+    # Start the export thread
+    export_threads = []
+    for i in range(num_export_threads):
+        print(f"Export thread {i} started")
+        thread = threading.Thread(
+            target=export_routes, args=(routes, error_count), daemon=True
+        )
+        export_threads.append(thread)
+        thread.start()
+
+    # Wait for all add threads to finish
+    time.sleep(1)
+
+    for thread in add_threads:
+        print("Closing add thread: " + thread.name)
+        thread.join(timeout=20 / 1000)
+
+    # Stop the export thread after adding routes
+    for thread in export_threads:
+        print("Closing export thread: " + thread.name)
+        thread.join(timeout=20 / 1000)
+
+    print(error_count)
+    assert error_count["errors_add"] == 0
+    assert error_count["errors_export"] == 0
+
+
+if __name__ == "__main__":
+    runtime_error_without_data_lock()