[sanitizer] Run Stack compression in background thread

[vitalybuka]

Summary

Depends on D114495.

[vitalybuka]

Abs

[dvyukov]

Do I understand it correctly that the point of splitting store and post_process was to move compression out of critical sections?
If yes, then it's not relevant with the background thread compression. It seems that the store/post_process logic can be simpler and we just Post to the semaphore whenever a block becomes full (or compress directly in tests).

[vitalybuka]

In D114498#3151544, @dvyukov wrote:

Do I understand it correctly that the point of splitting store and post_process was to move compression out of critical sections?
If yes, then it's not relevant with the background thread compression. It seems that the store/post_process logic can be simpler and we just Post to the semaphore whenever a block becomes full (or compress directly in tests).

Problem is that only tsan, asan, and memprof define DEFINE_REAL_PTHREAD_FUNCTIONS to have internal_start_thread. It should be trivial to define them of msan/lsan/hwasan.

Then there are platforms without internal_start_thread. we can fix some and disabled compression for the rest
WDYT?

[vitalybuka]

rebase

[dvyukov]

In D114498#3151982, @vitalybuka wrote:

In D114498#3151544, @dvyukov wrote:

Do I understand it correctly that the point of splitting store and post_process was to move compression out of critical sections?
If yes, then it's not relevant with the background thread compression. It seems that the store/post_process logic can be simpler and we just Post to the semaphore whenever a block becomes full (or compress directly in tests).

Problem is that only tsan, asan, and memprof define DEFINE_REAL_PTHREAD_FUNCTIONS to have internal_start_thread. It should be trivial to define them of msan/lsan/hwasan.

Then there are platforms without internal_start_thread. we can fix some and disabled compression for the rest
WDYT?

This sounds perfectly fine. If somebody will find async compression is critical for another sanitizer, they can add DEFINE_REAL_PTHREAD_FUNCTIONS.
Or maybe we could define real pthread_create in the sanitizer library (interceptor library may provide a way to query the "real" function w/o doing the rest)? I think overall it's better if sanitizer_common will provide fully working pieces rather than half-working. But I know it may be non-trivial. I tried to do stack printing in sanitizer_common and gave up because of all of _nolibc variants, etc.

