java: async-profiler v2.5 and CPU mode with fdtransfer

Dockerfile

@@ -108,6 +108,7 @@ COPY --from=phpspy-builder /binutils/binutils-2.25/bin/bin/strings gprofiler/res
 COPY --from=async-profiler-builder /async-profiler/build/jattach gprofiler/resources/java/jattach
 COPY --from=async-profiler-builder /async-profiler/build/async-profiler-version gprofiler/resources/java/async-profiler-version
 COPY --from=async-profiler-builder /async-profiler/build/libasyncProfiler.so gprofiler/resources/java/libasyncProfiler.so
+COPY --from=async-profiler-builder /async-profiler/build/fdtransfer gprofiler/resources/java/fdtransfer
 
 COPY --from=rbspy-builder /rbspy/rbspy gprofiler/resources/ruby/rbspy
 

gprofiler/profilers/java.py

@@ -24,13 +24,15 @@
     get_process_nspid,
     pgrep_maps,
     process_comm,
+    read_perf_event_mlock_kb,
     remove_path,
     remove_prefix,
     resolve_proc_root_links,
     resource_path,
     run_in_ns,
     run_process,
     touch_path,
+    write_perf_event_mlock_kb,
 )
 
 logger = get_logger_adapter(__name__)
@@ -57,6 +59,11 @@ def jattach_path() -> str:
     return resource_path("java/jattach")
 
 
+@functools.lru_cache(maxsize=1)
+def fdtransfer_path() -> str:
+    return resource_path("java/fdtransfer")
+
+
 @functools.lru_cache(maxsize=1)
 def get_ap_version() -> str:
     return Path(resource_path("java/async-profiler-version")).read_text()
