Daemon (#20)

* missed a log statement

* add daemon code (back) in, simplify

* start isn't user facing, update bools to reflect

* update manual scripts to make sure daemon is disabled

* init timer to None

* switch from bpython to IPython to avoid greenlet exception

Author: jacklowrie

pyproject.toml

@@ -77,7 +77,7 @@ explicit = true
 
 [dependency-groups]
 dev = [
-    "bpython>=0.25",
+    "ipython>=8.37.0",
     "mypy>=1.17.1",
     "pdoc>=15.0.4",
     "pre-commit>=4.3.0",

scripts/auto/anchor.py

@@ -1 +1,31 @@
 """anchor.py: creates and starts an anchor node for testing."""
+import sys
+
+import IPython
+
+from chordnet import Node as ChordNode
+
+
+def main() -> None:
+    """Creates a new ring with this computer as the only node."""
+    if len(sys.argv) != 3:
+        print("usage: [uv run] python anchor.py ip_addr port_no")
+        exit(1)
+
+    ip = sys.argv[1]
+    port = int(sys.argv[2])
+
+    node = ChordNode(ip, port, debug=True)
+    # create (start) the ring
+    node.create()
+    print(f"Node created as \"node\": {node.address}", file=sys.stderr)
+    repl_locals = {
+        'node': node,
+    }
+    print("starting repl. access `node`")
+    IPython.embed(user_ns=repl_locals)
+    node.stop()
+
+
+if __name__ == '__main__':
+    main()

scripts/auto/joiner.py

@@ -1 +1,34 @@
 """joiner.py: creates a joining node for debugging."""
+import sys
+
+import IPython
+
+from chordnet import Node as ChordNode
+
+
+def main() -> None:
+    """Creates a new ring with this computer as the only node."""
+    if len(sys.argv) != 5:
+        print("usage: [uv run] python " \
+              "joiner.py this_ip this_port target_ip target_port")
+        exit(1)
+
+    # Get IP and port from command line arguments
+    ip = sys.argv[1]
+    port = int(sys.argv[2])
+    target_ip = sys.argv[3]
+    target_port = int(sys.argv[4])
+
+    # Create and join node
+    node = ChordNode(ip, port, debug=True)
+    node.join(target_ip, target_port)
+    repl_locals = {
+        'node': node,
+    }
+    print("starting repl. access `node`, advance with `step(node)`")
+    IPython.embed(user_ns=repl_locals)
+    node.stop()
+
+
+if __name__ == '__main__':
+    main()

scripts/manual/anchor.py

@@ -1,7 +1,7 @@
 """anchor.py: creates and starts an anchor node for testing."""
 import sys
 
-import bpython
+import IPython
 from step import step  #type: ignore
 
 from chordnet import Node as ChordNode
@@ -16,7 +16,7 @@ def main() -> None:
     ip = sys.argv[1]
     port = int(sys.argv[2])
 
-    node = ChordNode(ip, port)
+    node = ChordNode(ip, port, daemon=False)
     # create (start) the ring
     node.create()
     print(f"Node created as \"node\": {node.address}", file=sys.stderr)
@@ -25,7 +25,7 @@ def main() -> None:
         'step': step,
     }
     print("starting repl. access `node`, advance with `step(node)`")
-    bpython.embed(locals_=repl_locals)
+    IPython.embed(user_ns=repl_locals)
     node.stop()
 
 

scripts/manual/joiner.py

@@ -1,7 +1,7 @@
 """joiner.py: creates a joining node for debugging."""
 import sys
 
-import bpython
+import IPython
 from step import step  #type: ignore
 
 from chordnet import Node as ChordNode
@@ -21,14 +21,14 @@ def main() -> None:
     target_port = int(sys.argv[4])
 
     # Create and join node
-    node = ChordNode(ip, port)
+    node = ChordNode(ip, port, daemon=False)
     node.join(target_ip, target_port)
     repl_locals = {
         'node': node,
         'step': step,
     }
     print("starting repl. access `node`, advance with `step(node)`")
-    bpython.embed(locals_=repl_locals)
+    IPython.embed(user_ns=repl_locals)
     node.stop()
 
 

src/chordnet/net.py

@@ -162,7 +162,6 @@ def _handle_connection(self, client_socket: socket.socket) -> None:
             # Send response
             client_socket.send(str(response).encode())
         except Exception as e:
-            sys.stderr.write(f"Error handling connection: {e}\n")
-            sys.stderr.flush()
+            log.error(f"Error handling connection: {e}")
         finally:
             client_socket.close()

src/chordnet/node.py

@@ -1,4 +1,5 @@
 """node.py: Represents a node on a ring."""
+import threading
 from typing import Callable, Tuple
 
 from loguru import logger as log
@@ -33,14 +34,27 @@ class Node:
     finger_table: list[Address | None]
     _next: int
     _net: _Net
+    _timer: threading.Timer | None
     is_running: bool
-
-    def __init__(self, ip: str, port: int) -> None:
+    _use_daemon: bool
+    _interval: float
+    _debug: bool
+
+    def __init__(self,
+                 ip: str,
+                 port: int,
+                 daemon:bool=True,
+                 interval:float=1.0,
+                 debug:bool=False
+    ) -> None:
         """Initializes a new Chord node.
 
         Args:
-            ip (str): IP address for the node.
-            port (int): Port number to listen on.
+            ip: IP address for the node.
+            port: Port number to listen on.
+            daemon: whether to run the daemon.
+            interval: daemon interval.
+            debug: whether to print node state after every daemon run.
         """
         self.address = Address(ip, port)
 
@@ -53,6 +67,11 @@ def __init__(self, ip: str, port: int) -> None:
         self._net = _Net(ip, port, self._process_request)
         self.is_running = False
 
+        self._use_daemon = daemon
+        self._interval = interval
+        self._debug = debug
+        self._timer = None
+
     def successor(self) -> Address | None:
         """Alias for self.finger_table[0]."""
         return self.finger_table[0]
@@ -131,43 +150,42 @@ def fix_fingers(self) -> None:
 
         # Move to the next finger table entry, wrapping around if necessary
         self._next = (self._next + 1) % Address._M
-    '''
-    def _run_fix_fingers(self, interval=1.0):
-        """
-        Periodically invokes fix_fingers every `interval` seconds.
+
+    def _daemon(self) -> None:
+        """Runs fix_fingers and stabilize periodically.
 
         Args:
-            interval (float): Time interval between updates (in seconds).
+            interval: Time interval between periodic calls.
+            debug: Whether to print node state
         """
-        self.fix_fingers()
-        # Schedule the next execution
-        self._fix_fingers_timer = threading.Timer(
-                        interval, self._run_fix_fingers, args=[interval])
-        self._fix_fingers_timer.start()
+        if self._use_daemon and self.is_running:
+            try:
+                self.stabilize()
+                self.fix_fingers()
+                if self._debug:
+                    print(f"pred: {self.predecessor}, succ: {self.successor()}")
+                    print(self.finger_table)
+
+            except Exception as e:
+                # Catch any unhandled exception within the daemon's tasks
+                # and log it properly. This is crucial for debugging.
+                log.error(
+                    f"Unhandled exception in daemon for {self.address}: {e}"
+                )
+                # You might want to optionally stop the daemon here
+                # if continuous failures are problematic
+                # self.stop()
+
+            finally:
+                # Always reschedule the timer, even if an exception occurred,
+                # unless you decided to stop the daemon above.
+                if self.is_running:
+                    self._timer = threading.Timer(self._interval, self._daemon)
+                    self._timer.daemon = True
+                    self._timer.start()
 
-    def start_periodic_tasks(self, interval=1.0):
-        """
-        Starts periodic tasks for the node, including fix_fingers.
 
-        Args:
-            interval (float): Time interval between periodic calls.
-        """
-        if self._fix_fingers_timer and self._fix_fingers_timer.is_alive():
-            # Timer is already running, no need to start again
-            log.info("Periodic tasks are already running.")
-            return
-        self.is_running = True
-        self._run_fix_fingers(interval)
 
-    def stop_periodic_tasks(self):
-        """
-        Stops periodic tasks for the node gracefully.
-        """
-        if self._fix_fingers_timer:
-            self._fix_fingers_timer.cancel()
-            self._fix_fingers_timer = None
-        self.is_running = False
-    '''
     def log_finger_table(self) -> None:
         """Logs the entire finger table to the log file."""
         message = "Current Finger Table:\n"
@@ -338,20 +356,35 @@ def notify(self, potential_successor: Address | None)-> bool:
 
 
     def start(self) -> None:
-        """Starts the Chord node's network listener.
+        """Starts the Chord node's daemon and network listener.
 
         Begins accepting incoming network connections in a separate thread.
+
+        Args:
+            daemon: whether to run a daemon
+                    (runs fix_fingers and stabilize periodically
+            interval: interval daemon sleeps for (only relevant if daemon=True)
+            debug: whether to print the state of the node
+                    (again only relevant to daemon)
         """
         self._net.start()
+        self.is_running = True
+        if self._use_daemon:
+            self._daemon()
 
 
 
     def stop(self) -> None:
-        """Gracefully stops the Chord node's network listener.
+        """Gracefully stops the Chord node's daemon and network listener.
 
         Closes the server socket and waits for the network thread to terminate.
         """
         self._net.stop()
+        if self._timer:
+            self._timer.cancel()
+            self._timer = None
+        self.is_running = False
+