But for other cases does it make sense to do compression inline instead (it happens episodically after start and compressing 8MB shouldn't take seconds). One environment w/o pthread is Go, it does not have/link pthread library.

But also note that background threads create problems with sandboxes which require programs to be single-threaded. At least this thread should be supported in the sandboxing callback. But I found that our sandbox2 does not call it after doing clone, so that may break (or turn into a latent bug since it executes relatively small amount of code after clone).

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

103

s/weak/strong/
weak can fail spuriously

[vitalybuka]

updated

[vitalybuka]

Probably simpler just fire-and-forget a thread when work is available
with just-in-case fallback to in-thread compression

[vitalybuka]

another version

[vitalybuka]

refrase comment

[dvyukov]

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

95

Shouldn't we call CompressStackStore first time before the sleep?
Will be surprising during debugging/looking at logs that compression is not happening after filling a block.

[vitalybuka]

no first sleep

[dvyukov]

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

95

Now it will call compress twice w/o sleep because of:

if (CompressStackStore())
  i = 0;

[vitalybuka]

fix loop

[dvyukov]

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

95

The loop looks ok now (probably, such non-standard loops are hard to analyze).

But we also need integration with __sanitizer_sandbox_on_notify to stop the thread (all threads) and something for Go which don't link in pthread.

... do we really need to do this in separate thread? how long does compression take?

[dvyukov]

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

95

Just noticed sandbox2 only calls __sanitizer_sandbox_on_notify if defined(THREAD_SANITIZER):
https://github.com/google/sandboxed-api/blob/aea8bb2ed0346b733cda00d4bd7d5234cb5f40fe/sandboxed_api/sandbox2/sanitizer.cc#L154-L156

while chromium does it properly:
https://source.chromium.org/chromium/chromium/src/+/main:sandbox/policy/linux/sandbox_linux.cc;l=173?q=__sanitizer_sandbox_on_notify&ss=chromium

[vitalybuka]

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

95

The loop looks ok now (probably, such non-standard loops are hard to analyze).

That's why I intentionally had the sleep even on the first run in the beginning.

But we also need integration with __sanitizer_sandbox_on_notify to stop the thread (all threads) and something for Go which don't link in pthread.

I am not sure I understand this comment. If we have empty thread internal_start_thread for SANITIZER_GO
so after this patch SANITIZER_GO will have no compression. So how GO is relevant here?

... do we really need to do this in separate thread? how long does compression take?

Actually not sure.
It takes ~60ms on my Intel(R) Xeon(R) CPU E5-2690 v4 @ 2.60GHz

In most cases it should called just a couple of times.
But on bad cases when the stack depot is 5GB+ it called frequently blocking everything.

I have real single threaded unittest which behave like this, and it regress about 20% with asan, without the thread. No regression with the thread.

If we go no-thread we need at least do that out of the hash table lock.

95

Just noticed sandbox2 only calls __sanitizer_sandbox_on_notify if defined(THREAD_SANITIZER):
https://github.com/google/sandboxed-api/blob/aea8bb2ed0346b733cda00d4bd7d5234cb5f40fe/sandboxed_api/sandbox2/sanitizer.cc#L154-L156

while chromium does it properly:
https://source.chromium.org/chromium/chromium/src/+/main:sandbox/policy/linux/sandbox_linux.cc;l=173?q=__sanitizer_sandbox_on_notify&ss=chromium

I guess I detected some test which fail like this.

[dvyukov]

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

95

I am not sure I understand this comment. If we have empty thread internal_start_thread for SANITIZER_GO

so after this patch SANITIZER_GO will have no compression. So how GO is relevant here?

Oh, I did not expect that there is a broken internal_start_thread defined for Go. Maybe it's better to enable it synchronously for Go then?

I have real single threaded unittest which behave like this, and it regress about 20% with asan

I guess we need the thread then.

[vitalybuka]

switch back to semaphore and remove sleep/heartbeat
wrap into class
support __sanitizer_sandbox_on_notify

[vitalybuka]

restore comment about negative flag

[vitalybuka]

fix typo in name

[dvyukov]

I like this version more than then previous 5 retries with sleeps in between.

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

154

It seems that Wait needs to be before CompressStackStore because NewWorkNotify always notifies newly started thread.
Or NewWorkNotify must not notify new thread.

218

This runs before fork, right? Why do we need to stop the thread before fork?

[vitalybuka]

improved loop

[vitalybuka]

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

154

Yes. I didn't this for readability.
Removed Post on start.

218

Because forked process will have no this thread, and we need to add some other stuff to solve restart it the new process or avoid deadlock in join of StackDepotStopBackgroundThread

[vitalybuka]

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

154

Removed Post on start.

Sorry, added draft comment before actual change.
I removed unnececary CompressStackStore by moving out some stuff into WaitForWork

[phosek]

We started seeing build failure on our builders after this change landed:

ld.lld: error: undefined symbol: __sanitizer::internal_start_thread(void* (*)(void*), void*)
>>> referenced by sanitizer_stackdepot.cpp:133 (compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp:133)
>>>               compiler-rt/lib/sanitizer_common/CMakeFiles/RTSanitizerCommonSymbolizer.x86_64.dir/sanitizer_stackdepot.cpp.obj:(__sanitizer::(anonymous namespace)::CompressThread::NewWorkNotify())

Would it be possible to take a look and revert the change if it cannot be resolved quickly?

[mstorsjo]

In D114498#3182110, @phosek wrote:

We started seeing build failure on our builders after this change landed:

ld.lld: error: undefined symbol: __sanitizer::internal_start_thread(void* (*)(void*), void*)
>>> referenced by sanitizer_stackdepot.cpp:133 (compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp:133)
>>>               compiler-rt/lib/sanitizer_common/CMakeFiles/RTSanitizerCommonSymbolizer.x86_64.dir/sanitizer_stackdepot.cpp.obj:(__sanitizer::(anonymous namespace)::CompressThread::NewWorkNotify())

Would it be possible to take a look and revert the change if it cannot be resolved quickly?

+1, this is also visible on https://lab.llvm.org/buildbot/#/builders/127.

I see an attempt to revert something to fix this in rG02997febe61597b7c3f215534c5adce851ba8963 by @browneee, but that seems to have reverted the wrong commit.

[phosek]

In D114498#3182468, @mstorsjo wrote:

In D114498#3182110, @phosek wrote:

We started seeing build failure on our builders after this change landed:

ld.lld: error: undefined symbol: __sanitizer::internal_start_thread(void* (*)(void*), void*)
>>> referenced by sanitizer_stackdepot.cpp:133 (compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp:133)
>>>               compiler-rt/lib/sanitizer_common/CMakeFiles/RTSanitizerCommonSymbolizer.x86_64.dir/sanitizer_stackdepot.cpp.obj:(__sanitizer::(anonymous namespace)::CompressThread::NewWorkNotify())

Would it be possible to take a look and revert the change if it cannot be resolved quickly?

+1, this is also visible on https://lab.llvm.org/buildbot/#/builders/127.

I see an attempt to revert something to fix this in rG02997febe61597b7c3f215534c5adce851ba8963 by @browneee, but that seems to have reverted the wrong commit.

I have reverted the change in rGa1e7f6280104.

[vitalybuka]

Thanks for revert.
Something wrong with the bot. Real error which connects breakage with this patch is in the step marked as green. I never looked into them.

[vitalybuka]

Windows bot accepted the updated revision. I don't know where to track Fuchsia.

[phosek]

In D114498#3183391, @vitalybuka wrote:

Windows bot accepted the updated revision. I don't know where to track Fuchsia.

I checked our builders and they're green as well. Thanks for looking into the issue.

[arphaman]

@vitalybuka Hi, this change caused some test failures on green dragon apple CI. This is the bot: https://green.lab.llvm.org/green/job/clang-stage1-RA/. The following tests are failing:

SanitizerCommon-asan-x86_64-Darwin.SanitizerCommon-asan-x86_64-Darwin.compress_stack_depot.cpp
SanitizerCommon-lsan-x86_64-Darwin.SanitizerCommon-lsan-x86_64-Darwin.compress_stack_depot.cpp
SanitizerCommon-tsan-x86_64-Darwin.SanitizerCommon-tsan-x86_64-Darwin.compress_stack_depot.cpp

I was able to fix the tests by also allowing SANITIZER_MAC target to call StackDepotStopBackgroundThread inside of sanitize_sanbox_on_notify. Do you think that would be the proper fix to commit?

[yln]

compiler-rt/lib/sanitizer_common/sanitizer_common_libcdep.cpp

207

I think this should not have been guarded on platforms here.

[yln]

Fixes the failing tests pointed out by Alex on Darwin: https://reviews.llvm.org/D116061

compiler-rt/lib/sanitizer_common/sanitizer_common_libcdep.cpp

 // run-time libraries.
 //===----------------------------------------------------------------------===//
 
 #include "sanitizer_allocator.h"
 #include "sanitizer_allocator_interface.h"
 #include "sanitizer_common.h"
 #include "sanitizer_flags.h"
 #include "sanitizer_procmaps.h"
-
+#include "sanitizer_stackdepot.h"
 
 namespace __sanitizer {
 
 #if (SANITIZER_LINUX || SANITIZER_NETBSD) && !SANITIZER_GO
 // Weak default implementation for when sanitizer_stackdepot is not linked in.
 SANITIZER_WEAK_ATTRIBUTE StackDepotStats StackDepotGetStats() { return {}; }
+SANITIZER_WEAK_ATTRIBUTE void StackDepotStopBackgroundThread() {}
 
 void *BackgroundThread(void *arg) {
   VPrintf(1, "%s: Started BackgroundThread\n", SanitizerToolName);
   const uptr hard_rss_limit_mb = common_flags()->hard_rss_limit_mb;
   const uptr soft_rss_limit_mb = common_flags()->soft_rss_limit_mb;
   const bool heap_profile = common_flags()->heap_profile;
   uptr prev_reported_rss = 0;
   uptr prev_reported_stack_depot_size = 0;

 }
 
 #endif  // !SANITIZER_FUCHSIA
 
 }  // namespace __sanitizer
 
 SANITIZER_INTERFACE_WEAK_DEF(void, __sanitizer_sandbox_on_notify,
                              __sanitizer_sandbox_arguments *args) {
+#if (SANITIZER_LINUX || SANITIZER_NETBSD) && !SANITIZER_GO
+  __sanitizer::StackDepotStopBackgroundThread();
+#endif
   __sanitizer::PlatformPrepareForSandboxing(args);
   if (__sanitizer::sandboxing_callback)
     __sanitizer::sandboxing_callback();
 }

compiler-rt/lib/sanitizer_common/sanitizer_fuchsia.cpp

bool GetRandom(void *buffer, uptr length, bool blocking) {
   _zx_cprng_draw(buffer, length);
   return true;
 }
 
 u32 GetNumberOfCPUs() { return zx_system_get_num_cpus(); }
 
 uptr GetRSS() { UNIMPLEMENTED(); }
 
+void *internal_start_thread(void *(*func)(void *arg), void *arg) { return 0; }
+void internal_join_thread(void *th) {}
+
 void InitializePlatformCommonFlags(CommonFlags *cf) {}
 
 }  // namespace __sanitizer
 
 using namespace __sanitizer;
 
 extern "C" {
 void __sanitizer_startup_hook(int argc, char **argv, char **envp,

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.h

 u32 StackDepotPut(StackTrace stack);
 StackDepotHandle StackDepotPut_WithHandle(StackTrace stack);
 // Retrieves a stored stack trace by the id.
 StackTrace StackDepotGet(u32 id);
 
 void StackDepotLockAll();
 void StackDepotUnlockAll();
 void StackDepotPrintAll();
+void StackDepotStopBackgroundThread();
 
 void StackDepotTestOnlyUnmap();
 
 } // namespace __sanitizer
 
 #endif // SANITIZER_STACKDEPOT_H

compiler-rt/lib/sanitizer_common/sanitizer_stackdepot.cpp

 //===-- sanitizer_stackdepot.cpp ------------------------------------------===//
 //
 // Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.
 // See https://llvm.org/LICENSE.txt for license information.
 // SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
 //
 //===----------------------------------------------------------------------===//
 //
 // This file is shared between AddressSanitizer and ThreadSanitizer
 // run-time libraries.
 //===----------------------------------------------------------------------===//
 
 #include "sanitizer_stackdepot.h"
 
+#include "sanitizer_atomic.h"
 #include "sanitizer_common.h"
 #include "sanitizer_hash.h"
+#include "sanitizer_mutex.h"
 #include "sanitizer_stack_store.h"
 #include "sanitizer_stackdepotbase.h"
 
 namespace __sanitizer {
 
 struct StackDepotNode {
   using hash_type = u64;
   hash_type stack_hash;

 
 uptr StackDepotNode::allocated() {
   return stackStore.Allocated() + useCounts.MemoryUsage();
 }
 
 static void CompressStackStore() {
   u64 start = MonotonicNanoTime();
   uptr diff = stackStore.Pack(static_cast<StackStore::Compression>(
-      common_flags()->compress_stack_depot));
+      Abs(common_flags()->compress_stack_depot)));
   if (!diff)
     return;
   u64 finish = MonotonicNanoTime();
   uptr total_before = theDepot.GetStats().allocated + diff;
   VPrintf(1, "%s: StackDepot released %zu KiB out of %zu KiB in %llu ms\n",
           SanitizerToolName, diff >> 10, total_before >> 10,
           (finish - start) / 1000000);
 }
 
+namespace {
+
+class CompressThread {
+ public:
+  constexpr CompressThread() = default;
+  void NewWorkNotify();
+  void Stop();
+  void LockAndStop() NO_THREAD_SAFETY_ANALYSIS;
+  void Unlock() NO_THREAD_SAFETY_ANALYSIS;
+
+ private:
+  enum class State {
+    NotStarted = 0,
+    Started,
+    Failed,
+    Stopped,
+  };
+
+  void Run();
+
+  bool WaitForWork() {
+    semaphore_.Wait();
+    return atomic_load(&run_, memory_order_acquire);
+  }
+
+  Semaphore semaphore_ = {};
+  StaticSpinMutex mutex_ = {};
+  State state_ GUARDED_BY(mutex_) = State::NotStarted;
+  void *thread_ GUARDED_BY(mutex_) = nullptr;
+  atomic_uint8_t run_ = {};
+};
+
+static CompressThread compress_thread;
+
+void CompressThread::NewWorkNotify() {
+  int compress = common_flags()->compress_stack_depot;
+  if (!compress)
+    return;
+  if (compress > 0 /* for testing or debugging */) {
+    SpinMutexLock l(&mutex_);
+    if (state_ == State::NotStarted) {
+      atomic_store(&run_, 1, memory_order_release);
+      CHECK_EQ(nullptr, thread_);
+      thread_ = internal_start_thread(
+          [](void *arg) -> void * {
+            reinterpret_cast<CompressThread *>(arg)->Run();
+            return nullptr;
+          },
+          this);
+      state_ = thread_ ? State::Started : State::Failed;
+    }
+    if (state_ == State::Started) {
+      semaphore_.Post();
+      return;
+    }
+  }
+  CompressStackStore();
+}
+
+void CompressThread::Run() {
+  VPrintf(1, "%s: StackDepot compression thread started\n", SanitizerToolName);
+  while (WaitForWork()) CompressStackStore();
+  VPrintf(1, "%s: StackDepot compression thread stopped\n", SanitizerToolName);
+}
+
+void CompressThread::Stop() {
+  void *t = nullptr;
+  {
+    SpinMutexLock l(&mutex_);
+    if (state_ != State::Started)
+      return;
+    state_ = State::Stopped;
+    CHECK_NE(nullptr, thread_);
+    t = thread_;
+    thread_ = nullptr;
+  }
+  atomic_store(&run_, 0, memory_order_release);
+  semaphore_.Post();
+  internal_join_thread(t);
+}
+
+void CompressThread::LockAndStop() {
+  mutex_.Lock();
+  if (state_ != State::Started)
+    return;
+  CHECK_NE(nullptr, thread_);
+
+  atomic_store(&run_, 0, memory_order_release);
+  semaphore_.Post();
+  internal_join_thread(thread_);
+  // Allow to restart after Unlock() if needed.
+  state_ = State::NotStarted;
+  thread_ = nullptr;
+}
+
+void CompressThread::Unlock() { mutex_.Unlock(); }
+
+}  // namespace
+
 void StackDepotNode::store(u32 id, const args_type &args, hash_type hash) {
   stack_hash = hash;
   uptr pack = 0;
   store_id = stackStore.Store(args, &pack);
-  if (pack && common_flags()->compress_stack_depot)
-    CompressStackStore();
+  if (LIKELY(!pack))
+    return;
+  compress_thread.NewWorkNotify();
 }
 
 StackDepotNode::args_type StackDepotNode::load(u32 id) const {
   if (!store_id)
     return {};
   return stackStore.Load(store_id);
 }
 
 StackDepotStats StackDepotGetStats() { return theDepot.GetStats(); }
 
 u32 StackDepotPut(StackTrace stack) { return theDepot.Put(stack); }
 
 StackDepotHandle StackDepotPut_WithHandle(StackTrace stack) {
   return StackDepotNode::get_handle(theDepot.Put(stack));
 }
 
 StackTrace StackDepotGet(u32 id) {
   return theDepot.Get(id);
 }
 
 void StackDepotLockAll() {
   theDepot.LockAll();
+  compress_thread.LockAndStop();
   stackStore.LockAll();
 }
 
 void StackDepotUnlockAll() {
   stackStore.UnlockAll();
+  compress_thread.Unlock();
   theDepot.UnlockAll();
 }
 
 void StackDepotPrintAll() {
 #if !SANITIZER_GO
   theDepot.PrintAll();
 #endif
 }
 
+void StackDepotStopBackgroundThread() { compress_thread.Stop(); }
+
 StackDepotHandle StackDepotNode::get_handle(u32 id) {
   return StackDepotHandle(&theDepot.nodes[id], id);
 }
 
 void StackDepotTestOnlyUnmap() {
   theDepot.TestOnlyUnmap();
   stackStore.TestOnlyUnmap();
 }
 
 } // namespace __sanitizer