@@ -67,12 +74,11 @@ class AsyncProfiledProcess:
     Represents a process profiled with async-profiler.
     """
 
-    AP_EVENT_TYPE = "itimer"
     FORMAT_PARAMS = "ann,sig"
     OUTPUT_FORMAT = "collapsed"
     OUTPUTS_MODE = 0o622  # readable by root, writable by all
 
-    def __init__(self, process: Process, storage_dir: str, buildids: bool):
+    def __init__(self, process: Process, storage_dir: str, buildids: bool, mode: str):
         self.process = process
         self._process_root = f"/proc/{process.pid}/root"
         # not using storage_dir for AP itself on purpose: this path should remain constant for the lifetime
@@ -98,6 +104,8 @@ def __init__(self, process: Process, storage_dir: str, buildids: bool):
         self._log_path_process = remove_prefix(self._log_path_host, self._process_root)
 
         self._buildids = buildids
+        assert mode in ("cpu", "itimer"), f"unexpected mode: {mode}"
+        self._mode = mode
 
     def __enter__(self):
         os.makedirs(self._ap_dir_host, 0o755, exist_ok=True)
@@ -178,9 +186,10 @@ def _get_base_cmd(self) -> List[str]:
 
     def _get_start_cmd(self, interval: int) -> List[str]:
         return self._get_base_cmd() + [
-            f"start,event={self.AP_EVENT_TYPE},file={self._output_path_process},{self.OUTPUT_FORMAT},"
-            f"{self.FORMAT_PARAMS},interval={interval},framebuf=2000000,log={self._log_path_process}"
-            f"{',buildids' if self._buildids else ''}",
+            f"start,event={self._mode},file={self._output_path_process},"
+            f"{self.OUTPUT_FORMAT},{self.FORMAT_PARAMS},interval={interval},framebuf=2000000,"
+            f"log={self._log_path_process}{',buildids' if self._buildids else ''}"
+            f"{',fdtransfer' if self._mode == 'cpu' else ''}",
         ]
 
     def _get_stop_cmd(self, with_output: bool) -> List[str]:
@@ -208,10 +217,19 @@ def _run_async_profiler(self, cmd: List[str]) -> None:
 
             raise JattachException(e.returncode, e.cmd, e.stdout, e.stderr, self.process.pid, ap_log) from None
 
-    def start_async_profiler(self, interval: int) -> bool:
+    def _run_fdtransfer(self) -> None:
+        """
+        Start fdtransfer; it will fork & exit once ready, so we can continue with jattach.
+        """
+        run_process([fdtransfer_path(), str(self.process.pid)], communicate=False)
+
+    def start_async_profiler(self, interval: int, second_try: bool = False) -> bool:
         """
         Returns True if profiling was started; False if it was already started.
         """
+        if self._mode == "cpu" and not second_try:
+            self._run_fdtransfer()
+
         start_cmd = self._get_start_cmd(interval)
         try:
             self._run_async_profiler(start_cmd)
@@ -264,11 +282,21 @@ def read_output(self) -> Optional[str]:
             action="store_false",
             help="Skip the JDK version check (that is done before invoking async-profiler)",
         ),
+        ProfilerArgument(
+            "--java-async-profiler-mode",
+            dest="java_async_profiler_mode",
+            choices=["cpu", "itimer"],
+            default="itimer",
+            help="Select async-profiler's mode: 'cpu' (based on perf_events & fdtransfer) or 'itimer' (no perf_events)."
+            " Defaults to '%(default)s'.",
+        ),
     ],
 )
 class JavaProfiler(ProcessProfilerBase):
     JDK_EXCLUSIONS = ["OpenJ9", "Zing"]
 
+    _new_perf_event_mlock_kb = 8192
+
     def __init__(
         self,
         frequency: int,
@@ -277,6 +305,7 @@ def __init__(
         storage_dir: str,
         java_async_profiler_buildids: bool,
         java_version_check: bool,
+        java_async_profiler_mode: str,
         java_mode: str,
     ):
         assert java_mode == "ap", "Java profiler should not be initialized, wrong java_mode value given"
@@ -286,6 +315,8 @@ def __init__(
         self._interval = int((1 / frequency) * 1000_000_000)
         self._buildids = java_async_profiler_buildids
         self._version_check = java_version_check
+        self._mode = java_async_profiler_mode
+        self._saved_mlock: Optional[int] = None
 
     def _is_jdk_version_supported(self, java_version_cmd_output: str) -> bool:
         return all(exclusion not in java_version_cmd_output for exclusion in self.JDK_EXCLUSIONS)
@@ -342,7 +373,7 @@ def _profile_process(self, process: Process) -> Optional[StackToSampleCount]:
                 logger.warning(f"Process {process.pid} running unsupported Java version, skipping...")
                 return None
 
-        with AsyncProfiledProcess(process, self._storage_dir, self._buildids) as ap_proc:
+        with AsyncProfiledProcess(process, self._storage_dir, self._buildids, self._mode) as ap_proc:
             return self._profile_ap_process(ap_proc)
 
     def _profile_ap_process(self, ap_proc: AsyncProfiledProcess) -> Optional[StackToSampleCount]:
@@ -355,7 +386,7 @@ def _profile_ap_process(self, ap_proc: AsyncProfiledProcess) -> Optional[StackTo
             # not using the "resume" action because I'm not sure it properly reconfigures all settings; while stop;start
             # surely does.
             ap_proc.stop_async_profiler(with_output=False)
-            started = ap_proc.start_async_profiler(self._interval)
+            started = ap_proc.start_async_profiler(self._interval, second_try=True)
             if not started:
                 raise Exception(
                     f"async-profiler is still running in {ap_proc.process.pid}, even after trying to stop it!"
@@ -386,3 +417,33 @@ def _profile_ap_process(self, ap_proc: AsyncProfiledProcess) -> Optional[StackTo
 
     def _select_processes_to_profile(self) -> List[Process]:
         return pgrep_maps(r"^.+/libjvm\.so$")
+
+    def start(self) -> None:
+        super().start()
+        if self._mode == "cpu":
+            # short tech dive:
+            # perf has this accounting logic when mmaping perf_event_open fds (from kernel/events/core.c:perf_mmap())
+            # 1. if the *user* locked pages have not exceeded the limit of perf_event_mlock_kb, charge from
+            #    that limit.
+            # 2. after exceeding, charge from ulimit (mlock)
+            # 3. if both are expended, fail the mmap unless task is privileged / perf_event_paranoid is
+            #    permissive (-1).
+            #
+            # in our case, we run "perf" alongside and it starts before async-profiler. so its mmaps get charged from
+            # the user (root) locked pages.
+            # then, when async-profiler starts, and we profile a Java process running in a container as root (which is
+            # common), it is treated as the same user, and since its limit is expended, and the container is not
+            # privileged & has low mlock ulimit (Docker defaults to 64) - async-profiler fails to mmap!
+            # instead, we update perf_event_mlock_kb here for the lifetime of gProfiler, leaving some room
+            # for async-profiler (and also make sure "perf" doesn't use it in its entirety)
+            #
+            # (alternatively, we could change async-profiler's fdtransfer to do the mmap as well as the perf_event_open;
+            # this way, the privileged program gets charged, and when async-profiler mmaps it again, it will use the
+            # same pages and won't get charged).
+            self._saved_mlock = read_perf_event_mlock_kb()
+            write_perf_event_mlock_kb(self._new_perf_event_mlock_kb)
+
+    def stop(self) -> None:
+        super().stop()
+        if self._saved_mlock is not None:
+            write_perf_event_mlock_kb(self._saved_mlock)

gprofiler/profilers/perf.py

@@ -27,6 +27,9 @@ def perf_path() -> str:
 class PerfProcess:
     _dump_timeout_s = 5
     _poll_timeout_s = 5
+    # default number of pages used by "perf record" when perf_event_mlock_kb=516
+    # we use double for dwarf.
+    _mmap_sizes = {"fp": 129, "dwarf": 257}
 
     def __init__(
         self,
@@ -56,6 +59,12 @@ def _get_perf_cmd(self) -> List[str]:
             "-o",
             self._output_path,
             "--switch-output=signal",
+            # explicitly pass '-m', otherwise perf defaults to deriving this number from perf_event_mlock_kb,
+            # and it ends up using it entirely (and we want to spare some for async-profiler)
+            # this number scales linearly with the number of active cores (so we don't need to do this calculation
+            # here)
+            "-m",
+            str(self._mmap_sizes[self._type]),
         ] + self._extra_args
 
     def start(self) -> None:
@@ -84,12 +93,14 @@ def switch_output(self) -> None:
         self._process.send_signal(signal.SIGUSR2)
 
     def wait_and_script(self) -> str:
-        perf_data = wait_for_file_by_prefix(f"{self._output_path}.", self._dump_timeout_s, self._stop_event)
-
-        # using read1() which performs just a single read() call and doesn't read until EOF
-        # (unlike Popen.communicate())
-        assert self._process is not None
-        logger.debug(f"perf stderr: {self._process.stderr.read1(4096)}")
+        try:
+            perf_data = wait_for_file_by_prefix(f"{self._output_path}.", self._dump_timeout_s, self._stop_event)
+        finally:
+            # always read its stderr
+            # using read1() which performs just a single read() call and doesn't read until EOF
+            # (unlike Popen.communicate())
+            assert self._process is not None
+            logger.debug(f"perf stderr: {self._process.stderr.read1(4096)}")
 
         if self._inject_jit:
             inject_data = Path(f"{str(perf_data)}.inject")

gprofiler/utils.py

@@ -165,15 +165,25 @@ def run_process(
     check: bool = True,
     timeout: int = None,
     kill_signal: signal.Signals = signal.SIGKILL,
+    communicate: bool = True,
     stdin: bytes = None,
     **kwargs,
 ) -> CompletedProcess:
+    stdout = None
+    stderr = None
     with start_process(cmd, via_staticx, **kwargs) as process:
         try:
             communicate_kwargs = dict(input=stdin) if stdin is not None else {}
             if stop_event is None:
-                stdout, stderr = process.communicate(timeout=timeout, **communicate_kwargs)
+                assert timeout is None, f"expected no timeout, got {timeout!r}"
+                if communicate:
+                    # wait for stderr & stdout to be closed
+                    stdout, stderr = process.communicate(timeout=timeout, **communicate_kwargs)
+                else:
+                    # just wait for the process to exit
+                    process.wait()
             else:
+                assert communicate, "expected communicate=True if stop_event is given"
                 end_time = (time.monotonic() + timeout) if timeout is not None else None
                 while True:
                     try:
@@ -494,3 +504,14 @@ def process_comm(process: Process) -> str:
     name_line = status.splitlines()[0]
     assert name_line.startswith("Name:\t")
     return name_line.split("\t", 1)[1]
+
+
+PERF_EVENT_MLOCK_KB = "/proc/sys/kernel/perf_event_mlock_kb"
+
+
+def read_perf_event_mlock_kb() -> int:
+    return int(Path(PERF_EVENT_MLOCK_KB).read_text())
+
+
+def write_perf_event_mlock_kb(value: int) -> None:
+    Path(PERF_EVENT_MLOCK_KB).write_text(str(value))

pyi.Dockerfile

@@ -126,6 +126,7 @@ COPY --from=centos:6 /usr/bin/xargs gprofiler/resources/php/xargs
 COPY --from=async-profiler-builder /async-profiler/build/jattach gprofiler/resources/java/jattach
 COPY --from=async-profiler-builder /async-profiler/build/async-profiler-version gprofiler/resources/java/async-profiler-version
 COPY --from=async-profiler-builder /async-profiler/build/libasyncProfiler.so gprofiler/resources/java/libasyncProfiler.so
+COPY --from=async-profiler-builder /async-profiler/build/fdtransfer gprofiler/resources/java/fdtransfer
 
 COPY --from=burn-builder /go/burn/burn gprofiler/resources/burn
 

scripts/async_profiler_build.sh

@@ -5,8 +5,8 @@
 #
 set -euo pipefail
 
-VERSION=v2.0g5
-GIT_REV=3f1505bbf4a4ec1575b54804e1fbc7b9f38ad96b
+VERSION=v2.5g1
+GIT_REV="02cb785f7cc0024581a1802126f51ed94d5b0475"
 
 git clone --depth 1 -b "$VERSION" https://github.com/Granulate/async-profiler.git && cd async-profiler && git reset --hard "$GIT_REV"
 set +eu  # this funny script has errors :shrug:

tests/containers/java/Fibonacci.java

@@ -2,12 +2,26 @@
 // Copyright (c) Granulate. All rights reserved.
 // Licensed under the AGPL3 License. See LICENSE.md in the project root for license information.
 //
+import java.io.File;
+
 public class Fibonacci {
     private static long fibonacci(final int n) {
         return n <= 1 ? n : fibonacci(n - 1) + fibonacci(n - 2);
     }
 
     public static void main(final String[] args) {
+        Thread thread = new Thread() {
+            public void run() {
+                while (true) {
+                    try {
+                        new File("/").list();
+                    } catch (Exception e) {
+                    }
+                }
+            }
+        };
+        thread.start();
+
         while (true) {
             fibonacci(30);
         }