Add a timestamp for the samples and weight the report by the sample duration

Author: mattip

src/compat.c

@@ -96,9 +96,48 @@ int vmp_write_time_now(int marker) {
     vmp_write_all(buffer, __SIZE);
     return 0;
 }
+
+#if defined(VMPROF_APPLE) && !defined(CLOCK_MONOTONIC)
+/* SDK older than macOS 10.12: no clock_gettime.  mach_absolute_time is a
+   monotonic clock; there is no signal-safe process cpu clock, so both
+   modes use it (it is what setitimer(ITIMER_REAL) measures anyway). */
+#include <mach/mach_time.h>
+int64_t vmp_sample_time_ns(int cpu_time)
+{
+    mach_timebase_info_data_t tb;
+    uint64_t t = mach_absolute_time();
+    (void)cpu_time;
+    if (mach_timebase_info(&tb) != KERN_SUCCESS || tb.denom == 0)
+        return (int64_t)t;
+    return (int64_t)(t / tb.denom * tb.numer + t % tb.denom * tb.numer / tb.denom);
+}
+#else
+int64_t vmp_sample_time_ns(int cpu_time)
+{
+    struct timespec ts;
+    clockid_t clk = cpu_time ? CLOCK_PROCESS_CPUTIME_ID : CLOCK_MONOTONIC;
+    if (clock_gettime(clk, &ts) != 0)
+        return 0;
+    return (int64_t)ts.tv_sec * 1000000000LL + ts.tv_nsec;
+}
+#endif
 #endif
 
 #ifdef VMPROF_WINDOWS
+int64_t vmp_sample_time_ns(int cpu_time)
+{
+    /* the frequency is fixed at boot, so it is safe to cache it */
+    static LARGE_INTEGER freq;
+    LARGE_INTEGER now;
+    (void)cpu_time;
+    if (freq.QuadPart == 0 && !QueryPerformanceFrequency(&freq))
+        return 0;
+    if (!QueryPerformanceCounter(&now))
+        return 0;
+    return (int64_t)(now.QuadPart / freq.QuadPart * 1000000000LL +
+                     now.QuadPart % freq.QuadPart * 1000000000LL / freq.QuadPart);
+}
+
 int vmp_write_time_now(int marker) {
     char buffer[__SIZE];
     struct timezone_buf buf;

src/compat.h

@@ -23,5 +23,13 @@ int vmp_write_all(const char *buf, size_t bufsize);
 int vmp_write_time_now(int marker);
 int vmp_write_meta(const char * key, const char * value);
 
+/* Nanosecond timestamp stored at the end of every stack sample.  With
+   'cpu_time' set it reads the process cpu-time clock, which is the clock
+   that drives ITIMER_PROF; otherwise it reads a monotonic wall clock, which
+   drives ITIMER_REAL and the Windows sampler thread.  The reader weights
+   each sample by the distance to the previous one on the same clock, so
+   timer signals that were dropped still count.  Async-signal-safe. */
+int64_t vmp_sample_time_ns(int cpu_time);
+
 int vmp_profile_fileno(void);
 void vmp_set_profile_fileno(int fileno);

src/vmprof.h

@@ -27,6 +27,9 @@
 #define VERSION_MODE_AWARE '\x04'
 #define VERSION_DURATION '\x05'
 #define VERSION_TIMESTAMP '\x06'
+/* every MARKER_STACKTRACE record ends with a 64-bit nanosecond timestamp,
+   see vmp_sample_time_ns() */
+#define VERSION_SAMPLE_TIME '\x07'
 
 #define PROFILE_MEMORY '\x01'
 #define PROFILE_LINES  '\x02'

src/vmprof_common.c

@@ -135,7 +135,7 @@ int opened_profile(const char *interp_name, int memory, int proflines, int nativ
     }
     header.interp_name[0] = MARKER_HEADER;
     header.interp_name[1] = '\x00';
-    header.interp_name[2] = VERSION_TIMESTAMP;
+    header.interp_name[2] = VERSION_SAMPLE_TIME;
     header.interp_name[3] = memory*PROFILE_MEMORY + proflines*PROFILE_LINES + \
                             native*PROFILE_NATIVE + real_time*PROFILE_REAL_TIME;
 #ifdef RPYTHON_VMPROF

src/vmprof_common.h

@@ -41,6 +41,12 @@ ssize_t remove_threads(void);
 #define MAX_STACK_DEPTH   \
     ((SINGLE_BUF_SIZE - sizeof(struct prof_stacktrace_s)) / sizeof(void *))
 
+/* Slots at the end of a stack sample that are not stack entries: the thread
+   id, the (optional) rss and the 64-bit sample timestamp.  Walk at most
+   MAX_STACK_DEPTH - STACK_TRAILER_SLOTS frames so that they always fit. */
+#define STACK_TRAILER_SLOTS \
+    (2 + (sizeof(int64_t) + sizeof(void *) - 1) / sizeof(void *))
+
 /*
  * NOTE SHOULD NOT BE DONE THIS WAY. Here is an example why:
  * assume the following struct content:
@@ -73,6 +79,9 @@ typedef struct prof_stacktrace_s {
     char marker;
     long count, depth;
     void *stack[];
+    /* the file record continues after stack[depth] with the thread id,
+       the rss if memory profiling is on, and an int64_t timestamp
+       (VERSION_SAMPLE_TIME, see vmp_sample_time_ns) */
 } prof_stacktrace_s;
 
 #define SIZEOF_PROF_STACKTRACE sizeof(long)+sizeof(long)+sizeof(char)

