Repository: scoutapp/scout_apm_python
Pull request #450 (2020-01-03 18:04:35)

Rewrite CoreAgentSocket as a SingletonThread
Refactor in order to:

* Use the simpler, tested, idempotent thread logic of `SingletonThread`, extracted from `SamplersThread` in #448.
* Use a shared command queue - commands enqueued while the thread happens to be shutting down will still be sent.
* Provide a simpler interface to users - simply call `send(command)`.
* Avoid `ResourceWarning`s from sockets being left open in tests - they should all be `close()`d now.

adamchainz: For the record, here's the `ResourceWarning`s I was seeing previously:

```

tests/integration/core/test_socket.py::test_socket_instance_is_recreated_if_not_running

  /home/travis/build/scoutapp/scout_apm_python/.tox/py36-django111/lib/python3.6/site-packages/scout_apm/core/socket.py:35: ResourceWarning: unclosed <socket.socket fd=24, family=AddressFamily.AF_UNIX, type=SocketKind.SOCK_STREAM, proto=0>

    cls._instance = cls()
```

Patch:
diff --git a/src/scout_apm/core/__init__.py b/src/scout_apm/core/__init__.py
index a108ff4..8f7c17e 100644
--- a/src/scout_apm/core/__init__.py
+++ b/src/scout_apm/core/__init__.py
@@ -10,7 +10,7 @@ from scout_apm.core import objtrace
 from scout_apm.core.config import scout_config
 from scout_apm.core.core_agent_manager import CoreAgentManager
 from scout_apm.core.metadata import report_app_metadata
-from scout_apm.core.socket import CoreAgentSocket
+from scout_apm.core.socket import CoreAgentSocketThread
 
 logger = logging.getLogger(__name__)
 
@@ -43,6 +43,6 @@ def install(config=None):
     report_app_metadata()
     if launched:
         # Stop the thread to avoid running threads pre-fork
-        CoreAgentSocket.instance().stop()
+        CoreAgentSocketThread.ensure_stopped()
 
     return True
diff --git a/src/scout_apm/core/metadata.py b/src/scout_apm/core/metadata.py
index 0b0d303..7b135fb 100644
--- a/src/scout_apm/core/metadata.py
+++ b/src/scout_apm/core/metadata.py
@@ -7,11 +7,11 @@ from os import getpid
 
 from scout_apm.core.commands import ApplicationEvent
 from scout_apm.core.config import scout_config
-from scout_apm.core.socket import CoreAgentSocket
+from scout_apm.core.socket import CoreAgentSocketThread
 
 
 def report_app_metadata():
