Merge pull request #35 from scivisum/bug/RD-41528_dont_wait_for_messages

Bug/rd 41528 dont wait for messages

Author: RodolfoSGomez

browserdebuggertools/wssessionmanager.py

@@ -4,7 +4,7 @@
 import socket
 import time
 import collections
-from threading import Thread, Event, Lock
+from threading import Thread, Lock
 
 from typing import Dict
 
@@ -22,43 +22,20 @@
 )
 
 
-class NotifiableDeque(collections.deque):
-    """ A Queue with the benefits of deque speed
-        It can wait until there are new messages or the timeout is met
-    """
-    _POLL_INTERVAL = 1
-
-    def __init__(self):
-        super(NotifiableDeque, self).__init__()
-        self._poll_signal = Event()
-
-    def append(self, message):
-        """ Appends to the queue and allows any waiting threads to start popping from it
-        """
-        super(NotifiableDeque, self).append(message)
-        self._poll_signal.set()
-
-    def wait_for_messages(self):
-        """ Waits until there are messages or the poll interval time
-        """
-        self._poll_signal.wait(self._POLL_INTERVAL)
-        if self._poll_signal.is_set():
-            self._poll_signal.clear()
-
-
 class _WSMessageProducer(Thread):
     """ Interfaces with the websocket to send messages from the send queue
         or put messages from the websocket into recv queue
     """
     _CONN_TIMEOUT = 15
     _BLOCKED_TIMEOUT = 5
+    _POLL_INTERVAL = 0.01  # How long to wait for new ws messages
 
     def __init__(self, port, send_queue, on_message):
         super(_WSMessageProducer, self).__init__()
         self._port = port
         self._send_queue = send_queue
         self._on_message = on_message
-        self._last_poll = None
+        self._last_ws_attempt = None
         self._continue = True
 
         self.exception = None
@@ -141,14 +118,12 @@ def run(self):
 
         with self._ws_io():
 
-            self._last_poll = time.time()
             while self._continue:
-
+                self._last_ws_attempt = time.time()
                 self._empty_send_queue()
                 self._empty_websocket()
 
-                self._send_queue.wait_for_messages()
-                self._last_poll = time.time()
+                time.sleep(self._POLL_INTERVAL)
 
     @property
     def blocked(self):
@@ -161,12 +136,13 @@ def blocked(self):
             some messages could allow us to reduce the load on the websocket
             so raising an exception in this case allows us to empty the send queue and try again.
 