compiler-rt/test/sanitizer_common/TestCases/compress_stack_depot.cpp

 // RUN: %clangxx %s -fsanitize-memory-track-origins=1 -o %t
 // RUN: %env_tool_opts="compress_stack_depot=0:malloc_context_size=128:verbosity=1" %run %t 2>&1 | FileCheck %s --implicit-check-not="StackDepot released"
-// RUN: %env_tool_opts="compress_stack_depot=1:malloc_context_size=128:verbosity=1" %run %t 2>&1 | FileCheck %s --check-prefixes=COMPRESS
-// RUN: %env_tool_opts="compress_stack_depot=2:malloc_context_size=128:verbosity=1" %run %t 2>&1 | FileCheck %s --check-prefixes=COMPRESS
+// RUN: %env_tool_opts="compress_stack_depot=-1:malloc_context_size=128:verbosity=1" %run %t 2>&1 | FileCheck %s --check-prefixes=COMPRESS
+// RUN: %env_tool_opts="compress_stack_depot=-2:malloc_context_size=128:verbosity=1" %run %t 2>&1 | FileCheck %s --check-prefixes=COMPRESS
+// RUN: %env_tool_opts="compress_stack_depot=1:malloc_context_size=128:verbosity=1" %run %t 2>&1 | FileCheck %s --check-prefixes=COMPRESS,THREAD
+// RUN: %env_tool_opts="compress_stack_depot=2:malloc_context_size=128:verbosity=1" %run %t 2>&1 | FileCheck %s --check-prefixes=COMPRESS,THREAD
 
 // Ubsan does not store stacks.
 // UNSUPPORTED: ubsan
 
 // FIXME: Fails for unknown reason.
 // UNSUPPORTED: s390x
 
+#include <sanitizer/common_interface_defs.h>
+
 #include <memory>
 
 __attribute__((noinline)) void a(unsigned v);
 __attribute__((noinline)) void b(unsigned v);
 
 std::unique_ptr<int[]> p;
 
 __attribute__((noinline)) void a(unsigned v) {
else
     a(v);
 }
 
 __attribute__((noinline)) void b(unsigned v) { return a(v); }
 
 int main(int argc, char *argv[]) {
   for (unsigned i = 0; i < 100000; ++i)
     a(i + (i << 16));
+
+  __sanitizer_sandbox_arguments args = {0};
+  __sanitizer_sandbox_on_notify(&args);
+
   return 0;
 }
 
+// THREAD: StackDepot compression thread started
 // COMPRESS: StackDepot released {{[0-9]+}}
+// THREAD: StackDepot compression thread stopped