-    CoreAgentSocket.instance().send(
+    CoreAgentSocketThread.send(
         ApplicationEvent(
             event_type="scout.metadata",
             event_value=get_metadata(),
diff --git a/src/scout_apm/core/samplers/thread.py b/src/scout_apm/core/samplers/thread.py
index f89421d..371aff1 100644
--- a/src/scout_apm/core/samplers/thread.py
+++ b/src/scout_apm/core/samplers/thread.py
@@ -9,7 +9,7 @@ import threading
 from scout_apm.core.commands import ApplicationEvent
 from scout_apm.core.samplers.cpu import Cpu
 from scout_apm.core.samplers.memory import Memory
-from scout_apm.core.socket import CoreAgentSocket
+from scout_apm.core.socket import CoreAgentSocketThread
 from scout_apm.core.threading import SingletonThread
 
 logger = logging.getLogger(__name__)
@@ -34,7 +34,7 @@ class SamplersThread(SingletonThread):
                         timestamp=dt.datetime.utcnow(),
                         source="Pid: " + str(os.getpid()),
                     )
-                    CoreAgentSocket.instance().send(event)
+                    CoreAgentSocketThread.send(event)
 
             should_stop = self._stop_event.wait(timeout=60)
             if should_stop:
diff --git a/src/scout_apm/core/socket.py b/src/scout_apm/core/socket.py
index df599ab..0efd429 100644
--- a/src/scout_apm/core/socket.py
+++ b/src/scout_apm/core/socket.py
@@ -12,123 +12,69 @@ import time
 from scout_apm.compat import queue
 from scout_apm.core.commands import Register
 from scout_apm.core.config import scout_config
+from scout_apm.core.threading import SingletonThread
 
-SECOND = 1  # time unit - monkey-patched in tests to make them run faster
+# Time unit - monkey-patched in tests to make them run faster
+SECOND = 1
 
 logger = logging.getLogger(__name__)
 
 
-class CoreAgentSocket(threading.Thread):
-    _instance = None
+class CoreAgentSocketThread(SingletonThread):
     _instance_lock = threading.Lock()
+    _stop_event = threading.Event()
+    _command_queue = queue.Queue(maxsize=500)
 
     @classmethod
-    def instance(cls):
-        with cls._instance_lock:
-            # No instance exists yet.
-            if cls._instance is None:
-                cls._instance = cls()
-                return cls._instance
-
-            # An instance exists but is no longer running.
-            if not cls._instance.running():
-                cls._instance = cls()
-                return cls._instance
-
-            # An instance exists and is running (or in the process of
-            # starting or in the process of stopping). In any case,
-            # return this instance.
-            return cls._instance
-
-    def __init__(self, *args, **kwargs):
-        super(CoreAgentSocket, self).__init__()
-        # Socket related
-        self.socket_path = scout_config.value("socket_path")
-        self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-
-        # Threading control related
-        self._started_event = threading.Event()
-        self._stop_event = threading.Event()
-        self._stopped_event = threading.Event()
-
-        # Command queues
-        self.command_queue = queue.Queue(maxsize=500)
-
-        # Set Thread options
-        self.daemon = True
+    def _on_stop(cls):
+        super(CoreAgentSocketThread, cls)._on_stop()
+        # Unblock _command_queue.get()
+        cls._command_queue.put(None, False)
 
-        # Set the started event here to avoid races in the class instance()
-        # method. If there is an exception in the socket thread then it will
-        # clear this event on exit.
-        self._started_event.set()
-
-        # Now call start() which eventually launches run() in another thread.
-        self.start()
-
-    def __del__(self):
-        self.stop()
-
-    def running(self):
-        return self._started_event.is_set()
+    @classmethod
+    def send(cls, command):
+        try:
+            cls._command_queue.put(command, False)
+        except queue.Full as exc:
+            # TODO mark the command as not queued?
+            logger.debug("CoreAgentSocketThread error on send: %r", exc, exc_info=exc)
 
-    def stop(self):
-        if self._started_event.is_set():
-            self._stop_event.set()
-            self.command_queue.put(None, False)  # unblock self.command_queue.get
-            stopped = self._stopped_event.wait(2 * SECOND)
-            if stopped:
-                return True
-            else:
-                logger.debug("CoreAgentSocket Failed to stop thread within timeout!")
-                return False
-        else:
-            return True
+        cls.ensure_started()
 
     def run(self):
-        """
-        Called by the threading system
-        """
+        self.socket_path = scout_config.value("socket_path")
+        self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
 
         try:
             self._connect()
             self._register()
             while True:
                 try:
-                    body = self.command_queue.get(block=True, timeout=1 * SECOND)
+                    body = self._command_queue.get(block=True, timeout=1 * SECOND)
                 except queue.Empty:
                     body = None
 
                 if body is not None:
                     result = self._send(body)
                     if result:
-                        self.command_queue.task_done()
+                        self._command_queue.task_done()
                     else:
                         # Something was wrong with the socket.
                         self._disconnect()
                         self._connect()
                         self._register()
 
-                # Check for stop event after a read from the queue. This is to
-                # allow you to open a socket, immediately send to it, and then
-                # stop it. We do this in the Metadata send at application start
-                # time
+                # Check for stop event after each read. This allows opening,
+                # sending, and then immediately stopping. We do this for
+                # the metadata event at application start time.
                 if self._stop_event.is_set():
-                    logger.debug("CoreAgentSocket thread stopping.")
+                    logger.debug("CoreAgentSocketThread stopping.")
                     break
         except Exception as exc:
-            logger.debug("CoreAgentSocket thread exception: %r", exc, exc_info=exc)
+            logger.debug("CoreAgentSocketThread exception: %r", exc, exc_info=exc)
         finally:
-            self._started_event.clear()
-            self._stop_event.clear()
-            self._stopped_event.set()
-            logger.debug("CoreAgentSocket thread stopped.")
-
-    def send(self, command):
-        try:
-            self.command_queue.put(command, False)
-        except queue.Full as exc:
-            # TODO mark the command as not queued?
-            logger.debug("CoreAgentSocket error on send: %r", exc, exc_info=exc)
+            self.socket.close()
+            logger.debug("CoreAgentSocketThread stopped.")
 
     def _send(self, command):
         msg = command.message()
@@ -145,7 +91,7 @@ class CoreAgentSocket(threading.Thread):
             self.socket.sendall(self._message_length(data))
         except OSError as exc:
             logger.debug(
-                "CoreAgentSocket exception on length _send: "
+                "CoreAgentSocketThread exception on length _send: "
                 "%r on PID: %s on thread: %s",
                 exc,
                 os.getpid(),
@@ -158,7 +104,7 @@ class CoreAgentSocket(threading.Thread):
             self.socket.sendall(data.encode())
         except OSError as exc:
             logger.debug(
-                "CoreAgentSocket exception on data _send: "
+                "CoreAgentSocketThread exception on data _send: "
                 "%r on PID: %s on thread: %s",
                 exc,
                 os.getpid(),
@@ -192,7 +138,7 @@ class CoreAgentSocket(threading.Thread):
             return message
         except OSError as exc:
             logger.debug(
-                "CoreAgentSocket error on read response: %r", exc, exc_info=exc
+                "CoreAgentSocketThread error on read response: %r", exc, exc_info=exc
             )
             return None
 
@@ -208,7 +154,10 @@ class CoreAgentSocket(threading.Thread):
     def _connect(self, connect_attempts=5, retry_wait_secs=1):
         for attempt in range(1, connect_attempts + 1):
             logger.debug(
-                "CoreAgentSocket attempt %d, connecting to %s, PID: %s, Thread: %s",
+                (
+                    "CoreAgentSocketThread attempt %d, connecting to %s, "
+                    + "PID: %s, Thread: %s"
+                ),
                 attempt,
                 self.socket_path,
                 os.getpid(),
@@ -217,22 +166,24 @@ class CoreAgentSocket(threading.Thread):
             try:
                 self.socket.connect(self.socket_path)
                 self.socket.settimeout(3 * SECOND)
-                logger.debug("CoreAgentSocket is connected")
+                logger.debug("CoreAgentSocketThread connected")
                 return True
             except socket.error as exc:
-                logger.debug("CoreAgentSocket connection error: %r", exc, exc_info=exc)
+                logger.debug(
+                    "CoreAgentSocketThread connection error: %r", exc, exc_info=exc
+                )
                 # Return without waiting when reaching the maximum number of attempts.
                 if attempt >= connect_attempts:
                     return False
                 time.sleep(retry_wait_secs * SECOND)
 
     def _disconnect(self):
-        logger.debug("CoreAgentSocket disconnecting from %s", self.socket_path)
+        logger.debug("CoreAgentSocketThread disconnecting from %s", self.socket_path)
         try:
             self.socket.close()
         except socket.error as exc:
             logger.debug(
-                "CoreAgentSocket exception on disconnect: %r", exc, exc_info=exc
+                "CoreAgentSocketThread exception on disconnect: %r", exc, exc_info=exc
             )
         finally:
             self.socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
diff --git a/src/scout_apm/core/tracked_request.py b/src/scout_apm/core/tracked_request.py
index ae29d93..283c619 100644
--- a/src/scout_apm/core/tracked_request.py
+++ b/src/scout_apm/core/tracked_request.py
@@ -10,7 +10,7 @@ from scout_apm.core.commands import BatchCommand
 from scout_apm.core.n_plus_one_tracker import NPlusOneTracker
 from scout_apm.core.samplers.memory import get_rss_in_mb
 from scout_apm.core.samplers.thread import SamplersThread
-from scout_apm.core.socket import CoreAgentSocket
+from scout_apm.core.socket import CoreAgentSocketThread
 
 logger = logging.getLogger(__name__)
 
@@ -123,7 +123,7 @@ class TrackedRequest(object):
             self.tag("mem_delta", self._get_mem_delta())
             if not self.is_ignored():
                 batch_command = BatchCommand.from_tracked_request(self)
-                CoreAgentSocket.instance().send(batch_command)
+                CoreAgentSocketThread.send(batch_command)
             SamplersThread.ensure_started()
 
         from scout_apm.core.context import context


Test patch:
diff --git a/tests/integration/core/test_socket.py b/tests/integration/core/test_socket.py
index 436c74a..11d4c02 100644
--- a/tests/integration/core/test_socket.py
+++ b/tests/integration/core/test_socket.py
@@ -5,7 +5,7 @@ import time
 
 import pytest
 
-from scout_apm.core.socket import CoreAgentSocket
+from scout_apm.core.socket import CoreAgentSocketThread
 from tests.compat import mock
 from tests.conftest import is_running, shutdown
 
@@ -25,37 +25,14 @@ def running_agent(core_agent_manager):
 
 @pytest.fixture
 def socket(running_agent):
-    socket = CoreAgentSocket.instance()
+    socket = CoreAgentSocketThread.ensure_started()
     try:
-        time.sleep(0.01)  # wait for socket to connect and register
-        yield socket
-    finally:
-        socket.stop()
-        socket.join()
-
+        # Wait for socket to connect and register:
+        time.sleep(0.01)
 
-def test_socket_instance_is_a_singleton(running_agent):
-    socket1 = CoreAgentSocket.instance()
-    socket2 = CoreAgentSocket.instance()
-    try:
-        assert socket2 is socket1
-    finally:
-        socket1.stop()
-        socket1.join()
-        socket2.stop()
-        socket2.join()
-
-
-def test_socket_instance_is_recreated_if_not_running(running_agent):
-    socket1 = CoreAgentSocket.instance()
-    socket1.stop()
-    socket1.join()
-    socket2 = CoreAgentSocket.instance()
-    try:
-        assert socket2 is not socket1
+        yield socket
     finally:
-        socket2.stop()
-        socket2.join()
+        CoreAgentSocketThread.ensure_stopped()
 
 
 class Command(object):
@@ -64,7 +41,7 @@ class Command(object):
 
 
 def test_send(socket):
-    socket.send(Command())
+    CoreAgentSocketThread.send(Command())
 
 
 class NonSerializableCommand(object):
@@ -73,10 +50,10 @@ class NonSerializableCommand(object):
 
 
 def test_send_serialization_error(socket):
-    socket.send(NonSerializableCommand())
+    CoreAgentSocketThread.send(NonSerializableCommand())
 
 
 @mock.patch("socket.socket.sendall")
 def test_send_network_error(sendall, socket):
     sendall.side_effect = OSError
-    socket.send(Command())
+    CoreAgentSocketThread.send(Command())
diff --git a/tests/unit/core/test_metadata.py b/tests/unit/core/test_metadata.py
index 3b69d99..e7d5243 100644
--- a/tests/unit/core/test_metadata.py
+++ b/tests/unit/core/test_metadata.py
@@ -10,7 +10,7 @@ from tests.compat import mock
 from tests.tools import pretend_package_unavailable
 
 
-@mock.patch("scout_apm.core.socket.CoreAgentSocket.send")
+@mock.patch("scout_apm.core.socket.CoreAgentSocketThread.send")
 def test_report_app_metadata(send):
     report_app_metadata()
 
@@ -26,7 +26,7 @@ def test_report_app_metadata(send):
     assert ("pytest", pytest.__version__) in data["libraries"]
 
 
-@mock.patch("scout_apm.core.socket.CoreAgentSocket.send")
+@mock.patch("scout_apm.core.socket.CoreAgentSocketThread.send")
 def test_report_app_metadata_no_importlib_metadata(send):
     if sys.version_info >= (3, 8):
         module_name = "importlib"