-            **  This could be solved by having a separate thread to handle sending messages,
-                assuming ws.send() doesn't hang and is also thread safe.
-                Then we could update self._last_poll after every successful ws.send()/ws.recv()
+            **  This could be solved by not handling sending messages in the thread,
+                assuming ws.send() doesn't hang and is atomic.
+                Then we could update self._last_ws_attempt after every successful ws send()/recv()
         """
         return (
-            self._last_poll and ((time.time() - self._last_poll) > self._BLOCKED_TIMEOUT)
+            self._last_ws_attempt
+            and ((time.time() - self._last_ws_attempt) > self._BLOCKED_TIMEOUT)
         )
 
     def health_check(self):
@@ -237,7 +213,7 @@ def __init__(self, port, timeout, domains=None):
         self._message_producer_lock = Lock()  # Lock making sure we don't create 2 ws connections
         self._last_not_ok = None
         self._message_producer_not_ok_count = 0
-        self._send_queue = NotifiableDeque()
+        self._send_queue = collections.deque()
 
         self.port = port
         self._message_producer = None

tests/integrationtests/test_wssessionmanager.py

@@ -9,7 +9,7 @@
 
 from browserdebuggertools.exceptions import DevToolsTimeoutException, MaxRetriesException
 from browserdebuggertools.wssessionmanager import (
-    WSSessionManager, _WSMessageProducer, NotifiableDeque
+    WSSessionManager, _WSMessageProducer
 )
 
 MODULE_PATH = "browserdebuggertools.WSSessionManager."
@@ -92,7 +92,6 @@ def test_locked_get_events(self):
         with patch.object(
             _WSMessageProducer, "_get_websocket", new=MagicMock(return_value=self.ws)
         ):
-            NotifiableDeque._MAX_QUEUE_BUFFER = 9999
             self.session_manager = WSSessionManager(1234, 1, {"Network": {}})
 
             events = list(reversed(self.session_manager.get_events("Network", clear=True)))
@@ -159,6 +158,12 @@ def unblock(self):
         self._continue = False
 
 
+class TimeoutBlockingWS(BlockingWS):
+
+    def send(self, data):
+        pass
+
+
 class ExceptionThrowingWS(_DummyWebsocket):
 
     exceptions = 0
@@ -229,14 +234,14 @@ def test_thread_blocked_twice(self):
     def test_thread_blocks_causes_timeout(self):
 
         with patch.object(_WSMessageProducer, "_get_websocket",
-                          new=MagicMock(return_value=BlockingWS(times_to_block=1))):
+                          new=MagicMock(return_value=TimeoutBlockingWS())):
 
             self.session_manager = WSSessionManager(1234, 3)
             self.resetWS()
             with self.assertRaises(DevToolsTimeoutException):
                 start = time.time()
                 self.session_manager.execute("Network", "enable")
-                self.assertLess(time.time() - start, 5)
+            self.assertLess(time.time() - start, 5)
 
     def test_max_thread_blocks_exceeded(self):
 

tests/unittests/test_wssessionmanager.py

@@ -1,6 +1,6 @@
+import collections
 import copy
 import socket
-from threading import Event
 from unittest import TestCase
 
 from mock import patch, MagicMock, call, PropertyMock
@@ -13,7 +13,7 @@
     MaxRetriesException
 )
 from browserdebuggertools.wssessionmanager import (
-    WSSessionManager, _WSMessageProducer, NotifiableDeque
+    WSSessionManager, _WSMessageProducer
 )
 
 MODULE_PATH = "browserdebuggertools.wssessionmanager."
@@ -31,7 +31,7 @@ def _get_websocket(self):
             return MagicMock()
 
     def setUp(self):
-        self.send_queue = NotifiableDeque()
+        self.send_queue = collections.deque()
         self.messaging_thread = self.MockWSMessageProducer(1111, self.send_queue, MagicMock())
         self.ws_message_producer = self.messaging_thread
 
@@ -179,7 +179,7 @@ def test_fail(self):
 class Test__WSMessageProducer_run(WSMessageProducerTest):
 
     def prepare(self, time):
-        NotifiableDeque._POLL_INTERVAL = 0
+        _WSMessageProducer._POLL_INTERVAL = 0
         self.next_time = 0
 
         def increment_time():
@@ -197,7 +197,7 @@ def test(self, time):
 
         self.ws_message_producer.run()
 
-        self.assertEqual(10, self.ws_message_producer._last_poll)
+        self.assertEqual(10, self.ws_message_producer._last_ws_attempt)
 
     def test_exception(self, time):
         exception = Exception()
@@ -206,15 +206,15 @@ def test_exception(self, time):
 
         self.ws_message_producer.run()
 
-        self.assertEqual(0, self.ws_message_producer._last_poll)
+        self.assertEqual(0, self.ws_message_producer._last_ws_attempt)
         self.assertEqual(exception, self.ws_message_producer.exception)
 
 
 class Test__WSMessagingThread_blocked(WSMessageProducerTest):
 
     def test_thread_not_started(self):
 
-        self.messaging_thread._last_poll = None
+        self.messaging_thread._last_ws_attempt = None
 
         self.assertFalse(self.messaging_thread.blocked)
 
@@ -224,7 +224,7 @@ def test_thread_blocked(self, _time):
         now = 100
         _time.time.return_value = now
 
-        self.messaging_thread._last_poll = now - self.messaging_thread._BLOCKED_TIMEOUT - 1
+        self.messaging_thread._last_ws_attempt = now - self.messaging_thread._BLOCKED_TIMEOUT - 1
 
         self.assertTrue(self.messaging_thread.blocked)
 
@@ -234,7 +234,7 @@ def test_thread_not_blocked(self, _time):
         now = 100
         _time.time.return_value = now
 
-        self.messaging_thread._last_poll = now - self.messaging_thread._BLOCKED_TIMEOUT + 1
+        self.messaging_thread._last_ws_attempt = now - self.messaging_thread._BLOCKED_TIMEOUT + 1
 
         self.assertFalse(self.messaging_thread.blocked)