1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
|
Author: lajoskatona <lajos.katona@est.tech>
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 <lajos.katona@est.tech>
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()
|
