8b7d2c8a93
This patch is the first one of a series of patches improving how the L3 agents update the router HA state to the Neutron server. This patch partially reverts the previous patch [1]. When the batch notifier sends events, it calls the callback method passed during the initialization, in this case AgentMixin.notify_server. The batch notifier spawns a new thread in charge of sending the notifications and then wait the specified "batch_interval" time. If the callback method is not synchronous with the notify thread execution (what [1] implemented), the thread can finish while the RPC client is still sending the HA router states. If another HA state update is received, then both updates can be executed at the same time. It is possible then that a new router state can be overwritten with an old one still not sent or processed. The batch notifier is refactored, to improve what initally was implemented [2] and then updated [3]. Currently, each new event thread can update the "pending_events" list. Then, a new thread is spawned to process this event list. This thread decouples the current execution from the calling thread, making the event processing a non-blocking process. But with the current implementation, each new process will spawn a new thread, synchronized with the previous and new ones (using a synchronized decorator). That means, during the batch interval time, the system can have as many threads waiting as new events received. Those threads will end secuentially when the previous threads end the batch interval sleep time. Instead of this, this patch receives and enqueue each new event and allows only one thread to be alive while processing the event list. If at the end of the processing loop new events are stored, the thread will process then. [1] I3f555a0c78fbc02d8214f12b62c37d140bc71da1 [2] I2f8cf261f48bdb632ac0bd643a337290b5297fce [3] I82f403441564955345f47877151e0c457712dd2f Partial-Bug: #1837635 Change-Id: I20cfa1cf5281198079f5e0dbf195755abc919581
78 lines
3.1 KiB
Python
78 lines
3.1 KiB
Python
# Copyright (c) 2014 OpenStack Foundation.
|
|
# All Rights Reserved.
|
|
#
|
|
# 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 eventlet
|
|
import mock
|
|
|
|
from neutron.common import utils
|
|
from neutron.notifiers import batch_notifier
|
|
from neutron.tests import base
|
|
|
|
|
|
class TestBatchNotifier(base.BaseTestCase):
|
|
def setUp(self):
|
|
super(TestBatchNotifier, self).setUp()
|
|
self._received_events = eventlet.Queue()
|
|
self.notifier = batch_notifier.BatchNotifier(2, self._queue_events)
|
|
self.spawn_n_p = mock.patch.object(eventlet, 'spawn_n')
|
|
|
|
def _queue_events(self, events):
|
|
for event in events:
|
|
self._received_events.put(event)
|
|
|
|
def test_queue_event_no_event(self):
|
|
spawn_n = self.spawn_n_p.start()
|
|
self.notifier.queue_event(None)
|
|
self.assertEqual(0, len(self.notifier._pending_events.queue))
|
|
self.assertEqual(0, spawn_n.call_count)
|
|
|
|
def test_queue_event_first_event(self):
|
|
spawn_n = self.spawn_n_p.start()
|
|
self.notifier.queue_event(mock.Mock())
|
|
self.assertEqual(1, len(self.notifier._pending_events.queue))
|
|
self.assertEqual(1, spawn_n.call_count)
|
|
|
|
def test_queue_event_multiple_events_notify_method(self):
|
|
def _batch_notifier_dequeue():
|
|
while not self.notifier._pending_events.empty():
|
|
self.notifier._pending_events.get()
|
|
|
|
c_mock = mock.patch.object(self.notifier, '_notify',
|
|
side_effect=_batch_notifier_dequeue).start()
|
|
events = 20
|
|
for i in range(events):
|
|
self.notifier.queue_event('Event %s' % i)
|
|
eventlet.sleep(0) # yield to let coro execute
|
|
|
|
utils.wait_until_true(self.notifier._pending_events.empty,
|
|
timeout=5)
|
|
# Called twice: when the first thread calls "synced_send" and then,
|
|
# in the same loop, when self._pending_events is not empty(). All
|
|
# self.notifier.queue_event calls are done in just one
|
|
# "batch_interval" (2 secs).
|
|
self.assertEqual(2, c_mock.call_count)
|
|
|
|
def test_queue_event_multiple_events_callback_method(self):
|
|
events = 20
|
|
for i in range(events):
|
|
self.notifier.queue_event('Event %s' % i)
|
|
eventlet.sleep(0) # yield to let coro execute
|
|
|
|
utils.wait_until_true(self.notifier._pending_events.empty,
|
|
timeout=5)
|
|
expected = ['Event %s' % i for i in range(events)]
|
|
# Check the events have been handled in the same input order.
|
|
self.assertEqual(expected, list(self._received_events.queue))
|