refactor(ipc): more general shared file

We refactor the file queue into a more generic shared string file for
IPC that can be used by more components.

Author: P403n1x87

ddtrace/internal/_file_queue.py

@@ -0,0 +1,151 @@
+import os
+import secrets
+import tempfile
+import typing
+
+from ddtrace.internal._unpatched import unpatched_open
+from ddtrace.internal.compat import Path
+from ddtrace.internal.logger import get_logger
+
+
+log = get_logger(__name__)
+
+
+MAX_FILE_SIZE = 8192
+
+
+class BaseLock:
+    def __init__(self, file: typing.IO[typing.Any]):
+        self.file = file
+
+    def acquire(self):
+        ...
+
+    def release(self):
+        ...
+
+    def __enter__(self):
+        self.acquire()
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_tb):
+        self.release()
+
+
+try:
+    # Unix based file locking
+    # Availability: Unix, not Emscripten, not WASI.
+    import fcntl
+
+    class ReadLock(BaseLock):
+        def __init__(self, file: typing.IO[typing.Any]):
+            self.file = file
+
+        def acquire(self):
+            fcntl.lockf(self.file, fcntl.LOCK_SH)
+
+        def release(self):
+            fcntl.lockf(self.file, fcntl.LOCK_UN)
+
+    class WriteLock(BaseLock):
+        def __init__(self, file: typing.IO[typing.Any]):
+            self.file = file
+
+        def acquire(self):
+            fcntl.lockf(self.file, fcntl.LOCK_EX)
+
+        def release(self):
+            fcntl.lockf(self.file, fcntl.LOCK_UN)
+
+    def open_file(path, mode):
+        return unpatched_open(path, mode)
+
+except ModuleNotFoundError:
+    # Availability: Windows
+    import msvcrt
+
+    class ReadLock(BaseLock):
+        def __init__(self, file: typing.IO[typing.Any]):
+            self.file = file
+
+        def acquire(self):
+            fcntl.lockf(self.file, fcntl.LOCK_SH)
+
+        def release(self):
+            fcntl.lockf(self.file, fcntl.LOCK_UN)
+
+    class WriteLock(BaseLock):
+        def __init__(self, file: typing.IO[typing.Any]):
+            self.file = file
+
+        def acquire(self):
+            self.file.seek(0)
+            msvcrt.locking(self.file.fileno(), msvcrt.LK_RLCK, MAX_FILE_SIZE)
+
+        def release(self):
+            self.file.seek(0)
+            msvcrt.locking(self.file.fileno(), msvcrt.LK_UNLCK, MAX_FILE_SIZE)
+
+    def open_file(path, mode):
+        import _winapi
+
+        # force all modes to be read/write binary
+        mode = "r+b"
+        flag = _winapi.GENERIC_READ | _winapi.GENERIC_WRITE
+        fd_flag = os.O_RDWR | os.O_CREAT | os.O_BINARY | os.O_RANDOM
+        SHARED_READ_WRITE = 0x7
+        OPEN_ALWAYS = 4
+        RANDOM_ACCESS = 0x10000000
+        handle = _winapi.CreateFile(path, flag, SHARED_READ_WRITE, 0, OPEN_ALWAYS, RANDOM_ACCESS, 0)
+        fd = msvcrt.open_osfhandle(handle, fd_flag | os.O_NOINHERIT)
+        return unpatched_open(fd, mode)
+
+
+TMPDIR = Path(tempfile.gettempdir())
+
+
+class SharedStringFile:
+    """A simple shared-file implementation for multiprocess communication."""
+
+    def __init__(self) -> None:
+        self.filename: typing.Optional[str] = str(TMPDIR / secrets.token_hex(8))
+
+    def put(self, data: str) -> None:
+        """Put a string into the file."""
+        if self.filename is None:
+            return
+
+        try:
+            with open_file(self.filename, "ab") as f, WriteLock(f):
+                f.seek(0, os.SEEK_END)
+                dt = (data + "\x00").encode()
+                if f.tell() + len(dt) <= MAX_FILE_SIZE:
+                    f.write(dt)
+        except Exception:  # nosec
+            pass
+
+    def peekall(self) -> typing.List[str]:
+        """Peek at all strings from the file."""
+        if self.filename is None:
+            return []
+
+        try:
+            with open_file(self.filename, "r+b") as f, ReadLock(f):
+                return f.read().strip(b"\x00").decode().split("\x00")
+        except Exception:  # nosec
+            return []
+
+    def snatchall(self) -> typing.List[str]:
+        """Retrieve and remove all strings from the file."""
+        if self.filename is None:
+            return []
+
+        try:
+            with open_file(self.filename, "r+b") as f, WriteLock(f):
+                try:
+                    return f.read().strip(b"\x00").decode().split("\x00")
+                finally:
+                    f.seek(0)
+                    f.truncate()
+        except Exception:  # nosec
+            return []

ddtrace/internal/ipc.py

@@ -510,9 +510,9 @@ def __init__(self):
         self._extra_services_queue = None
         if self._remote_config_enabled and not in_aws_lambda():
             # lazy load slow import
-            from ddtrace.internal._file_queue import File_Queue
+            from ddtrace.internal.ipc import SharedStringFile
 
-            self._extra_services_queue = File_Queue()
+            self._extra_services_queue = SharedStringFile()
 
         self._unparsed_service_mapping = _get_config("DD_SERVICE_MAPPING", "")
         self.service_mapping = parse_tags_str(self._unparsed_service_mapping)
@@ -703,7 +703,7 @@ def _get_extra_services(self):
         # type: () -> set[str]
         if self._extra_services_queue is None:
             return set()
-        self._extra_services.update(self._extra_services_queue.get_all())
+        self._extra_services.update(set(self._extra_services_queue.snatchall()))
         while len(self._extra_services) > 64:
             self._extra_services.pop()
         return self._extra_services