Author: lajoskatona Date: Fri, 27 Mar 2026 10:02:09 +0100 Description: Add native LoopingCall to BGP Bug: https://launchpad.net/bugs/2146526 Change-Id: I86d967174f9e6e0e2882045f2a6f00880837b1bb Signed-off-by: lajoskatona Assisted-By: claude-sonnet-4 Origin: upstream, https://review.opendev.org/c/openstack/os-ken/+/982380 Last-Update: 2026-03-27 diff --git a/os_ken/services/protocols/bgp/base.py b/os_ken/services/protocols/bgp/base.py index 5b2f35f..3e26629 100644 --- a/os_ken/services/protocols/bgp/base.py +++ b/os_ken/services/protocols/bgp/base.py @@ -42,7 +42,7 @@ from os_ken.lib.packet.bgp import RF_L2VPN_FLOWSPEC from os_ken.lib.packet.bgp import RF_RTC_UC from os_ken.services.protocols.bgp.utils.circlist import CircularListType -from os_ken.services.protocols.bgp.utils.evtlet import LoopingCall +from os_ken.services.protocols.bgp import utils # Logger instance for this module. @@ -225,7 +225,7 @@ return greenthread def _create_timer(self, name, func, *arg, **kwarg): - timer = LoopingCall(func, *arg, **kwarg) + timer = utils.LoopingCall(func, *arg, **kwarg) self._timers[name] = timer return timer @@ -456,8 +456,8 @@ self.index = Sink.next_index() # Create an event for signal enqueuing. - from .utils.evtlet import EventletIOFactory - self.outgoing_msg_event = EventletIOFactory.create_custom_event() + from os_ken.services.protocols.bgp import utils + self.outgoing_msg_event = utils.IOFactory.create_custom_event() self.messages_queued = 0 # List of msgs. that are to be sent to this peer. Each item diff --git a/os_ken/services/protocols/bgp/peer.py b/os_ken/services/protocols/bgp/peer.py index b686295..99cb660 100644 --- a/os_ken/services/protocols/bgp/peer.py +++ b/os_ken/services/protocols/bgp/peer.py @@ -43,7 +43,7 @@ from os_ken.services.protocols.bgp.info_base.vpnv6 import Vpnv6Path from os_ken.services.protocols.bgp.rtconf.vrfs import VRF_RF_IPV4, VRF_RF_IPV6 from os_ken.services.protocols.bgp.utils import bgp as bgp_utils -from os_ken.services.protocols.bgp.utils.evtlet import EventletIOFactory +from os_ken.services.protocols.bgp import utils from os_ken.services.protocols.bgp.utils import stats from os_ken.services.protocols.bgp.utils.validation import is_valid_old_asn @@ -345,7 +345,7 @@ # Setting this event starts the connect_loop loop again # Clearing this event will stop the connect_loop loop - self._connect_retry_event = EventletIOFactory.create_custom_event() + self._connect_retry_event = utils.IOFactory.create_custom_event() # Reference to threads related to enhanced refresh timers. self._refresh_stalepath_timer = None diff --git a/os_ken/services/protocols/bgp/processor.py b/os_ken/services/protocols/bgp/processor.py index 9ca4f3f..d1347d3 100644 --- a/os_ken/services/protocols/bgp/processor.py +++ b/os_ken/services/protocols/bgp/processor.py @@ -23,7 +23,7 @@ from os_ken.services.protocols.bgp.base import BGP_PROCESSOR_ERROR_CODE from os_ken.services.protocols.bgp.base import BGPSException from os_ken.services.protocols.bgp.utils import circlist -from os_ken.services.protocols.bgp.utils.evtlet import EventletIOFactory +from os_ken.services.protocols.bgp import utils from os_ken.lib.packet.bgp import RF_RTC_UC from os_ken.lib.packet.bgp import BGP_ATTR_TYPE_AS_PATH @@ -82,7 +82,7 @@ self._core_service = core_service self._dest_queue = BgpProcessor._DestQueue() self._rtdest_queue = BgpProcessor._DestQueue() - self.dest_que_evt = EventletIOFactory.create_custom_event() + self.dest_que_evt = utils.IOFactory.create_custom_event() self.work_units_per_cycle =\ work_units_per_cycle or BgpProcessor.MAX_DEST_PROCESSED_PER_CYCLE diff --git a/os_ken/services/protocols/bgp/utils/__init__.py b/os_ken/services/protocols/bgp/utils/__init__.py index e69de29..fb5546d 100644 --- a/os_ken/services/protocols/bgp/utils/__init__.py +++ b/os_ken/services/protocols/bgp/utils/__init__.py @@ -0,0 +1,31 @@ +# Copyright 2026 Openinfra Foundation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +# implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from os_ken.lib import hub + + +LoopingCall = None +IOFactory = None + +if hub.HUB_TYPE == 'eventlet': + from os_ken.services.protocols.bgp.utils import evtlet + LoopingCall = evtlet.LoopingCall + IOFactory = evtlet.EventletIOFactory + + +if hub.HUB_TYPE == 'native': + from os_ken.services.protocols.bgp.utils import thrding + LoopingCall = thrding.LoopingCall + IOFactory = thrding.ThreadingIOFactory diff --git a/os_ken/services/protocols/bgp/utils/thrding.py b/os_ken/services/protocols/bgp/utils/thrding.py new file mode 100644 index 0000000..2b04c7b --- /dev/null +++ b/os_ken/services/protocols/bgp/utils/thrding.py @@ -0,0 +1,76 @@ +# Copyright 2026 Openinfra Foundation +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or +# implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import threading +import logging + +from os_ken.lib import hub + + +LOG = logging.getLogger('utils.thrding') + + +class ThreadingIOFactory(object): + + @staticmethod + def create_custom_event(): + LOG.debug('Create CustomEvent called') + return hub.Event() + + @staticmethod + def create_looping_call(funct, *args, **kwargs): + LOG.debug('create_looping_call called') + return LoopingCall(funct, *args, **kwargs) + + +class LoopingCall(object): + def __init__(self, funct, *args, **kwargs): + self._funct = funct + self._args = args + self._kwargs = kwargs + self._running = False + self._interval = 0 + self._cancel_event = threading.Event() + self._thread = None + + def _loop(self): + while self._running: + # Wait for interval, but can be interrupted by cancel_event + if self._cancel_event.wait(self._interval): + # Event was set => reset or stop requested + self._cancel_event.clear() + if not self._running: + break + continue + if self._running: + self._funct(*self._args, **self._kwargs) + + def start(self, interval, now=True): + if self._running: + self.stop() + self._running = True + self._interval = interval + self._cancel_event.clear() + self._thread = threading.Thread(target=self._loop, daemon=True) + self._thread.start() + if now: + self._funct(*self._args, **self._kwargs) + + def stop(self): + self._running = False + self._cancel_event.set() + + def reset(self): + self._cancel_event.set()