src/vmprof_unix.c

@@ -95,13 +95,16 @@ void segfault_handler(int arg)
 int _vmprof_sample_stack(struct profbuf_s *p, PY_THREAD_STATE_T * tstate, ucontext_t * uc)
 {
     int depth;
+    int64_t sample_time;
     struct prof_stacktrace_s *st = (struct prof_stacktrace_s *)p->data;
     st->marker = MARKER_STACKTRACE;
     st->count = 1;
 #ifdef RPYTHON_VMPROF
-    depth = get_stack_trace(get_vmprof_stack(), st->stack, MAX_STACK_DEPTH-1, (intptr_t)GetPC(uc));
+    depth = get_stack_trace(get_vmprof_stack(), st->stack,
+                            MAX_STACK_DEPTH - STACK_TRAILER_SLOTS, (intptr_t)GetPC(uc));
 #else
-    depth = get_stack_trace(tstate, st->stack, MAX_STACK_DEPTH-1, (intptr_t)NULL);
+    depth = get_stack_trace(tstate, st->stack,
+                            MAX_STACK_DEPTH - STACK_TRAILER_SLOTS, (intptr_t)NULL);
 #endif
     // useful for tests (see test_stop_sampling)
 #ifndef RPYTHON_LL2CTYPES
@@ -114,8 +117,13 @@ int _vmprof_sample_stack(struct profbuf_s *p, PY_THREAD_STATE_T * tstate, uconte
     long rss = get_current_proc_rss();
     if (rss >= 0)
         st->stack[depth++] = (void*)rss;
+    /* ITIMER_PROF counts process cpu time, ITIMER_REAL wall time: stamp the
+       sample with the clock that drives the timer.  memcpy: the slot is only
+       pointer-aligned, which is less than int64_t needs on 32 bit. */
+    sample_time = vmp_sample_time_ns(vmprof_get_itimer_type() == ITIMER_PROF);
+    memcpy(&st->stack[depth], &sample_time, sizeof(sample_time));
     p->data_offset = offsetof(struct prof_stacktrace_s, marker);
-    p->data_size = (depth * sizeof(void *) +
+    p->data_size = (depth * sizeof(void *) + sizeof(sample_time) +
                     sizeof(struct prof_stacktrace_s) -
                     offsetof(struct prof_stacktrace_s, marker));
     return 1;

src/vmprof_win.c

@@ -90,11 +90,17 @@ HANDLE write_mutex;
 
 #include "vmprof_common.h"
 
+/* Fill 'stack' with a sample of the given thread.  Returns the number of
+   pointer-sized slots written after the header (frames plus the thread id),
+   or <= 0 if nothing was sampled.  The record is followed by the int64_t
+   sample timestamp, so the caller must write
+   SIZEOF_PROF_STACKTRACE + depth * sizeof(void*) + sizeof(int64_t) bytes. */
 int vmprof_snapshot_thread(DWORD thread_id, PY_WIN_THREAD_STATE *tstate, prof_stacktrace_s *stack)
 {
     HRESULT result;
     HANDLE hThread;
     int depth;
+    int64_t sample_time;
     CONTEXT ctx;
 #ifdef RPYTHON_LL2CTYPES
     return 0; // not much we can do
@@ -115,9 +121,11 @@ int vmprof_snapshot_thread(DWORD thread_id, PY_WIN_THREAD_STATE *tstate, prof_st
     if (!GetThreadContext(hThread, &ctx))
         return -1;
     depth = get_stack_trace(tstate->vmprof_tl_stack,
-                     stack->stack, MAX_STACK_DEPTH-2, ctx.Eip);
+                     stack->stack, MAX_STACK_DEPTH - STACK_TRAILER_SLOTS, ctx.Eip);
     stack->depth = depth;
     stack->stack[depth++] = thread_id;
+    sample_time = vmp_sample_time_ns(0);
+    memcpy(&stack->stack[depth], &sample_time, sizeof(sample_time));
     stack->count = 1;
     stack->marker = MARKER_STACKTRACE;
     ResumeThread(hThread);
@@ -142,9 +150,9 @@ int vmprof_snapshot_thread(DWORD thread_id, PY_WIN_THREAD_STATE *tstate, prof_st
 #else
         frame = PyThreadState_GetFrame(tstate);
 #endif
-        /* leave room for the thread id appended below */
+        /* leave room for the thread id and timestamp appended below */
         depth = vmp_walk_and_record_stack(frame, stack->stack,
-                                          MAX_STACK_DEPTH - 1, 0, 0);
+                                          MAX_STACK_DEPTH - STACK_TRAILER_SLOTS, 0, 0);
 #ifdef _MSC_VER
     } __except (EXCEPTION_EXECUTE_HANDLER) {
         depth = -1;
@@ -160,6 +168,9 @@ int vmprof_snapshot_thread(DWORD thread_id, PY_WIN_THREAD_STATE *tstate, prof_st
     }
     stack->depth = depth;
     stack->stack[depth++] = (void*)((ULONG_PTR)thread_id);
+    /* the sampler thread sleeps in wall time, so use the wall clock */
+    sample_time = vmp_sample_time_ns(0);
+    memcpy(&stack->stack[depth], &sample_time, sizeof(sample_time));
     stack->count = 1;
     stack->marker = MARKER_STACKTRACE;
     ResumeThread(hThread);
@@ -226,7 +237,8 @@ long __stdcall vmprof_mainloop(void *arg)
         depth = vmprof_snapshot_thread(tstate->thread_id, tstate, stack);
         if (depth > 0) {
             vmp_write_all((char*)stack + offsetof(prof_stacktrace_s, marker),
-                          SIZEOF_PROF_STACKTRACE + depth * sizeof(void*));
+                          SIZEOF_PROF_STACKTRACE + depth * sizeof(void*) +
+                          sizeof(int64_t));
         }
         sampler_busy = 0;
     }
@@ -246,7 +258,7 @@ long __stdcall vmprof_mainloop(void *arg)
                 depth = vmprof_snapshot_thread(tstate->thread_ident, tstate, stack);
                 if (depth > 0) {
                     vmp_write_all((char*)stack + offsetof(prof_stacktrace_s, marker),
-                         depth * sizeof(void *) +
+                         depth * sizeof(void *) + sizeof(int64_t) +
                          sizeof(struct prof_stacktrace_s) -
                          offsetof(struct prof_stacktrace_s, marker));
                 }

vmprof/profiler.py

@@ -2,7 +2,7 @@
 import tempfile
 
 from vmprof.stats import Stats
-from vmprof.reader import _read_prof
+from vmprof.reader import _read_prof, DEFAULT_MAX_SAMPLE_GAP
 
 
 class VMProfError(Exception):
@@ -32,12 +32,18 @@ def __exit__(self, type, value, traceback):
         self.done = True
 
 
-def read_profile(prof_file):
+def read_profile(prof_file, max_sample_gap=DEFAULT_MAX_SAMPLE_GAP):
+    """ Read a profile file into a Stats object.
+
+    max_sample_gap caps, in seconds, how much time a single sample may
+    stand for when the timestamps show that timer signals were lost;
+    see vmprof.reader.LogReader.sample_weight.
+    """
     file_to_close = None
     if not hasattr(prof_file, 'read'):
         prof_file = file_to_close = open(str(prof_file), 'rb')
 
-    state = _read_prof(prof_file)
+    state = _read_prof(prof_file, max_sample_gap=max_sample_gap)
 
     if file_to_close:
         file_to_close.close()

vmprof/reader.py

@@ -26,11 +26,21 @@
 VERSION_MODE_AWARE = 4
 VERSION_DURATION = 5
 VERSION_TIMESTAMP = 6
+# every stack sample ends with a 64-bit nanosecond timestamp
+VERSION_SAMPLE_TIME = 7
 
 PROFILE_MEMORY = 1
 PROFILE_LINES = 2
 PROFILE_NATIVE = 4
 PROFILE_RPYTHON = 8
+PROFILE_REAL_TIME = 16
+
+# A sample is weighted by the time elapsed since the previous one, so that
+# timer signals the process never received (it was off the cpu, or the
+# signal was still pending) are still accounted for.  The gap is capped at
+# this many seconds: a process that was stopped in a debugger or suspended
+# with the laptop lid should not attribute minutes to a single frame.
+DEFAULT_MAX_SAMPLE_GAP = 1.0
 
 VMPROF_CODE_TAG = 1
 VMPROF_BLACKHOLE_TAG = 2
@@ -96,6 +106,8 @@ def __init__(self, fileobj, state):
         self.state = state
         self.word_size = None
         self.addr_size = None
+        # timestamp of the previous sample, see sample_weight()
+        self.last_sample_time = {}
         self.setup()
 
     def setup(self):
@@ -163,10 +175,12 @@ def read_header(self):
             s.profile_memory = (mode & PROFILE_MEMORY) != 0
             s.profile_lines = (mode & PROFILE_LINES) != 0
             s.profile_rpython = (mode & PROFILE_RPYTHON) != 0
+            s.profile_real_time = (mode & PROFILE_REAL_TIME) != 0
         else:
             s.profile_memory = s.version == VERSION_MEMORY
             s.profile_lines = False
             s.profile_rpython = False
+            s.profile_real_time = False
 
         lgt = ord(fileobj.read(1))
         s.interp_name = fileobj.read(lgt)
@@ -230,7 +244,43 @@ def read_addresses(self, count):
         return addrs
 
     def read_s64(self):
-        return struct.unpack('q', self.fileobj.read(8))[0]
+        return struct.unpack('<q', self.fileobj.read(8))[0]
+
+    def sample_weight(self, thread_id, sample_time):
+        """ How many timer periods this sample stands for.
+
+        Files older than VERSION_SAMPLE_TIME carry no timestamps and every
+        sample counts as one period.  Otherwise the sample is worth the
+        time since the previous sample divided by the period, at least 1,
+        and the gap is capped at state.max_sample_gap seconds; time beyond
+        the cap is summed up in state.lost_time.
+
+        In real time mode every registered thread receives its own signal
+        per period, so the previous sample is tracked per thread.  In cpu
+        time mode there is one process wide timer whose signal goes to the
+        running thread, and the timestamp is process cpu time, so the
+        previous sample is tracked process wide.
+        """
+        s = self.state
+        s.n_samples += 1
+        if sample_time is None or s.period <= 0:
+            s.expected_samples += 1
+            return 1
+        key = thread_id if s.profile_real_time else None
+        prev = self.last_sample_time.get(key)
+        if prev is None or sample_time > prev:
+            self.last_sample_time[key] = sample_time
+        if prev is None:
+            weight = 1.0
+        else:
+            gap = sample_time - prev
+            max_gap = s.max_sample_gap * 10**9
+            if gap > max_gap:
+                s.lost_time += (gap - max_gap) / 10.0**9
+                gap = max_gap
+            weight = max(gap / (s.period * 1000.0), 1.0)
+        s.expected_samples += weight
+        return weight
 
     def read_time_and_zone(self):
         return datetime.datetime.fromtimestamp(
@@ -274,12 +324,16 @@ def read_all(self):
                 trace = self.read_trace(depth)
                 thread_id = 0
                 mem_in_kb = 0
+                sample_time = None
                 if s.version >= VERSION_THREAD_ID:
                     thread_id = self.read_addr()
                 if s.profile_memory:
                     mem_in_kb = self.read_addr()
+                if s.version >= VERSION_SAMPLE_TIME:
+                    sample_time = self.read_s64()
                 trace.reverse()
-                self.add_trace(trace, 1, thread_id, mem_in_kb)
+                weight = self.sample_weight(thread_id, sample_time)
+                self.add_trace(trace, weight, thread_id, mem_in_kb)
             elif marker == MARKER_VIRTUAL_IP or marker == MARKER_NATIVE_SYMBOLS:
                 unique_id = self.read_addr()
                 name = self.read_string()
@@ -356,7 +410,7 @@ class ReaderState(object):
     pass
 
 class LogReaderState(ReaderState):
-    def __init__(self):
+    def __init__(self, max_sample_gap=DEFAULT_MAX_SAMPLE_GAP):
         self.virtual_ips = []
         self.profiles = []
         self.interp_name = None
@@ -365,14 +419,21 @@ def __init__(self):
         self.version = 0
         self.profile_memory = False
         self.profile_lines = False
+        self.profile_real_time = False
         self.meta = {}
         self.little_endian = True
-        self.period = 0
-
-def _read_prof(fileobj, virtual_ips_only=False):
+        self.period = 0 # microseconds
+        # see LogReader.sample_weight
+        self.max_sample_gap = max_sample_gap
+        self.n_samples = 0         # stack samples in the file
+        self.expected_samples = 0  # timer periods those samples stand for
+        self.lost_time = 0.0       # seconds cut off by max_sample_gap
+
+def _read_prof(fileobj, virtual_ips_only=False,
+               max_sample_gap=DEFAULT_MAX_SAMPLE_GAP):
     fileobj = gunzip(fileobj)
 
-    state = LogReaderState()
+    state = LogReaderState(max_sample_gap)
     reader = LogReader(fileobj, state)
     reader.read_all()