fix: improve graceful shutdown and timed execution handling.

* fix: improve graceful shutdown and timed execution handling.

* Removed unused exception and improved exception management for some cases by avoiding blind

* fix: improve error logging in inbound and outbound connection threads

* fix: update string formatting in stop_program function for consistency

* fix: increase timeout for queue retrieval and improve stop method for graceful exit

---------

Co-authored-by: Andrea Lacava <a.lacava@northeastern.edu>

Author: absentio

examples/spectrum_dapp.py

@@ -4,7 +4,7 @@
 """
 
 import argparse
-import multiprocessing
+import threading
 import time
 
 from e3interface.e3_connector import E3LinkLayer, E3TransportLayer
@@ -14,12 +14,12 @@
 
 def stop_program(time_to_wait, dapp: SpectrumSharingDApp):
     time.sleep(time_to_wait)
-    print("Stop is called")
-    dapp.stop()
+    print(f"[INFO] Timer elapsed after {time_to_wait} seconds")
+    dapp.stop_event.set()
     time.sleep(0.5) # to allow proper closure of the dApp threads, irrelevant to profiling
-    print("Test completed")
+    print("[INFO] Stopping of the dApp completed")
 
-def main(args, time_to_wait: float = 60.0):
+def main(args):
     # with open(f"{LOG_DIR}/busy.txt", "w") as f:
     #     f.close()
 
@@ -70,24 +70,28 @@ def main(args, time_to_wait: float = 60.0):
     response, ranFunctionList = dapp.setup_connection()   
 
     if not response:
-        raise ValueError("RAN refused Setup")
+        raise ValueError("[WARNING] RAN refused Setup")
 
-    print(f"Setup Complete - RAN function available: {ranFunctionList}")
+    print(f"[INFO] Setup Complete - RAN function available: {ranFunctionList}")
 
     # atm we subscribe to all
     dapp.send_subscription_request(ranFunctionList)
 
     if args.timed:
-        timer = multiprocessing.Process(target=stop_program, args=(time_to_wait, dapp))
+        timer = threading.Thread(target=stop_program, args=(args.timed, dapp), daemon=False)
         timer.start()
     else:
         timer = None
 
     try:
         dapp.control_loop()
     finally:
-        if args.timed:
-            timer.kill()
+        dapp.stop()
+        if args.timed and timer is not None:
+            if timer.is_alive():
+                timer.join(timeout=2)
+                if timer.is_alive():
+                    print("[ERROR] Timer thread did not terminate in time")
 
 
 if __name__ == "__main__":
@@ -105,7 +109,7 @@ def main(args, time_to_wait: float = 60.0):
     parser.add_argument('--num-subcarrier-spacing', type=int, default=30, help="Subcarrier spacing in kHz (FR1 is 30)")
     parser.add_argument('--e', action='store_true', default=False, help="Set if 3/4 sampling for FFT size is set on the gNB (-E option on OAI)")
     parser.add_argument('--center-freq', type=float, default=3.6192e9, help="Center frequency in Hz")
-    parser.add_argument('--timed', action='store_true', default=False, help="Run with a 5-minute time limit")
+    parser.add_argument('--timed', type=int, default=0, metavar='SECONDS', help="Run with a time limit (in seconds). 0 means no limit.")
     parser.add_argument('--model', type=str, default='', help="Path to the CNN model file to be used")
     parser.add_argument('--time-window', type=int, default=5, help="Number of input vectors to pass to the CNN model.")
     parser.add_argument('--moving-avg-window', type=int, default=30, help="Window size (in samples) for the moving average used to detect energy peaks in the spectrum.")

src/e3interface/e3_connector.py

@@ -125,13 +125,19 @@ def send_setup_request(self, payload):
         while request_retries > 0:
             setup_socket = self.setup_context.socket(zmq.REQ)
             setup_socket.connect(self.setup_endpoint)
-            e3_logger.debug("Send E3 Setup request")
-            setup_socket.send(payload)
-
-            if (setup_socket.poll(request_timeout) & zmq.POLLIN) != 0:
-                reply = setup_socket.recv()
-                e3_logger.debug('ZMQ setup socket replied')
-                return reply
+            try:
+                e3_logger.debug("Send E3 Setup request")
+                setup_socket.send(payload)
+                if (setup_socket.poll(request_timeout) & zmq.POLLIN) != 0:
+                    reply = setup_socket.recv()
+                    e3_logger.debug('ZMQ setup socket replied')
+                    setup_socket.close()
+                    return reply
+            except KeyboardInterrupt:
+                e3_logger.debug("Keyboard interrupt, closing E3 Setup Socket")
+                setup_socket.setsockopt(zmq.LINGER, 0)
+                setup_socket.close()
+                raise
 
             request_retries -= 1
             e3_logger.error("ZMQ setup did not reply")

src/e3interface/e3_interface.py

@@ -149,9 +149,11 @@ def _inbound_connection(self):
 
                     case _:
                         raise ValueError("Unrecognized PDU type ", pdu[0])
-                    
-        except Exception as e:
-            e3_logger.error(f"Error in inbound thread: {e}")
+        except KeyboardInterrupt:
+            e3_logger.debug("Inbound thread received SIGINT, stopping")
+            self.stop_event.set()
+        except Exception:
+            e3_logger.exception(f"Error in inbound thread")
             self.stop_event.set()
         finally:
             e3_logger.info("Close inbound connection")
@@ -188,9 +190,11 @@ def _outbound_connection(self):
 
                 e3_logger.debug(f"Send the pdu encoded {payload}")
                 self.e3_connector.send(payload)
-
-        except Exception as e:
-            e3_logger.error(f"Error outbound thread: {e}")
+        except KeyboardInterrupt:
+            e3_logger.debug("Outbound thread received SIGINT, stopping")
+            self.stop_event.set()
+        except Exception:
+            e3_logger.exception(f"Error in outbound thread")
             self.stop_event.set()
         finally:
             e3_logger.info("Close outbound connection")
@@ -259,10 +263,19 @@ def terminate_connections(self):
         e3_logger.info("Stop event")
         self.stop_event.set()
 
-        if hasattr(self, "inbound_process"):
-            self.inbound_process.join()
-        if hasattr(self, "outbound_process"):
-            self.outbound_process.join()
+        if hasattr(self, "inbound_process") and self.inbound_process.is_alive():
+            self.inbound_process.join(timeout=2)
+            if self.inbound_process.is_alive():
+                e3_logger.warning("Inbound process did not terminate gracefully, forcing termination")
+                self.inbound_process.terminate()
+                self.inbound_process.join(timeout=1)
+                
+        if hasattr(self, "outbound_process") and self.outbound_process.is_alive():
+            self.outbound_process.join(timeout=2)
+            if self.outbound_process.is_alive():
+                e3_logger.warning("Outbound process did not terminate gracefully, forcing termination")
+                self.outbound_process.terminate()
+                self.outbound_process.join(timeout=1)
 
         self.e3_connector.dispose()
 

src/spectrum/spectrum_dapp.py

@@ -6,6 +6,7 @@
 __author__ = "Andrea Lacava"
 
 import multiprocessing
+import queue
 import time
 import os
 import numpy as np
@@ -299,8 +300,8 @@ def get_iqs_from_ran(self, dapp_identifier, data):
                         if update_sampling:
                             self.sampling_threshold = new_sampling_threshold
                             dapp_logger.info(f"Custom logic updated sampling threshold to {self.sampling_threshold}")
-                    except Exception as e:
-                        dapp_logger.error(f"Error in custom control callback: {e}")
+                    except Exception:
+                        dapp_logger.exception(f"Error in custom control callback")
                         update_sampling = False
 
                 prb_new = prb_blk_list.view(prb_blk_list.dtype.newbyteorder('>'))
@@ -324,17 +325,27 @@ def get_iqs_from_ran(self, dapp_identifier, data):
                 self.control_count = 1  
 
     def _control_loop(self):
-        if self.energyGui:
-            abs_iq_av_db = self.sig_queue.get()
-            self.energyPlotter.process_iq_data(abs_iq_av_db)
-
-        if self.iqPlotterGui:
-            iq_data = self.iq_queue.get()
-            self.iqPlotter.process_iq_data(iq_data)
-
-        if self.dashboard:
-            message = self.demo_queue.get()
-            self.demo.process_iq_data(message)
+        # If no GUIs are enabled, just sleep to avoid busy-waiting
+        if not (self.energyGui or self.iqPlotterGui or self.dashboard):
+            time.sleep(1)
+            return
+
+        try:
+            if self.energyGui:    
+                abs_iq_av_db = self.sig_queue.get(timeout=0.1)
+                self.energyPlotter.process_iq_data(abs_iq_av_db)
+
+            if self.iqPlotterGui:
+                iq_data = self.iq_queue.get(timeout=0.1)
+                self.iqPlotter.process_iq_data(iq_data)
+    
+            if self.dashboard:
+                message = self.demo_queue.get(timeout=0.1)
+                self.demo.process_iq_data(message)
+        except queue.Empty:
+            pass # This is allowed
+        except Exception:
+            dapp_logger.exception("[SPECTRUM] Error in the control loop")
 
     def _stop(self):        
         if self.save_iqs:

src/visualization/dashboard.py

@@ -47,10 +47,10 @@ def index(self):
         return render_template("index.html")
 
     def run(self):
-        self.socketio.run(self.app, host="0.0.0.0", port=7778)
+        self.socketio.run(self.app, host="0.0.0.0", port=7778, debug=False, use_reloader=False)
 
     def _initialize_plot(self):
-        self.run_thread = threading.Thread(target=self.run)
+        self.run_thread = threading.Thread(target=self.run, daemon=True)
         self.run_thread.start()
 
     def handle_initial_connection(self):
@@ -97,8 +97,10 @@ def process_iq_data(self, message):
 
     def stop(self):
         """Stops the server and kills the thread."""
+        # This is probably overkill since now the run thread it's a demon,
+        # still with timeout it does not impact the graceful exit
         if self.run_thread and self.run_thread.is_alive():
-            self.run_thread.join()
+            self.run_thread.join(timeout=1)
 
 if __name__ == "__main__":
     server_app = Dashboard()