Drop eol python

.github/workflows/tests-codecheck.yml

@@ -28,7 +28,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [3.8, 3.9, '3.10']
+        python-version: ['3.10', 3.11, 3.12, 3.13, 3.14]
     steps:
       - name: Checkout repo
         uses: actions/checkout@v5

.github/workflows/tests-unit.yml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [3.8, 3.9, '3.10', 3.11]
+        python-version: ['3.10', 3.11, 3.12, 3.13, 3.14]
     steps:
       - name: Checkout repo
         uses: actions/checkout@v5

beka/peering.py

@@ -1,23 +1,23 @@
+import threading
 import time
 
-from eventlet import sleep, GreenPool
-from eventlet.queue import Queue
-import eventlet.greenthread as greenthread
-
 from .chopper import Chopper
 from .event import EventTimerExpired, EventMessageReceived
 from .bgp_message import BgpMessageParser, BgpMessagePacker
 from .error import SocketClosedError, IdleError
 
+# Sentinel placed on output queues during shutdown to wake any thread
+# blocked in ``Queue.get()`` so it can observe ``_stop_event`` and exit.
+_QUEUE_POISON = object()
+
 
 class Peering(object):
     def __init__(
         self, state_machine, peer_address, socket, route_handler, error_handler=None
     ):
         self.input_stream = None
         self.chopper = None
-        self.pool = None
-        self.eventlets = None
+        self.threads = None
         self.parser = None
         self.packer = None
         self.state_machine = state_machine
