diff --git a/src/_vmprof.c b/src/_vmprof.c index dcd9546..aa93232 100644 --- a/src/_vmprof.c +++ b/src/_vmprof.c @@ -422,9 +422,13 @@ static PyObject * insert_real_time_thread(PyObject *module, PyObject * args) { ssize_t thread_count; unsigned long thread_id = 0; + unsigned long native_id = 0; pthread_t th = pthread_self(); - if (!PyArg_ParseTuple(args, "|k", &thread_id)) { + /* thread_id is threading.Thread.ident (a pthread_t), native_id is + threading.Thread.native_id (the kernel thread id, linux only); both + default to the calling thread */ + if (!PyArg_ParseTuple(args, "|kk", &thread_id, &native_id)) { return NULL; } @@ -434,6 +438,8 @@ insert_real_time_thread(PyObject *module, PyObject * args) { #else th = (pthread_t) *(unsigned long *) &thread_id; #endif + } else { + native_id = (unsigned long) vmp_native_thread_id(); } if (!vmprof_is_enabled()) { @@ -447,7 +453,7 @@ insert_real_time_thread(PyObject *module, PyObject * args) { } vmprof_aquire_lock(); - thread_count = insert_thread(th, -1); + thread_count = insert_thread(th, (long) native_id, -1); vmprof_release_lock(); return PyLong_FromSsize_t(thread_count); diff --git a/src/vmprof_common.c b/src/vmprof_common.c index 6b968d4..5a64466 100644 --- a/src/vmprof_common.c +++ b/src/vmprof_common.c @@ -32,6 +32,8 @@ static long profile_interval_usec = 0; static int signal_type = SIGPROF; static int itimer_type = ITIMER_PROF; static pthread_t *threads = NULL; +/* kernel thread ids of 'threads', 0 when unknown; only used on linux */ +static long *thread_tids = NULL; static size_t threads_size = 0; static size_t thread_count = 0; static size_t threads_size_step = 8; @@ -225,20 +227,32 @@ ssize_t search_thread(pthread_t tid, ssize_t i) return -1; } -ssize_t insert_thread(pthread_t tid, ssize_t i) +long vmp_native_thread_id(void) +{ +#ifdef VMPROF_LINUX + return (long)syscall(SYS_gettid); +#else + return 0; +#endif +} + +ssize_t insert_thread(pthread_t tid, long native_id, ssize_t i) { assert(signal_type == SIGALRM); i = search_thread(tid, i); - if (i > 0) + if (i >= 0) return -1; if (thread_count == threads_size) { threads_size += threads_size_step; threads = realloc(threads, sizeof(pthread_t) * threads_size); - assert(threads != NULL); + thread_tids = realloc(thread_tids, sizeof(long) * threads_size); + assert(threads != NULL && thread_tids != NULL); memset(threads + thread_count, 0, sizeof(pthread_t) * threads_size_step); + memset(thread_tids + thread_count, 0, sizeof(long) * threads_size_step); } - threads[thread_count++] = tid; - return thread_count; + threads[thread_count] = tid; + thread_tids[thread_count] = native_id; + return ++thread_count; } ssize_t remove_thread(pthread_t tid, ssize_t i) @@ -251,8 +265,11 @@ ssize_t remove_thread(pthread_t tid, ssize_t i) i = search_thread(tid, i); if (i < 0) return -1; - threads[i] = threads[--thread_count]; + --thread_count; + threads[i] = threads[thread_count]; + thread_tids[i] = thread_tids[thread_count]; threads[thread_count] = 0; + thread_tids[thread_count] = 0; return thread_count; } @@ -263,23 +280,52 @@ ssize_t remove_threads(void) free(threads); threads = NULL; } + if (thread_tids != NULL) { + free(thread_tids); + thread_tids = NULL; + } thread_count = 0; threads_size = 0; return 0; } +/* Forwarding SIGALRM to the registered threads. + + Nothing removes a thread from 'threads' when it exits, and pthread_kill() + on an exited thread is undefined behaviour: on glibc the descriptor lives + on the thread's stack, which is freed as soon as a detached thread exits, + so it segfaults. On linux the signal is therefore sent with tgkill() to + the kernel thread id, which never touches user memory and just fails + with ESRCH for a thread that is gone. macOS and the BSDs validate the + thread inside pthread_kill() and return ESRCH themselves. Either way a + failed delivery drops the entry from the list. A thread registered + without a known kernel id falls back to pthread_kill(). */ +static int signal_thread(size_t i) +{ +#ifdef VMPROF_LINUX + if (thread_tids[i] != 0) { + int saved_errno = errno; + long res = syscall(SYS_tgkill, getpid(), (pid_t)thread_tids[i], SIGALRM); + int err = (res == 0) ? 0 : errno; + errno = saved_errno; + return err; + } +#endif + return pthread_kill(threads[i], SIGALRM); +} + int broadcast_signal_for_threads(void) { int done = 1; size_t i = 0; pthread_t self = pthread_self(); - pthread_t tid; while (i < thread_count) { - tid = threads[i]; - if (pthread_equal(tid, self)) { + if (pthread_equal(threads[i], self)) { done = 0; - } else if (pthread_kill(tid, SIGALRM)) { - remove_thread(tid, i); + } else if (signal_thread(i)) { + /* the last entry is moved into slot i, look at it next */ + remove_thread(threads[i], i); + continue; } i++; } diff --git a/src/vmprof_common.h b/src/vmprof_common.h index a1fe569..ec6b323 100644 --- a/src/vmprof_common.h +++ b/src/vmprof_common.h @@ -32,9 +32,13 @@ #ifdef VMPROF_UNIX ssize_t search_thread(pthread_t tid, ssize_t i); -ssize_t insert_thread(pthread_t tid, ssize_t i); +/* native_id: the kernel thread id (gettid) of 'tid' on linux, 0 if unknown + or on other platforms */ +ssize_t insert_thread(pthread_t tid, long native_id, ssize_t i); ssize_t remove_thread(pthread_t tid, ssize_t i); ssize_t remove_threads(void); +/* the kernel thread id of the calling thread on linux, 0 elsewhere */ +long vmp_native_thread_id(void); #endif diff --git a/src/vmprof_unix.c b/src/vmprof_unix.c index 6b46331..2aa2050 100644 --- a/src/vmprof_unix.c +++ b/src/vmprof_unix.c @@ -367,7 +367,7 @@ int vmprof_enable(int memory, int native, int real_time) if (memory && setup_rss() == -1) goto error; #if VMPROF_UNIX - if (real_time && insert_thread(pthread_self(), -1) == -1) + if (real_time && insert_thread(pthread_self(), vmp_native_thread_id(), -1) == -1) goto error; #endif if (install_pthread_atfork_hooks() == -1) diff --git a/vmprof/__init__.py b/vmprof/__init__.py index 7198530..98dfeda 100644 --- a/vmprof/__init__.py +++ b/vmprof/__init__.py @@ -108,7 +108,20 @@ def insert_real_time_thread(thread_id=0): Returns the number of registered threads, or -1 if we can't insert thread. Inserts the current thread if thread_id is not provided. """ - return _vmprof.insert_real_time_thread(thread_id) + if IS_PYPY: + return _vmprof.insert_real_time_thread(thread_id) + # On linux the signal is sent to the kernel thread id, which stays safe + # to use after the thread has exited (its pthread id does not). It is + # None until the thread has started running; the C side then records + # it from the thread's first sample. + native_id = 0 + if thread_id: + import threading + for thread in threading.enumerate(): + if thread.ident == thread_id: + native_id = thread.native_id or 0 + break + return _vmprof.insert_real_time_thread(thread_id, native_id) def remove_real_time_thread(thread_id=0): """ Removes a thread from the list of threads to be sampled in real time mode. diff --git a/vmprof/test/test_run.py b/vmprof/test/test_run.py index 21f199f..80754b7 100644 --- a/vmprof/test/test_run.py +++ b/vmprof/test/test_run.py @@ -323,7 +323,6 @@ def test_insert_other_real_time_thread(insert_foo, remove_bar): @pytest.mark.skipif("'__pypy__' in sys.builtin_module_names") @pytest.mark.skipif("sys.platform == 'win32'") -@pytest.mark.skip("seems to crash") def test_vmprof_real_time_many_threads(): import threading prof = vmprof.Profiler() @@ -349,6 +348,30 @@ def test_vmprof_real_time_many_threads(): assert bar_time_name in d +@pytest.mark.skipif("'__pypy__' in sys.builtin_module_names") +@pytest.mark.skipif("sys.platform == 'win32'") +def test_vmprof_real_time_threads_exit(): + # Registered threads that exit while profiling goes on used to leave + # stale pthread ids in the list, and forwarding SIGALRM to one of them + # segfaulted. + import threading + prof = vmprof.Profiler() + with prof.measure(period=0.001, real_time=True): + for _ in range(5): + threads = [threading.Thread(target=functime_foo, args=[0.01, True]) + for _ in range(20)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + # keep forwarding signals now that all of them are gone + functime_bar(0.1) + stats = prof.get_stats() + d = dict(stats.top_profile()) + assert foo_time_name in d + assert bar_time_name in d + + if GZIP: def test_gzip_problem(): tmpfile = tempfile.NamedTemporaryFile(delete=False)