Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix issue with asyncio.Queue #2445

Merged
merged 3 commits into from
Nov 18, 2024
Merged

Fix issue with asyncio.Queue #2445

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
31a2cb9
Fix issue with asyncio.Queue which is not thread safe.
grossmj Nov 17, 2024
b37db57
Fix tests
grossmj Nov 17, 2024
fa0d7d7
Use call_soon_threadsafe() in notification manager
grossmj Nov 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions gns3server/compute/notification_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.


import asyncio
from contextlib import contextmanager
from gns3server.utils.notification_queue import NotificationQueue

Expand Down Expand Up @@ -54,7 +54,7 @@ def emit(self, action, event, **kwargs):
"""

for listener in self._listeners:
listener.put_nowait((action, event, kwargs))
asyncio.get_event_loop().call_soon_threadsafe(listener.put_nowait, (action, event, kwargs))

@staticmethod
def reset():
Expand Down
8 changes: 4 additions & 4 deletions gns3server/controller/notification.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

import os
import asyncio
from contextlib import contextmanager

from gns3server.utils.notification_queue import NotificationQueue
Expand Down Expand Up @@ -73,7 +73,7 @@ def controller_emit(self, action, event):
"""

for controller_listener in self._controller_listeners:
controller_listener.put_nowait((action, event, {}))
asyncio.get_event_loop().call_soon_threadsafe(controller_listener.put_nowait, (action, event, {}))

def project_has_listeners(self, project_id):
"""
Expand Down Expand Up @@ -134,7 +134,7 @@ def _send_event_to_project(self, project_id, action, event):
except KeyError:
return
for listener in project_listeners:
listener.put_nowait((action, event, {}))
asyncio.get_event_loop().call_soon_threadsafe(listener.put_nowait, (action, event, {}))

def _send_event_to_all_projects(self, action, event):
"""
Expand All @@ -146,4 +146,4 @@ def _send_event_to_all_projects(self, action, event):
"""
for project_listeners in self._project_listeners.values():
for listener in project_listeners:
listener.put_nowait((action, event, {}))
asyncio.get_event_loop().call_soon_threadsafe(listener.put_nowait, (action, event, {}))
4 changes: 2 additions & 2 deletions tests/compute/qemu/test_qemu_vm.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -202,11 +202,11 @@ async def test_termination_callback_error(vm, tmpdir):

await queue.get(1) # Ping

(action, event, kwargs) = queue.get_nowait()
(action, event, kwargs) = await queue.get(1)
assert action == "node.updated"
assert event == vm

(action, event, kwargs) = queue.get_nowait()
(action, event, kwargs) = await queue.get(1)
assert action == "log.error"
assert event["message"] == "QEMU process has stopped, return code: 1\nBOOMM"

Expand Down