@@ -27,37 +27,42 @@ def __init__(
         self.route_handler = route_handler
         self.error_handler = error_handler
         self.start_time = int(time.time())
+        self._stop_event = threading.Event()
 
     def uptime(self):
         return int(time.time()) - self.start_time
 
     def run(self):
         self.input_stream = self.socket.makefile(mode="rb")
         self.chopper = Chopper(self.input_stream)
-        self.pool = GreenPool()
         self.parser = BgpMessageParser()
         self.packer = BgpMessagePacker()
         self.state_machine.open_handler = self.open_handler
-        self.eventlets = []
-
-        self.eventlets.append(self.pool.spawn(self.send_messages))
-        self.eventlets.append(self.pool.spawn(self.print_route_updates))
-        self.eventlets.append(self.pool.spawn(self.kick_timers))
-        self.eventlets.append(self.pool.spawn(self.receive_messages))
 
-        self.pool.waitall()
+        targets = (
+            self.send_messages,
+            self.print_route_updates,
+            self.kick_timers,
+            self.receive_messages,
+        )
+        self.threads = [
+            threading.Thread(target=t, name=t.__name__, daemon=True) for t in targets
+        ]
+        for thread in self.threads:
+            thread.start()
+        for thread in self.threads:
+            thread.join()
 
     def open_handler(self, capabilities):
         self.parser.capabilities = capabilities
         self.packer.capabilities = capabilities
 
     def receive_messages(self):
-        while True:
-            sleep(0)
+        while not self._stop_event.is_set():
             try:
                 message_type, serialised_message = self.chopper.next()
             except SocketClosedError as e:
-                if self.error_handler:
+                if self.error_handler and not self._stop_event.is_set():
                     self.error_handler("Peering %s: %s" % (self.peer_address, e))
                 self.shutdown()
                 break
@@ -73,25 +78,30 @@ def receive_messages(self):
                 break
 
     def send_messages(self):
-        while True:
-            sleep(0)
+        while not self._stop_event.is_set():
             message = self.state_machine.output_messages.get()
+            if message is _QUEUE_POISON:
+                break
             self.socket.send(self.packer.pack(message))
 
     def empty_message_queue(self):
         while self.state_machine.output_messages.qsize():
             message = self.state_machine.output_messages.get()
+            if message is _QUEUE_POISON:
+                continue
             self.socket.send(self.packer.pack(message))
 
     def print_route_updates(self):
-        while True:
-            sleep(0)
+        while not self._stop_event.is_set():
             route_update = self.state_machine.route_updates.get()
+            if route_update is _QUEUE_POISON:
+                break
             self.route_handler(route_update)
 
     def kick_timers(self):
-        while True:
-            sleep(1)
+        # ``Event.wait(timeout)`` returns True as soon as the event is set,
+        # giving prompt shutdown without burning CPU between ticks.
+        while not self._stop_event.wait(timeout=1):
             tick = int(time.time())
             try:
                 self.state_machine.event(EventTimerExpired(), tick)
@@ -102,6 +112,16 @@ def kick_timers(self):
                 break
 
     def shutdown(self):
+        if self._stop_event.is_set():
+            return
+        self._stop_event.set()
         self.empty_message_queue()
-        for eventlet in self.eventlets:
-            eventlet.kill()
+        # Unblock any thread parked in ``Queue.get()``. Each consumer
+        # checks ``_stop_event`` after waking and exits.
+        self.state_machine.output_messages.put(_QUEUE_POISON)
+        self.state_machine.route_updates.put(_QUEUE_POISON)
+        # Force ``chopper.next()``'s underlying recv to return.
+        try:
+            self.socket.shutdown(2)  # socket.SHUT_RDWR
+        except OSError:
+            pass

beka/state_machine.py

@@ -1,4 +1,4 @@
-from eventlet.queue import Queue
+from queue import Queue
 from collections import OrderedDict
 
 from .event import Event

beka/stream_server.py

@@ -1,16 +1,20 @@
 import socket
-
-from eventlet import GreenPool, listen
-import eventlet.greenthread as greenthread
+import threading
+from concurrent.futures import ThreadPoolExecutor
 
 
 class StreamServer:
+    """Listen on ``address`` and dispatch each accepted socket to ``handler`` in its own thread."""
+
+    DEFAULT_MAX_HANDLERS = 64
+
     def __init__(self, address, handler):
         self.address = address
         self.handler = handler
-        self.greenlets = set()
         self.running = False
         self.server = None
+        self._executor = None
+        self._accept_thread = None
 
     def _family(self):
         if ":" in self.address[0]:
@@ -19,24 +23,35 @@ def _family(self):
 
     def serve_forever(self):
         self.running = True
-        self.server = listen(self.address, self._family())
-        pool = GreenPool()
-
+        self.server = socket.socket(self._family(), socket.SOCK_STREAM)
+        self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self.server.bind(self.address)
+        self.server.listen()
+        self._executor = ThreadPoolExecutor(
+            max_workers=self.DEFAULT_MAX_HANDLERS,
+            thread_name_prefix="beka-handler",
+        )
+        self._accept_thread = threading.current_thread()
         try:
             while self.running:
-                sock, address = self.server.accept()
-                pool.spawn(self.call_handler, sock, address)
-                self.greenlets
-        except OSError:
-            pass
-
-    def call_handler(self, sock, address):
-        self.greenlets.add(greenthread.getcurrent())
-        self.handler(sock, address)
-        self.greenlets.remove(greenthread.getcurrent())
+                try:
+                    sock, address = self.server.accept()
+                except OSError:
+                    # accept() raises after stop() shuts the listening socket
+                    break
+                self._executor.submit(self.handler, sock, address)
+        finally:
+            self._executor.shutdown(wait=False)
+            self._executor = None
 
     def stop(self):
         self.running = False
-        for greenlet in self.greenlets:
-            greenlet.kill()
-        self.server.shutdown(socket.SHUT_RDWR)
+        if self.server is not None:
+            try:
+                self.server.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+            try:
+                self.server.close()
+            except OSError:
+                pass

requirements.txt

@@ -1,2 +1 @@
-eventlet>=0.33.3
 pbr>=1.9

run.py

@@ -1,9 +1,8 @@
 import signal
 import sys
+import threading
 import yaml
 
-from eventlet import GreenPool
-
 from beka.beka import Beka
 
 
@@ -20,7 +19,7 @@ def __init__(self):
 
     def run(self):
         signal.signal(signal.SIGINT, self.signal_handler)
-        pool = GreenPool()
+        threads = []
 
         with open("beka.yaml", encoding="utf-8") as file:
             config = yaml.safe_load(file.read())
@@ -46,9 +45,14 @@ def run(self):
                 for route in router["routes"]:
                     beka.add_route(route["prefix"], route["next_hop"])
             self.bekas.append(beka)
-            pool.spawn_n(beka.run)
-        pool.waitall()
-        printmsg("All greenlets gone, exiting")
+            thread = threading.Thread(
+                target=beka.run, name="beka-%s" % router["local_address"]
+            )
+            thread.start()
+            threads.append(thread)
+        for thread in threads:
+            thread.join()
+        printmsg("All threads gone, exiting")
 
     def signal_handler(self, _signal, _frame):
         printmsg("[SIGINT] Shutting down")

setup.cfg

@@ -1,20 +1,17 @@
 [metadata]
-name = beka
+name = c65beka
 summary = A bare-bones BGP speaker
 long_description =
     Beka is a fairly basic BGP speaker. It can send
     and receive unicast route updates in IPv4 and IPv6,
     but not too much else. It is designed to be simple to use
     and to extend, without too much overhead.
 
-    It uses eventlet for concurrency, but is easy enough to port to
-    gevent if that takes your fancy.
-
-    More information at https://github.com/faucetsdn/beka
+    More information at https://github.com/c65sdn/beka
 license = Apache-2
 author = Sam Russell
 author-email = sam.h.russell@gmail.com
-home-page = https://github.com/faucetsdn/beka
+home-page = https://github.com/c65sdn/beka
 classifiers =
     Development Status :: 2 - Pre-Alpha
     Intended Audience :: System Administrators

setup.py

@@ -6,11 +6,11 @@
 
 from setuptools import setup
 
-if sys.version_info < (3,):
+if sys.version_info < (3, 10):
     print(
         """You are trying to install beka on python {py}
 
-beka is not compatible with python 2, please upgrade to python 3.8 or newer.""".format(
+beka is not compatible with python earlier than 3.10, please upgrade.""".format(
             py=".".join([str(v) for v in sys.version_info[:3]])
         ),
         file=sys.stderr,
@@ -20,6 +20,6 @@
 setup(
     name="beka",
     setup_requires=["pbr>=1.9", "setuptools>=17.1"],
-    python_requires=">=3.8",
+    python_requires=">=3.10",
     pbr=True,
 )