8323807: Async UL: Add a stalling mode to async UL

Reviewed-by: dholmes, aboldtch

Author: jdksjolen

src/hotspot/share/logging/logAsyncWriter.cpp

@@ -26,45 +26,63 @@
 #include "logging/logConfiguration.hpp"
 #include "logging/logFileOutput.hpp"
 #include "logging/logFileStreamOutput.hpp"
-#include "logging/logHandle.hpp"
+#include "memory/allocation.hpp"
 #include "memory/resourceArea.hpp"
 #include "runtime/atomic.hpp"
-#include "runtime/os.inline.hpp"
-#include "runtime/globals.hpp"
 
+class AsyncLogWriter::Locker : public StackObj {
+  Thread*& _holder;
+  PlatformMonitor& _lock;
 
-class AsyncLogWriter::AsyncLogLocker : public StackObj {
-  static Thread* _holder;
 public:
-  static Thread* current_holder() { return _holder; }
-  AsyncLogLocker() {
-    assert(_instance != nullptr, "AsyncLogWriter::_lock is unavailable");
-    _instance->_lock.lock();
+  Locker(Thread*& holder, PlatformMonitor& lock)
+  : _holder(holder),
+    _lock(lock) {
+    _lock.lock();
     _holder = Thread::current_or_null();
   }
 
-  ~AsyncLogLocker() {
+  ~Locker() {
     assert(_holder == Thread::current_or_null(), "must be");
     _holder = nullptr;
-    _instance->_lock.unlock();
+    _lock.unlock();
+  }
+
+  void notify() {
+    _lock.notify();
   }
 
   void wait() {
     Thread* saved_holder = _holder;
     _holder = nullptr;
-    _instance->_lock.wait(0/* no timeout */);
+    _lock.wait(0 /* no timeout */);
     _holder = saved_holder;
   }
 };
 
-Thread* AsyncLogWriter::AsyncLogLocker::_holder = nullptr;
+class AsyncLogWriter::ProducerLocker : public Locker {
+  static Thread* _holder;
+public:
+  static Thread* current_holder() { return _holder; }
+  ProducerLocker() : Locker(_holder, _instance->_producer_lock) {}
+};
+
+class AsyncLogWriter::ConsumerLocker : public Locker {
+  static Thread* _holder;
+public:
+  static Thread* current_holder() { return _holder; }
+  ConsumerLocker() : Locker(_holder, _instance->_consumer_lock) {}
+};
+
+Thread* AsyncLogWriter::ProducerLocker::_holder = nullptr;
+Thread* AsyncLogWriter::ConsumerLocker::_holder = nullptr;
 
 // LogDecorator::None applies to 'constant initialization' because of its constexpr constructor.
 const LogDecorations& AsyncLogWriter::None = LogDecorations(LogLevel::Warning, LogTagSetMapping<LogTag::__NO_TAG>::tagset(),
                                       LogDecorators::None);
 
-bool AsyncLogWriter::Buffer::push_back(LogFileStreamOutput* output, const LogDecorations& decorations, const char* msg) {
-  const size_t len = strlen(msg);
+bool AsyncLogWriter::Buffer::push_back(LogFileStreamOutput* output, const LogDecorations& decorations, const char* msg, const size_t msg_len) {
+  const size_t len = msg_len;
   const size_t sz = Message::calc_size(len);
   const bool is_token = output == nullptr;
   // Always leave headroom for the flush token. Pushing a token must succeed.
@@ -80,7 +98,7 @@ bool AsyncLogWriter::Buffer::push_back(LogFileStreamOutput* output, const LogDec
 }
 
 void AsyncLogWriter::Buffer::push_flush_token() {
-  bool result = push_back(nullptr, AsyncLogWriter::None, "");
+  bool result = push_back(nullptr, AsyncLogWriter::None, "", 0);
   assert(result, "fail to enqueue the flush token.");
 }
 
@@ -89,22 +107,45 @@ void AsyncLogWriter::enqueue_locked(LogFileStreamOutput* output, const LogDecora
   // client should use "" instead.
   assert(msg != nullptr, "enqueuing a null message!");
 
-  if (!_buffer->push_back(output, decorations, msg)) {
-    bool p_created;
-    uint32_t* counter = _stats.put_if_absent(output, 0, &p_created);
-    *counter = *counter + 1;
-    return;
-  }
+  size_t msg_len = strlen(msg);
+  void* stalled_message = nullptr;
+  {
+    ConsumerLocker clocker;
+    if (_buffer->push_back(output, decorations, msg, msg_len)) {
+      _data_available = true;
+      clocker.notify();
+      return;
+    }
 
-  _data_available = true;
-  _lock.notify();
+    if (LogConfiguration::async_mode() == LogConfiguration::AsyncMode::Stall) {
+      size_t size = Message::calc_size(msg_len);
+      stalled_message = os::malloc(size, mtLogging);
+      if (stalled_message == nullptr) {
+        // Out of memory. We bail without any notice.
+        // Some other part of the system will probably fail later.
+        return;
+      }
+      _stalled_message = new (stalled_message) Message(output, decorations, msg, msg_len);
+      _data_available = true;
+      clocker.notify();
+      // Note: we still hold the producer lock so cannot race against other threads trying to log a message
+      while (_stalled_message != nullptr) {
+        clocker.wait();
+      }
+    } else {
+      bool p_created;
+      uint32_t* counter = _stats.put_if_absent(output, 0, &p_created);
+      *counter = *counter + 1;
+    }
+  } // ConsumerLocker out of scope
+  os::free(stalled_message);
 }
 
 // This function checks for cases where continuing with asynchronous logging may lead to stability issues, such as a deadlock.
 // If this returns false then we give up on logging asynchronously and do so synchronously instead.
 bool AsyncLogWriter::is_enqueue_allowed() {
   AsyncLogWriter* alw = AsyncLogWriter::instance();
-  Thread* holding_thread = AsyncLogWriter::AsyncLogLocker::current_holder();
+  Thread* holding_thread = AsyncLogWriter::ProducerLocker::current_holder();
   Thread* this_thread = Thread::current_or_null();
   if (this_thread == nullptr) {
     // The current thread is unattached.
@@ -142,7 +183,7 @@ bool AsyncLogWriter::enqueue(LogFileStreamOutput& output, const LogDecorations&
     return false;
   }
 
-  AsyncLogLocker locker;
+  ProducerLocker plocker;
 
 #ifdef ASSERT
   if (TestingAsyncLoggingDeathTest || TestingAsyncLoggingDeathTestNoCrash) {
@@ -162,17 +203,21 @@ bool AsyncLogWriter::enqueue(LogFileStreamOutput& output, LogMessageBuffer::Iter
   }
 
   // If we get here we know the AsyncLogWriter is initialized.
-  AsyncLogLocker locker;
+  ProducerLocker plocker;
   for (; !msg_iterator.is_at_end(); msg_iterator++) {
     AsyncLogWriter::instance()->enqueue_locked(&output, msg_iterator.decorations(), msg_iterator.message());
   }
   return true;
 }
 
 AsyncLogWriter::AsyncLogWriter()
-  : _flush_sem(0), _lock(), _data_available(false),
-    _initialized(false),
-    _stats() {
+: _flush_sem(0),
+  _producer_lock(),
+  _consumer_lock(),
+  _data_available(false),
+  _initialized(false),
+  _stats(),
+  _stalled_message(nullptr) {
 
   size_t size = AsyncLogBufferSize / 2;
   _buffer = new Buffer(size);
@@ -185,7 +230,7 @@ AsyncLogWriter::AsyncLogWriter()
   }
 }
 
-void AsyncLogWriter::write(AsyncLogMap<AnyObj::RESOURCE_AREA>& snapshot) {
+bool AsyncLogWriter::write(AsyncLogMap<AnyObj::RESOURCE_AREA>& snapshot) {
   int req = 0;
   auto it = _buffer_staging->iterator();
   while (it.hasNext()) {
@@ -213,20 +258,21 @@ void AsyncLogWriter::write(AsyncLogMap<AnyObj::RESOURCE_AREA>& snapshot) {
 
   if (req > 0) {
     assert(req == 1, "Only one token is allowed in queue. AsyncLogWriter::flush() is NOT MT-safe!");
-    _flush_sem.signal(req);
+    return true;
   }
+  return false;
 }
 
 void AsyncLogWriter::run() {
   while (true) {
     ResourceMark rm;
     AsyncLogMap<AnyObj::RESOURCE_AREA> snapshot;
     {
-      AsyncLogLocker locker;
-
+      ConsumerLocker clocker;
       while (!_data_available) {
-        locker.wait();
+        clocker.wait();
       }
+
       // Only doing a swap and statistics under the lock to
       // guarantee that I/O jobs don't block logsites.
       _buffer_staging->reset();
@@ -243,7 +289,23 @@ void AsyncLogWriter::run() {
       });
       _data_available = false;
     }
-    write(snapshot);
+
+    bool saw_flush_token = write(snapshot);
+
+    // Any stalled message must be written *after* the buffer has been written.
+    // This is because we try hard to output messages in program-order.
+    if (_stalled_message != nullptr) {
+      assert(LogConfiguration::async_mode() == LogConfiguration::AsyncMode::Stall, "must be");
+      ConsumerLocker clocker;
+      Message* m = (Message*)_stalled_message;
+      m->output()->write_blocking(m->decorations(), m->message());
+      _stalled_message = nullptr;
+      clocker.notify();
+    }
+
+    if (saw_flush_token) {
+      _flush_sem.signal(1);
+    }
   }
 }
 
@@ -281,19 +343,20 @@ AsyncLogWriter* AsyncLogWriter::instance() {
 void AsyncLogWriter::flush() {
   if (_instance != nullptr) {
     {
-      AsyncLogLocker locker;
+      ProducerLocker plocker;
+      ConsumerLocker clocker;
       // Push directly in-case we are at logical max capacity, as this must not get dropped.
       _instance->_buffer->push_flush_token();
       _instance->_data_available = true;
-      _instance->_lock.notify();
+      clocker.notify();
     }
 
     _instance->_flush_sem.wait();
   }
 }
 
 AsyncLogWriter::BufferUpdater::BufferUpdater(size_t newsize) {
-  AsyncLogLocker locker;
+  ConsumerLocker clocker;
   auto p = AsyncLogWriter::_instance;
 
   _buf1 = p->_buffer;
@@ -307,7 +370,7 @@ AsyncLogWriter::BufferUpdater::~BufferUpdater() {
   auto p = AsyncLogWriter::_instance;
 
   {
-    AsyncLogLocker locker;
+    ConsumerLocker clocker;
 
     delete p->_buffer;
     delete p->_buffer_staging;

src/hotspot/share/logging/logAsyncWriter.hpp

@@ -29,6 +29,7 @@
 #include "logging/logMessageBuffer.hpp"
 #include "memory/allocation.hpp"
 #include "runtime/mutex.hpp"
+#include "runtime/os.inline.hpp"
 #include "runtime/nonJavaThread.hpp"
 #include "runtime/semaphore.hpp"
 #include "utilities/resourceHash.hpp"
@@ -59,7 +60,9 @@ class LogFileStreamOutput;
 class AsyncLogWriter : public NonJavaThread {
   friend class AsyncLogTest;
   friend class AsyncLogTest_logBuffer_vm_Test;
-  class AsyncLogLocker;
+  class Locker;
+  class ProducerLocker;
+  class ConsumerLocker;
 
   // account for dropped messages
   template <AnyObj::allocation_type ALLOC_TYPE>
@@ -125,7 +128,7 @@ class AsyncLogWriter : public NonJavaThread {
     }
 
     void push_flush_token();
-    bool push_back(LogFileStreamOutput* output, const LogDecorations& decorations, const char* msg);
+    bool push_back(LogFileStreamOutput* output, const LogDecorations& decorations, const char* msg, const size_t msg_len);
 
     void reset() {
       // Ensure _pos is Message-aligned
@@ -159,20 +162,31 @@ class AsyncLogWriter : public NonJavaThread {
   static AsyncLogWriter* _instance;
   Semaphore _flush_sem;
   // Can't use a Monitor here as we need a low-level API that can be used without Thread::current().
-  PlatformMonitor _lock;
+  // Producers take both locks in the order producer lock and then consumer lock.
+  // The consumer protects the buffers and performs all communication between producer and consumer via wait/notify.
+  // This allows a producer to await progress from the consumer thread (by only releasing the producer lock)), whilst preventing all other producers from progressing.
+  PlatformMonitor _producer_lock;
+  PlatformMonitor _consumer_lock;
   bool _data_available;
+  // _initialized is set to true if the constructor succeeds
   volatile bool _initialized;
   AsyncLogMap<AnyObj::C_HEAP> _stats;
 
   // ping-pong buffers
   Buffer* _buffer;
   Buffer* _buffer_staging;
 
+  // Stalled message
+  // Stalling is implemented by the producer writing to _stalled_message, notifying the consumer lock and releasing it.
+  // The consumer will then write all of the current buffers' content and then write the stalled message, at the end notifying the consumer lock and releasing it for the
+  // owning producer thread of the stalled message. This thread will finally release both locks in order, allowing for other producers to continue.
+  volatile Message* _stalled_message;
+
   static const LogDecorations& None;
 
   AsyncLogWriter();
   void enqueue_locked(LogFileStreamOutput* output, const LogDecorations& decorations, const char* msg);
-  void write(AsyncLogMap<AnyObj::RESOURCE_AREA>& snapshot);
+  bool write(AsyncLogMap<AnyObj::RESOURCE_AREA>& snapshot);
   void run() override;
   void pre_run() override {
     NonJavaThread::pre_run();

src/hotspot/share/logging/logConfiguration.cpp

@@ -636,10 +636,15 @@ void LogConfiguration::print_command_line_help(outputStream* out) {
   out->cr();
 
   out->print_cr("Asynchronous logging (off by default):");
-  out->print_cr(" -Xlog:async");
+  out->print_cr(" -Xlog:async[:[mode]]");
   out->print_cr("  All log messages are written to an intermediate buffer first and will then be flushed"
                 " to the corresponding log outputs by a standalone thread. Write operations at logsites are"
                 " guaranteed non-blocking.");
+  out->print_cr(" A mode, either 'drop' or 'stall', may be provided. If 'drop' is provided then"
+                " messages will be dropped if there is no room in the intermediate buffer."
+                " If 'stall' is provided then the log operation will wait for room to be made by the output thread, without dropping any messages."
+                " The default mode is 'drop'.");
+
   out->cr();
 
   out->print_cr("Some examples:");
@@ -715,4 +720,20 @@ void LogConfiguration::notify_update_listeners() {
   }
 }
 
-bool LogConfiguration::_async_mode = false;
+LogConfiguration::AsyncMode LogConfiguration::_async_mode = AsyncMode::Off;
+
+bool LogConfiguration::parse_async_argument(const char* async_tail) {
+  bool ret = true;
+  if (*async_tail == '\0') {
+    // Default is to drop.
+    LogConfiguration::set_async_mode(LogConfiguration::AsyncMode::Drop);
+  } else if (strcmp(async_tail, ":stall") == 0) {
+    LogConfiguration::set_async_mode(LogConfiguration::AsyncMode::Stall);
+  } else if (strcmp(async_tail, ":drop") == 0) {
+    LogConfiguration::set_async_mode(LogConfiguration::AsyncMode::Drop);
+  } else {
+    // User provided unknown async option
+    ret = false;
+  }
+  return ret;
+}