perf(rc): improve performance of subprocess service sharing [backport 3.9] (#13685)

Backport b686305 from #13651 to 3.9.

This change will only write service names to the shared file queue only
if we have not shared that service name yet.

This improvement avoids locking and writing to disk every time we call
Pin.onto which can happen a lot when database connections are
cycled/recreated frequently.

It also helps ensure that the same process doesn't continually write the
same service name to the same file over and over again.

I also updated the tests to use subprocesses/forking as is expected with
it's behavior for RC.

## Checklist
- [x] PR author has checked that all the criteria below are met
- The PR description includes an overview of the change
- The PR description articulates the motivation for the change
- The change includes tests OR the PR description describes a testing
strategy
- The PR description notes risks associated with the change, if any
- Newly-added code is easy to change
- The change follows the [library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
- The change includes or references documentation updates if necessary
- Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))

## Reviewer Checklist
- [x] Reviewer has checked that all the criteria below are met 
- Title is accurate
- All changes are related to the pull request's stated goal
- Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- Testing strategy adequately addresses listed risks
- Newly-added code is easy to change
- Release note makes sense to a user of the library
- If necessary, author has acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment
- Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

Co-authored-by: Brett Langdon <[email protected]>
Co-authored-by: Christophe Papazian <[email protected]>

Author: 3 people

ddtrace/settings/_config.py

@@ -504,6 +504,7 @@ def __init__(self):
         self.version = _get_config("DD_VERSION", self.tags.get("version"))
         self._http_server = self._HTTPServerConfig()
 
+        self._extra_services_sent = set()  # type: set[str]
         self._extra_services_queue = None
         if self._remote_config_enabled and not in_aws_lambda():
             # lazy load slow import
@@ -667,8 +668,12 @@ def __getattr__(self, name) -> Any:
     def _add_extra_service(self, service_name: str) -> None:
         if self._extra_services_queue is None:
             return
-        if service_name != self.service:
-            self._extra_services_queue.put(service_name)
+
+        if service_name == self.service or service_name in self._extra_services_sent:
+            return
+
+        self._extra_services_queue.put(service_name)
+        self._extra_services_sent.add(service_name)
 
     def _get_extra_services(self):
         # type: () -> set[str]

tests/internal/service_name/test_extra_services_names.py

@@ -1,42 +1,87 @@
-import random
-import re
-import threading
-import time
+import os
+import sys
 
 import pytest
 
+
+@pytest.mark.skipif(sys.platform in ("win32", "cygwin"), reason="Fork not supported on Windows")
+def test_config_extra_service_names_fork(run_python_code_in_subprocess):
+    code = """
+import ddtrace.auto
 import ddtrace
 
+import re
+import os
+import sys
+import time
+
+children = []
+for i in range(10):
+    pid = os.fork()
+    if pid == 0:
+        # Child process
+        ddtrace.config._add_extra_service(f"extra_service_{i}")
+        time.sleep(0.1)  # Ensure the child has time to save the service
+        sys.exit(0)
+    else:
+        # Parent process
+        children.append(pid)
+
+for pid in children:
+    os.waitpid(pid, 0)
+
+extra_services = ddtrace.config._get_extra_services()
+extra_services.discard("sqlite")  # coverage
+assert len(extra_services) == 10, extra_services
+assert all(re.match(r"extra_service_\\d+", service) for service in extra_services), extra_services
+"""
 
-MAX_NAMES = 64
+    env = os.environ.copy()
+    env["DD_REMOTE_CONFIGURATION_ENABLED"] = "true"
+    stdout, stderr, status, _ = run_python_code_in_subprocess(code, env=env)
+    assert status == 0, (stdout, stderr, status)
 
 
-@pytest.mark.parametrize("nb_service", [2, 16, 64, 256])
-def test_service_name(nb_service):
-    ddtrace.config._extra_services = set()
+def test_config_extra_service_names_duplicates(run_python_code_in_subprocess):
+    code = """
+import ddtrace.auto
+import ddtrace
+import re
+import os
+import sys
+import time
+
+for _ in range(10):
+    ddtrace.config._add_extra_service("extra_service_1")
 
-    def write_in_subprocess(id_nb):
-        time.sleep(random.random())
-        ddtrace.config._add_extra_service(f"extra_service_{id_nb}")
+extra_services = ddtrace.config._get_extra_services()
+extra_services.discard("sqlite")  # coverage
+assert extra_services == {"extra_service_1"}
+    """
 
-    default_remote_config_enabled = ddtrace.config._remote_config_enabled
-    ddtrace.config._remote_config_enabled = True
-    if ddtrace.config._extra_services_queue is None:
-        import ddtrace.internal._file_queue as file_queue
+    env = os.environ.copy()
+    env["DD_REMOTE_CONFIGURATION_ENABLED"] = "true"
+    stdout, stderr, status, _ = run_python_code_in_subprocess(code, env=env)
+    assert status == 0, (stdout, stderr, status)
 
-        ddtrace.config._extra_services_queue = file_queue.File_Queue()
 
-    threads = [threading.Thread(target=write_in_subprocess, args=(i,)) for i in range(nb_service)]
-    for thread in threads:
-        thread.start()
-    for thread in threads:
-        thread.join()
+def test_config_extra_service_names_rc_disabled(run_python_code_in_subprocess):
+    code = """
+import ddtrace.auto
+import ddtrace
+import re
+import os
+import sys
+import time
+
+for _ in range(10):
+    ddtrace.config._add_extra_service("extra_service_1")
 
-    extra_services = ddtrace.config._get_extra_services()
-    assert len(extra_services) == min(nb_service, MAX_NAMES)
-    assert all(re.match(r"extra_service_\d+", service) for service in extra_services)
+extra_services = ddtrace.config._get_extra_services()
+assert len(extra_services) == 0
+    """
 
-    ddtrace.config._remote_config_enabled = default_remote_config_enabled
-    if not default_remote_config_enabled:
-        ddtrace.config._extra_services_queue = None
-    ddtrace.config._extra_services = set()
+    env = os.environ.copy()
+    env["DD_REMOTE_CONFIGURATION_ENABLED"] = "false"
+    stdout, stderr, status, _ = run_python_code_in_subprocess(code, env=env)
+    assert status == 0, (stdout, stderr, status)