diff --git a/src/hotspot/os/windows/os_windows.cpp b/src/hotspot/os/windows/os_windows.cpp
index b9f48898d9d5d..7db1b9b7039a0 100644
--- a/src/hotspot/os/windows/os_windows.cpp
+++ b/src/hotspot/os/windows/os_windows.cpp
@@ -714,6 +714,7 @@ bool os::create_thread(Thread* thread, ThreadType thr_type,
     case os::vm_thread:
     case os::pgc_thread:
     case os::cgc_thread:
+    case os::asynclog_thread:
     case os::watcher_thread:
       if (VMThreadStackSize > 0) stack_size = (size_t)(VMThreadStackSize * K);
       break;
diff --git a/src/hotspot/share/logging/logAsyncWriter.cpp b/src/hotspot/share/logging/logAsyncWriter.cpp
new file mode 100644
index 0000000000000..b3b61f8a7c099
--- /dev/null
+++ b/src/hotspot/share/logging/logAsyncWriter.cpp
@@ -0,0 +1,194 @@
+/*
+ * Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ *
+ */
+#include "precompiled.hpp"
+#include "logging/logAsyncWriter.hpp"
+#include "logging/logConfiguration.hpp"
+#include "logging/logFileOutput.hpp"
+#include "logging/logHandle.hpp"
+#include "runtime/atomic.hpp"
+
+Semaphore AsyncLogWriter::_sem(0);
+Semaphore AsyncLogWriter::_io_sem(1);
+
+class AsyncLogLocker : public StackObj {
+ private:
+  static Semaphore _lock;
+ public:
+  AsyncLogLocker() {
+    _lock.wait();
+  }
+
+  ~AsyncLogLocker() {
+    _lock.signal();
+  }
+};
+
+Semaphore AsyncLogLocker::_lock(1);
+
+void AsyncLogWriter::enqueue_locked(const AsyncLogMessage& msg) {
+  if (_buffer.size() >= _buffer_max_size)  {
+    bool p_created;
+    uint32_t* counter = _stats.add_if_absent(msg.output(), 0, &p_created);
+    *counter = *counter + 1;
+    // drop the enqueueing message.
+    return;
+  }
+
+  assert(_buffer.size() < _buffer_max_size, "_buffer is over-sized.");
+  _buffer.push_back(msg);
+  _sem.signal();
+}
+
+void AsyncLogWriter::enqueue(LogFileOutput& output, const LogDecorations& decorations, const char* msg) {
+  AsyncLogMessage m(output, decorations, os::strdup(msg));
+
+  { // critical area
+    AsyncLogLocker lock;
+    enqueue_locked(m);
+  }
+}
+
+// LogMessageBuffer consists of a multiple-part/multiple-line messsage.
+// The lock here guarantees its integrity.
+void AsyncLogWriter::enqueue(LogFileOutput& output, LogMessageBuffer::Iterator msg_iterator) {
+  AsyncLogLocker lock;
+
+  for (; !msg_iterator.is_at_end(); msg_iterator++) {
+    AsyncLogMessage m(output, msg_iterator.decorations(), os::strdup(msg_iterator.message()));
+    enqueue_locked(m);
+  }
+}
+
+AsyncLogWriter::AsyncLogWriter()
+  : _initialized(false),
+    _stats(17 /*table_size*/) {
+  if (os::create_thread(this, os::asynclog_thread)) {
+    _initialized = true;
+  } else {
+    log_warning(logging, thread)("AsyncLogging failed to create thread. Falling back to synchronous logging.");
+  }
+
+  log_info(logging)("The maximum entries of AsyncLogBuffer: " SIZE_FORMAT ", estimated memory use: " SIZE_FORMAT " bytes",
+                    _buffer_max_size, AsyncLogBufferSize);
+}
+
+class AsyncLogMapIterator {
+  AsyncLogBuffer& _logs;
+
+ public:
+  AsyncLogMapIterator(AsyncLogBuffer& logs) :_logs(logs) {}
+  bool do_entry(LogFileOutput* output, uint32_t* counter) {
+    using none = LogTagSetMapping<LogTag::__NO_TAG>;
+
+    if (*counter > 0) {
+      LogDecorations decorations(LogLevel::Warning, none::tagset(), output->decorators());
+      stringStream ss;
+      ss.print(UINT32_FORMAT_W(6) " messages dropped due to async logging", *counter);
+      AsyncLogMessage msg(*output, decorations, ss.as_string(true /*c_heap*/));
+      _logs.push_back(msg);
+      *counter = 0;
+    }
+
+    return true;
+  }
+};
+
+void AsyncLogWriter::write() {
+  // Use kind of copy-and-swap idiom here.
+  // Empty 'logs' swaps the content with _buffer.
+  // Along with logs destruction, all processed messages are deleted.
+  //
+  // The operation 'pop_all()' is done in O(1). All I/O jobs are then performed without
+  // lock protection. This guarantees I/O jobs don't block logsites.
+  AsyncLogBuffer logs;
+  bool own_io = false;
+
+  { // critical region
+    AsyncLogLocker lock;
+
+    _buffer.pop_all(&logs);
+    // append meta-messages of dropped counters
+    AsyncLogMapIterator dropped_counters_iter(logs);
+    _stats.iterate(&dropped_counters_iter);
+    own_io = _io_sem.trywait();
+  }
+
+  LinkedListIterator<AsyncLogMessage> it(logs.head());
+  if (!own_io) {
+    _io_sem.wait();
+  }
+
+  while (!it.is_empty()) {
+    AsyncLogMessage* e = it.next();
+    char* msg = e->message();
+
+    if (msg != nullptr) {
+      e->output()->write_blocking(e->decorations(), msg);
+      os::free(msg);
+    }
+  }
+  _io_sem.signal();
+}
+
+void AsyncLogWriter::run() {
+  while (true) {
+    // The value of a semphore cannot be negative. Therefore, the current thread falls asleep
+    // when its value is zero. It will be waken up when new messages are enqueued.
+    _sem.wait();
+    write();
+  }
+}
+
+AsyncLogWriter* AsyncLogWriter::_instance = nullptr;
+
+void AsyncLogWriter::initialize() {
+  if (!LogConfiguration::is_async_mode()) return;
+
+  assert(_instance == nullptr, "initialize() should only be invoked once.");
+
+  AsyncLogWriter* self = new AsyncLogWriter();
+  if (self->_initialized) {
+    Atomic::release_store_fence(&AsyncLogWriter::_instance, self);
+    // All readers of _instance after the fence see non-NULL.
+    // We use LogOutputList's RCU counters to ensure all synchronous logsites have completed.
+    // After that, we start AsyncLog Thread and it exclusively takes over all logging I/O.
+    for (LogTagSet* ts = LogTagSet::first(); ts != NULL; ts = ts->next()) {
+      ts->wait_until_no_readers();
+    }
+    os::start_thread(self);
+    log_debug(logging, thread)("Async logging thread started.");
+  }
+}
+
+AsyncLogWriter* AsyncLogWriter::instance() {
+  return _instance;
+}
+
+// write() acquires and releases _io_sem even _buffer is empty.
+// This guarantees all logging I/O of dequeued messages are done when it returns.
+void AsyncLogWriter::flush() {
+  if (_instance != nullptr) {
+    _instance->write();
+  }
+}
diff --git a/src/hotspot/share/logging/logAsyncWriter.hpp b/src/hotspot/share/logging/logAsyncWriter.hpp
new file mode 100644
index 0000000000000..736a957b218c3
--- /dev/null
+++ b/src/hotspot/share/logging/logAsyncWriter.hpp
@@ -0,0 +1,176 @@
+/*
+ * Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ *
+ */
+#ifndef SHARE_LOGGING_LOGASYNCWRITER_HPP
+#define SHARE_LOGGING_LOGASYNCWRITER_HPP
+#include "logging/log.hpp"
+#include "logging/logDecorations.hpp"
+#include "logging/logFileOutput.hpp"
+#include "logging/logMessageBuffer.hpp"
+#include "memory/resourceArea.hpp"
+#include "runtime/nonJavaThread.hpp"
+#include "utilities/hashtable.hpp"
+#include "utilities/linkedlist.hpp"
+
+template <typename E, MEMFLAGS F>
+class LinkedListDeque : private LinkedListImpl<E, ResourceObj::C_HEAP, F> {
+ private:
+  LinkedListNode<E>* _tail;
+  size_t _size;
+
+ public:
+  LinkedListDeque() : _tail(NULL), _size(0) {}
+  void push_back(const E& e) {
+    if (!_tail) {
+      _tail = this->add(e);
+    } else {
+      _tail = this->insert_after(e, _tail);
+    }
+
+    ++_size;
+  }
+
+  // pop all elements to logs.
+  void pop_all(LinkedList<E>* logs) {
+    logs->move(static_cast<LinkedList<E>* >(this));
+    _tail = NULL;
+    _size = 0;
+  }
+
+  void pop_all(LinkedListDeque<E, F>* logs) {
+    logs->_size = _size;
+    logs->_tail = _tail;
+    pop_all(static_cast<LinkedList<E>* >(logs));
+  }
+
+  void pop_front() {
+    LinkedListNode<E>* h = this->unlink_head();
+    if (h == _tail) {
+      _tail = NULL;
+    }
+
+    if (h != NULL) {
+      --_size;
+      this->delete_node(h);
+    }
+  }
+
+  size_t size() const { return _size; }
+
+  const E* front() const {
+    return this->_head == NULL ? NULL : this->_head->peek();
+  }
+
+  const E* back() const {
+    return _tail == NULL ? NULL : _tail->peek();
+  }
+
+  LinkedListNode<E>* head() const {
+    return this->_head;
+  }
+};
+
+class AsyncLogMessage {
+  LogFileOutput& _output;
+  const LogDecorations _decorations;
+  char* _message;
+
+public:
+  AsyncLogMessage(LogFileOutput& output, const LogDecorations& decorations, char* msg)
+    : _output(output), _decorations(decorations), _message(msg) {}
+
+  // placeholder for LinkedListImpl.
+  bool equals(const AsyncLogMessage& o) const { return false; }
+
+  LogFileOutput* output() const { return &_output; }
+  const LogDecorations& decorations() const { return _decorations; }
+  char* message() const { return _message; }
+};
+
+typedef LinkedListDeque<AsyncLogMessage, mtLogging> AsyncLogBuffer;
+typedef KVHashtable<LogFileOutput*, uint32_t, mtLogging> AsyncLogMap;
+
+//
+// ASYNC LOGGING SUPPORT
+//
+// Summary:
+// Async Logging is working on the basis of singleton AsyncLogWriter, which manages an intermediate buffer and a flushing thread.
+//
+// Interface:
+//
+// initialize() is called once when JVM is initialized. It creates and initializes the singleton instance of AsyncLogWriter.
+// Once async logging is established, there's no way to turn it off.
+//
+// instance() is MT-safe and returns the pointer of the singleton instance if and only if async logging is enabled and has well
+// initialized. Clients can use its return value to determine async logging is established or not.
+//
+// The basic operation of AsyncLogWriter is enqueue(). 2 overloading versions of it are provided to match LogOutput::write().
+// They are both MT-safe and non-blocking. Derived classes of LogOutput can invoke the corresponding enqueue() in write() and
+// return 0. AsyncLogWriter is responsible of copying neccessary data.
+//
+// The static member function flush() is designated to flush out all pending messages when JVM is terminating.
+// In normal JVM termination, flush() is invoked in LogConfiguration::finalize(). flush() is MT-safe and can be invoked arbitrary
+// times. It is no-op if async logging is not established.
+//
+class AsyncLogWriter : public NonJavaThread {
+  static AsyncLogWriter* _instance;
+  // _sem is a semaphore whose value denotes how many messages have been enqueued.
+  // It decreases in AsyncLogWriter::run()
+  static Semaphore _sem;
+  // A lock of IO
+  static Semaphore _io_sem;
+
+  volatile bool _initialized;
+  AsyncLogMap _stats; // statistics for dropped messages
+  AsyncLogBuffer _buffer;
+
+  // The memory use of each AsyncLogMessage (payload) consists of itself and a variable-length c-str message.
+  // A regular logging message is smaller than vwrite_buffer_size, which is defined in logtagset.cpp
+  const size_t _buffer_max_size = {AsyncLogBufferSize / (sizeof(AsyncLogMessage) + vwrite_buffer_size)};
+
+  AsyncLogWriter();
+  void enqueue_locked(const AsyncLogMessage& msg);
+  void write();
+  void run() override;
+  void pre_run() override {
+    NonJavaThread::pre_run();
+    log_debug(logging, thread)("starting AsyncLog Thread tid = " INTX_FORMAT, os::current_thread_id());
+  }
+  char* name() const override { return (char*)"AsyncLog Thread"; }
+  bool is_Named_thread() const override { return true; }
+  void print_on(outputStream* st) const override {
+    st->print("\"%s\" ", name());
+    Thread::print_on(st);
+    st->cr();
+  }
+
+ public:
+  void enqueue(LogFileOutput& output, const LogDecorations& decorations, const char* msg);
+  void enqueue(LogFileOutput& output, LogMessageBuffer::Iterator msg_iterator);
+
+  static AsyncLogWriter* instance();
+  static void initialize();
+  static void flush();
+};
+
+#endif // SHARE_LOGGING_LOGASYNCWRITER_HPP
diff --git a/src/hotspot/share/logging/logConfiguration.cpp b/src/hotspot/share/logging/logConfiguration.cpp
index 2fddbe8bedddb..5e2f786a65730 100644
--- a/src/hotspot/share/logging/logConfiguration.cpp
+++ b/src/hotspot/share/logging/logConfiguration.cpp
@@ -24,6 +24,7 @@
 #include "precompiled.hpp"
 #include "jvm.h"
 #include "logging/log.hpp"
+#include "logging/logAsyncWriter.hpp"
 #include "logging/logConfiguration.hpp"
 #include "logging/logDecorations.hpp"
 #include "logging/logDecorators.hpp"
@@ -261,6 +262,10 @@ void LogConfiguration::configure_output(size_t idx, const LogSelectionList& sele
   }
 
   if (!enabled && idx > 1) {
+    // User may disable a logOuput like this:
+    // LogConfiguration::parse_log_arguments(filename, "all=off", "", "", &stream);
+    // Just be conservative. Flush them all before deleting idx.
+    AsyncLogWriter::flush();
     // Output is unused and should be removed, unless it is stdout/stderr (idx < 2)
     delete_output(idx);
     return;
@@ -278,6 +283,12 @@ void LogConfiguration::disable_outputs() {
     ts->disable_outputs();
   }
 
+  // Handle jcmd VM.log disable
+  // ts->disable_outputs() above has deleted output_list with RCU synchronization.
+  // Therefore, no new logging entry can enter AsyncLog buffer for the time being.
+  // flush pending entries before LogOutput instances die.
+  AsyncLogWriter::flush();
+
   while (idx > 0) {
     LogOutput* out = _outputs[--idx];
     // Delete the output unless stdout or stderr (idx 0 or 1)
@@ -545,6 +556,12 @@ void LogConfiguration::print_command_line_help(outputStream* out) {
                                     " If set to 0, log rotation is disabled."
                                     " This will cause existing log files to be overwritten.");
   out->cr();
+  out->print_cr("\nAsynchronous logging (off by default):");
+  out->print_cr(" -Xlog:async");
+  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->cr();
 
   out->print_cr("Some examples:");
   out->print_cr(" -Xlog");
@@ -587,6 +604,10 @@ void LogConfiguration::print_command_line_help(outputStream* out) {
   out->print_cr(" -Xlog:disable -Xlog:safepoint=trace:safepointtrace.txt");
   out->print_cr("\t Turn off all logging, including warnings and errors,");
   out->print_cr("\t and then enable messages tagged with 'safepoint' up to 'trace' level to file 'safepointtrace.txt'.");
+
+  out->print_cr(" -Xlog:async -Xlog:gc=debug:file=gc.log -Xlog:safepoint=trace");
+  out->print_cr("\t Write logs asynchronously. Enable messages tagged with 'safepoint' up to 'trace' level to stdout ");
+  out->print_cr("\t and messages tagged with 'gc' up to 'debug' level to file 'gc.log'.");
 }
 
 void LogConfiguration::rotate_all_outputs() {
@@ -613,3 +634,5 @@ void LogConfiguration::notify_update_listeners() {
     _listener_callbacks[i]();
   }
 }
+
+bool LogConfiguration::_async_mode = false;
diff --git a/src/hotspot/share/logging/logConfiguration.hpp b/src/hotspot/share/logging/logConfiguration.hpp
index 34c0c07b76a76..542aaa2433bc0 100644
--- a/src/hotspot/share/logging/logConfiguration.hpp
+++ b/src/hotspot/share/logging/logConfiguration.hpp
@@ -58,6 +58,7 @@ class LogConfiguration : public AllStatic {
 
   static UpdateListenerFunction*    _listener_callbacks;
   static size_t                     _n_listener_callbacks;
+  static bool                       _async_mode;
 
   // Create a new output. Returns NULL if failed.
   static LogOutput* new_output(const char* name, const char* options, outputStream* errstream);
@@ -123,6 +124,11 @@ class LogConfiguration : public AllStatic {
 
   // Rotates all LogOutput
   static void rotate_all_outputs();
+
+  static bool is_async_mode() { return _async_mode; }
+  static void set_async_mode(bool value) {
+    _async_mode = value;
+  }
 };
 
 #endif // SHARE_LOGGING_LOGCONFIGURATION_HPP
diff --git a/src/hotspot/share/logging/logDecorators.hpp b/src/hotspot/share/logging/logDecorators.hpp
index c6877b5867a4a..cddd9c3bf85d1 100644
--- a/src/hotspot/share/logging/logDecorators.hpp
+++ b/src/hotspot/share/logging/logDecorators.hpp
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2019, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2021, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -84,7 +84,7 @@ class LogDecorators {
   static const LogDecorators None;
 
   LogDecorators() : _decorators(DefaultDecoratorsMask) {
-  };
+  }
 
   void clear() {
     _decorators = 0;
diff --git a/src/hotspot/share/logging/logFileOutput.cpp b/src/hotspot/share/logging/logFileOutput.cpp
index 22f01186b8899..1728317e0249c 100644
--- a/src/hotspot/share/logging/logFileOutput.cpp
+++ b/src/hotspot/share/logging/logFileOutput.cpp
@@ -24,6 +24,7 @@
 #include "precompiled.hpp"
 #include "jvm.h"
 #include "logging/log.hpp"
+#include "logging/logAsyncWriter.hpp"
 #include "logging/logConfiguration.hpp"
 #include "logging/logFileOutput.hpp"
 #include "memory/allocation.inline.hpp"
@@ -284,12 +285,7 @@ bool LogFileOutput::initialize(const char* options, outputStream* errstream) {
   return true;
 }
 
-int LogFileOutput::write(const LogDecorations& decorations, const char* msg) {
-  if (_stream == NULL) {
-    // An error has occurred with this output, avoid writing to it.
-    return 0;
-  }
-
+int LogFileOutput::write_blocking(const LogDecorations& decorations, const char* msg) {
   _rotation_semaphore.wait();
   int written = LogFileStreamOutput::write(decorations, msg);
   if (written > 0) {
@@ -304,12 +300,33 @@ int LogFileOutput::write(const LogDecorations& decorations, const char* msg) {
   return written;
 }
 
+int LogFileOutput::write(const LogDecorations& decorations, const char* msg) {
+  if (_stream == NULL) {
+    // An error has occurred with this output, avoid writing to it.
+    return 0;
+  }
+
+  AsyncLogWriter* aio_writer = AsyncLogWriter::instance();
+  if (aio_writer != nullptr) {
+    aio_writer->enqueue(*this, decorations, msg);
+    return 0;
+  }
+
+  return write_blocking(decorations, msg);
+}
+
 int LogFileOutput::write(LogMessageBuffer::Iterator msg_iterator) {
   if (_stream == NULL) {
     // An error has occurred with this output, avoid writing to it.
     return 0;
   }
 
+  AsyncLogWriter* aio_writer = AsyncLogWriter::instance();
+  if (aio_writer != nullptr) {
+    aio_writer->enqueue(*this, msg_iterator);
+    return 0;
+  }
+
   _rotation_semaphore.wait();
   int written = LogFileStreamOutput::write(msg_iterator);
   if (written > 0) {
@@ -461,7 +478,8 @@ void LogFileOutput::describe(outputStream *out) {
   LogOutput::describe(out);
   out->print(" ");
 
-  out->print("filecount=%u,filesize=" SIZE_FORMAT "%s", _file_count,
+  out->print("filecount=%u,filesize=" SIZE_FORMAT "%s,async=%s", _file_count,
              byte_size_in_proper_unit(_rotate_size),
-             proper_unit_for_byte_size(_rotate_size));
+             proper_unit_for_byte_size(_rotate_size),
+             LogConfiguration::is_async_mode() ? "true" : "false");
 }
diff --git a/src/hotspot/share/logging/logFileOutput.hpp b/src/hotspot/share/logging/logFileOutput.hpp
index 5301816313d08..9b22969368ed5 100644
--- a/src/hotspot/share/logging/logFileOutput.hpp
+++ b/src/hotspot/share/logging/logFileOutput.hpp
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2019, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2021, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -85,6 +85,7 @@ class LogFileOutput : public LogFileStreamOutput {
   virtual bool initialize(const char* options, outputStream* errstream);
   virtual int write(const LogDecorations& decorations, const char* msg);
   virtual int write(LogMessageBuffer::Iterator msg_iterator);
+  int write_blocking(const LogDecorations& decorations, const char* msg);
   virtual void force_rotate();
   virtual void describe(outputStream* out);
 
diff --git a/src/hotspot/share/logging/logOutputList.hpp b/src/hotspot/share/logging/logOutputList.hpp
index d02f50cc8b2c2..e2179cc68ea9a 100644
--- a/src/hotspot/share/logging/logOutputList.hpp
+++ b/src/hotspot/share/logging/logOutputList.hpp
@@ -63,7 +63,6 @@ class LogOutputList {
   // Bookkeeping functions to keep track of number of active readers/iterators for the list.
   jint increase_readers();
   jint decrease_readers();
-  void wait_until_no_readers() const;
 
  public:
   LogOutputList() : _active_readers(0) {
@@ -91,6 +90,7 @@ class LogOutputList {
   // Removes all outputs. Equivalent of set_output_level(out, Off)
   // for all outputs.
   void clear();
+  void wait_until_no_readers() const;
 
   class Iterator {
     friend class LogOutputList;
diff --git a/src/hotspot/share/logging/logTagSet.hpp b/src/hotspot/share/logging/logTagSet.hpp
index 52fdf19ef111d..f932a6d175a3b 100644
--- a/src/hotspot/share/logging/logTagSet.hpp
+++ b/src/hotspot/share/logging/logTagSet.hpp
@@ -67,6 +67,10 @@ class LogTagSet {
   static void describe_tagsets(outputStream* out);
   static void list_all_tagsets(outputStream* out);
 
+  void wait_until_no_readers() const {
+    _output_list.wait_until_no_readers();
+  }
+
   static LogTagSet* first() {
     return _list;
   }
@@ -163,4 +167,5 @@ class LogTagSetMapping : public AllStatic {
 template <LogTagType T0, LogTagType T1, LogTagType T2, LogTagType T3, LogTagType T4, LogTagType GuardTag>
 LogTagSet LogTagSetMapping<T0, T1, T2, T3, T4, GuardTag>::_tagset(&LogPrefix<T0, T1, T2, T3, T4>::prefix, T0, T1, T2, T3, T4);
 
+extern const size_t vwrite_buffer_size;
 #endif // SHARE_LOGGING_LOGTAGSET_HPP
diff --git a/src/hotspot/share/runtime/arguments.cpp b/src/hotspot/share/runtime/arguments.cpp
index 5d45fce46065d..24cb6d66c05b8 100644
--- a/src/hotspot/share/runtime/arguments.cpp
+++ b/src/hotspot/share/runtime/arguments.cpp
@@ -2745,6 +2745,9 @@ jint Arguments::parse_each_vm_init_arg(const JavaVMInitArgs* args, bool* patch_m
       } else if (strcmp(tail, ":disable") == 0) {
         LogConfiguration::disable_logging();
         ret = true;
+      } else if (strcmp(tail, ":async") == 0) {
+        LogConfiguration::set_async_mode(true);
+        ret = true;
       } else if (*tail == '\0') {
         ret = LogConfiguration::parse_command_line_arguments();
         assert(ret, "-Xlog without arguments should never fail to parse");
diff --git a/src/hotspot/share/runtime/globals.hpp b/src/hotspot/share/runtime/globals.hpp
index 3a379106e3e1c..5ffb1e755cc1b 100644
--- a/src/hotspot/share/runtime/globals.hpp
+++ b/src/hotspot/share/runtime/globals.hpp
@@ -2016,6 +2016,11 @@ const intx ObjectAlignmentInBytes = 8;
              "Use the FP register for holding the frame pointer "           \
              "and not as a general purpose register.")                      \
                                                                             \
+  product(size_t, AsyncLogBufferSize, 2*M,                                  \
+          "Memory budget (in bytes) for the buffer of Asynchronous "        \
+          "Logging (-Xlog:async).")                                         \
+          range(100*K, 50*M)                                                \
+                                                                            \
   product(bool, CheckIntrinsics, true, DIAGNOSTIC,                          \
              "When a class C is loaded, check that "                        \
              "(1) all intrinsics defined by the VM for class C are present "\
diff --git a/src/hotspot/share/runtime/init.cpp b/src/hotspot/share/runtime/init.cpp
index b7f5015ee2a8f..06974b89e4225 100644
--- a/src/hotspot/share/runtime/init.cpp
+++ b/src/hotspot/share/runtime/init.cpp
@@ -33,6 +33,7 @@
 #endif
 #include "interpreter/bytecodes.hpp"
 #include "logging/log.hpp"
+#include "logging/logAsyncWriter.hpp"
 #include "logging/logTag.hpp"
 #include "memory/universe.hpp"
 #include "prims/jvmtiExport.hpp"
@@ -123,6 +124,7 @@ jint init_globals() {
   if (status != JNI_OK)
     return status;
 
+  AsyncLogWriter::initialize();
   gc_barrier_stubs_init();  // depends on universe_init, must be before interpreter_init
   interpreter_init_stub();  // before methods get loaded
   accessFlags_init();
diff --git a/src/hotspot/share/runtime/os.hpp b/src/hotspot/share/runtime/os.hpp
index 290c4c8abb69b..7eaaa9db984f4 100644
--- a/src/hotspot/share/runtime/os.hpp
+++ b/src/hotspot/share/runtime/os.hpp
@@ -450,6 +450,7 @@ class os: AllStatic {
     java_thread,       // Java, CodeCacheSweeper, JVMTIAgent and Service threads.
     compiler_thread,
     watcher_thread,
+    asynclog_thread,   // dedicated to flushing logs
     os_thread
   };
 
diff --git a/src/hotspot/share/runtime/thread.cpp b/src/hotspot/share/runtime/thread.cpp
index 1b94779a49879..3ff07b2676e7f 100644
--- a/src/hotspot/share/runtime/thread.cpp
+++ b/src/hotspot/share/runtime/thread.cpp
@@ -53,6 +53,7 @@
 #include "jfr/jfrEvents.hpp"
 #include "jvmtifiles/jvmtiEnv.hpp"
 #include "logging/log.hpp"
+#include "logging/logAsyncWriter.hpp"
 #include "logging/logConfiguration.hpp"
 #include "logging/logStream.hpp"
 #include "memory/allocation.inline.hpp"
@@ -3768,6 +3769,7 @@ void Threads::print_on(outputStream* st, bool print_stacks,
     StringDedup::threads_do(&cl);
   }
   cl.do_thread(WatcherThread::watcher_thread());
+  cl.do_thread(AsyncLogWriter::instance());
 
   st->flush();
 }
@@ -3821,6 +3823,7 @@ void Threads::print_on_error(outputStream* st, Thread* current, char* buf,
   st->print_cr("Other Threads:");
   print_on_error(VMThread::vm_thread(), st, current, buf, buflen, &found_current);
   print_on_error(WatcherThread::watcher_thread(), st, current, buf, buflen, &found_current);
+  print_on_error(AsyncLogWriter::instance(), st, current, buf, buflen, &found_current);
 
   if (Universe::heap() != NULL) {
     PrintOnErrorClosure print_closure(st, current, buf, buflen, &found_current);
diff --git a/src/hotspot/share/runtime/thread.hpp b/src/hotspot/share/runtime/thread.hpp
index c77a94974388d..76076709df67f 100644
--- a/src/hotspot/share/runtime/thread.hpp
+++ b/src/hotspot/share/runtime/thread.hpp
@@ -103,6 +103,7 @@ class JavaThread;
 //         - GangWorker
 //     - WatcherThread
 //     - JfrThreadSampler
+//     - LogAsyncWriter
 //
 // All Thread subclasses must be either JavaThread or NonJavaThread.
 // This means !t->is_Java_thread() iff t is a NonJavaThread, or t is
diff --git a/src/hotspot/share/runtime/vmStructs.cpp b/src/hotspot/share/runtime/vmStructs.cpp
index 34bd42bd5f826..af69cf1aa09e8 100644
--- a/src/hotspot/share/runtime/vmStructs.cpp
+++ b/src/hotspot/share/runtime/vmStructs.cpp
@@ -51,6 +51,7 @@
 #include "gc/shared/vmStructs_gc.hpp"
 #include "interpreter/bytecodes.hpp"
 #include "interpreter/interpreter.hpp"
+#include "logging/logAsyncWriter.hpp"
 #include "memory/allocation.hpp"
 #include "memory/allocation.inline.hpp"
 #include "memory/heap.hpp"
@@ -1336,6 +1337,7 @@ typedef HashtableEntry<InstanceKlass*, mtClass>  KlassHashtableEntry;
       declare_type(NonJavaThread, Thread)                                 \
         declare_type(NamedThread, NonJavaThread)                          \
         declare_type(WatcherThread, NonJavaThread)                        \
+        declare_type(AsyncLogWriter, NonJavaThread)                       \
       declare_type(JavaThread, Thread)                                    \
         declare_type(JvmtiAgentThread, JavaThread)                        \
         declare_type(MonitorDeflationThread, JavaThread)                  \
diff --git a/src/hotspot/share/utilities/hashtable.cpp b/src/hotspot/share/utilities/hashtable.cpp
index c3c62e4c20b12..5b99bba8bec18 100644
--- a/src/hotspot/share/utilities/hashtable.cpp
+++ b/src/hotspot/share/utilities/hashtable.cpp
@@ -279,6 +279,7 @@ template class BasicHashtable<mtModule>;
 template class BasicHashtable<mtCompiler>;
 template class BasicHashtable<mtTracing>;
 template class BasicHashtable<mtServiceability>;
+template class BasicHashtable<mtLogging>;
 
 template void BasicHashtable<mtClass>::verify_table<DictionaryEntry>(char const*);
 template void BasicHashtable<mtModule>::verify_table<ModuleEntry>(char const*);
diff --git a/test/hotspot/gtest/logging/logTestFixture.cpp b/test/hotspot/gtest/logging/logTestFixture.cpp
index bc153b76fb900..c9d974230976f 100644
--- a/test/hotspot/gtest/logging/logTestFixture.cpp
+++ b/test/hotspot/gtest/logging/logTestFixture.cpp
@@ -46,6 +46,7 @@ LogTestFixture::LogTestFixture() : _n_snapshots(0), _configuration_snapshot(NULL
 }
 
 LogTestFixture::~LogTestFixture() {
+  AsyncLogWriter::flush();
   restore_config();
   clear_snapshot();
   delete_file(TestLogFileName);
diff --git a/test/hotspot/gtest/logging/logTestUtils.inline.hpp b/test/hotspot/gtest/logging/logTestUtils.inline.hpp
index c1bc288a38d1f..e7f9f1daf8957 100644
--- a/test/hotspot/gtest/logging/logTestUtils.inline.hpp
+++ b/test/hotspot/gtest/logging/logTestUtils.inline.hpp
@@ -22,6 +22,7 @@
  */
 
 #include "logging/log.hpp"
+#include "logging/logAsyncWriter.hpp"
 #include "logging/logConfiguration.hpp"
 #include "logging/logStream.hpp"
 #include "memory/resourceArea.hpp"
@@ -46,6 +47,7 @@ static inline bool file_exists(const char* filename) {
 }
 
 static inline void delete_file(const char* filename) {
+  AsyncLogWriter::flush();
   if (!file_exists(filename)) {
     return;
   }
@@ -135,6 +137,7 @@ static inline char* read_line(FILE* fp) {
 }
 
 static bool file_contains_substrings_in_order(const char* filename, const char* substrs[]) {
+  AsyncLogWriter::flush();
   FILE* fp = fopen(filename, "r");
   assert(fp != NULL, "error opening file %s: %s", filename, strerror(errno));
 
diff --git a/test/hotspot/gtest/logging/test_asynclog.cpp b/test/hotspot/gtest/logging/test_asynclog.cpp
new file mode 100644
index 0000000000000..ffabbb524c2b2
--- /dev/null
+++ b/test/hotspot/gtest/logging/test_asynclog.cpp
@@ -0,0 +1,216 @@
+/*
+ * Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ *
+ */
+#include "precompiled.hpp"
+#include "jvm.h"
+#include "logging/log.hpp"
+#include "logging/logAsyncWriter.hpp"
+#include "logging/logMessage.hpp"
+#include "logTestFixture.hpp"
+#include "logTestUtils.inline.hpp"
+#include "unittest.hpp"
+
+class AsyncLogTest : public LogTestFixture {
+ public:
+  AsyncLogTest() {
+    if(!LogConfiguration::is_async_mode()) {
+      fprintf(stderr, "Warning: asynclog is OFF.\n");
+    }
+  }
+
+  void test_asynclog_ls() {
+    LogStream ls(Log(logging)::info());
+    outputStream* os = &ls;
+    os->print_cr("LogStreamWithAsyncLogImpl");
+    os->print_cr("LogStreamWithAsyncLogImpl secondline");
+
+    //multi-lines
+    os->print("logStream msg1-");
+    os->print("msg2-");
+    os->print("msg3\n");
+    os->print_cr("logStream newline");
+  }
+
+  void test_asynclog_raw() {
+    Log(logging) logger;
+#define LOG_LEVEL(level, name) logger.name("1" #level);
+LOG_LEVEL_LIST
+#undef LOG_LEVEL
+
+    LogTarget(Trace, logging) t;
+    LogTarget(Debug, logging) d;
+    EXPECT_FALSE(t.is_enabled());
+    EXPECT_TRUE(d.is_enabled());
+
+    d.print("AsyncLogTarget.print = %d", 1);
+    log_trace(logging)("log_trace-test");
+    log_debug(logging)("log_debug-test");
+  }
+};
+
+TEST_VM(AsyncLogBufferTest, fifo) {
+  LinkedListDeque<int, mtLogging> fifo;
+  LinkedListImpl<int, ResourceObj::C_HEAP, mtLogging> result;
+
+  fifo.push_back(1);
+  EXPECT_EQ((size_t)1, fifo.size());
+  EXPECT_EQ(1, *(fifo.back()));
+
+  fifo.pop_all(&result);
+  EXPECT_EQ((size_t)0, fifo.size());
+  EXPECT_EQ(NULL, fifo.back());
+  EXPECT_EQ((size_t)1, result.size());
+  EXPECT_EQ(1, *(result.head()->data()));
+  result.clear();
+
+  fifo.push_back(2);
+  fifo.push_back(1);
+  fifo.pop_all(&result);
+  EXPECT_EQ((size_t)2, result.size());
+  EXPECT_EQ(2, *(result.head()->data()));
+  EXPECT_EQ(1, *(result.head()->next()->data()));
+  result.clear();
+  const int N = 1000;
+  for (int i=0; i<N; ++i) {
+    fifo.push_back(i);
+  }
+  fifo.pop_all(&result);
+
+  EXPECT_EQ((size_t)N, result.size());
+  LinkedListIterator<int> it(result.head());
+  for (int i=0; i<N; ++i) {
+    int* e = it.next();
+    EXPECT_EQ(i, *e);
+  }
+}
+
+TEST_VM(AsyncLogBufferTest, deque) {
+  LinkedListDeque<int, mtLogging> deque;
+  const int N = 10;
+
+  EXPECT_EQ(NULL, deque.front());
+  EXPECT_EQ(NULL, deque.back());
+  for (int i = 0; i < N; ++i) {
+    deque.push_back(i);
+  }
+
+  EXPECT_EQ(0, *(deque.front()));
+  EXPECT_EQ(N-1, *(deque.back()));
+  EXPECT_EQ((size_t)N, deque.size());
+
+  deque.pop_front();
+  EXPECT_EQ((size_t)(N - 1), deque.size());
+  EXPECT_EQ(1, *(deque.front()));
+  EXPECT_EQ(N - 1, *(deque.back()));
+
+  deque.pop_front();
+  EXPECT_EQ((size_t)(N - 2), deque.size());
+  EXPECT_EQ(2, *(deque.front()));
+  EXPECT_EQ(N - 1, *(deque.back()));
+
+
+  for (int i=2; i < N-1; ++i) {
+    deque.pop_front();
+  }
+  EXPECT_EQ((size_t)1, deque.size());
+  EXPECT_EQ(N - 1, *(deque.back()));
+  EXPECT_EQ(deque.back(), deque.front());
+
+  deque.pop_front();
+  EXPECT_EQ((size_t)0, deque.size());
+}
+
+TEST_VM_F(AsyncLogTest, asynclog) {
+  set_log_config(TestLogFileName, "logging=debug");
+
+  test_asynclog_ls();
+  test_asynclog_raw();
+  AsyncLogWriter::flush();
+
+  EXPECT_TRUE(file_contains_substring(TestLogFileName, "LogStreamWithAsyncLogImpl"));
+  EXPECT_TRUE(file_contains_substring(TestLogFileName, "logStream msg1-msg2-msg3"));
+  EXPECT_TRUE(file_contains_substring(TestLogFileName, "logStream newline"));
+
+  EXPECT_TRUE(file_contains_substring(TestLogFileName, "1Debug"));
+  EXPECT_TRUE(file_contains_substring(TestLogFileName, "1Info"));
+  EXPECT_TRUE(file_contains_substring(TestLogFileName, "1Warning"));
+  EXPECT_TRUE(file_contains_substring(TestLogFileName, "1Error"));
+  EXPECT_FALSE(file_contains_substring(TestLogFileName, "1Trace")); // trace message is masked out
+
+  EXPECT_TRUE(file_contains_substring(TestLogFileName, "AsyncLogTarget.print = 1"));
+  EXPECT_FALSE(file_contains_substring(TestLogFileName, "log_trace-test")); // trace message is masked out
+  EXPECT_TRUE(file_contains_substring(TestLogFileName, "log_debug-test"));
+}
+
+TEST_VM_F(AsyncLogTest, logMessage) {
+  set_log_config(TestLogFileName, "logging=debug");
+
+  const int MULTI_LINES = 20;
+  {
+
+    LogMessage(logging) msg;
+    Log(logging) logger;
+
+    for (int i = 0; i < MULTI_LINES; ++i) {
+      msg.debug("nonbreakable log message line-%02d", i);
+
+      if (0 == (i % 4)) {
+        logger.debug("a noisy message from other logger");
+      }
+    }
+    logger.debug("a noisy message from other logger");
+  }
+  AsyncLogWriter::flush();
+
+  ResourceMark rm;
+  LogMessageBuffer buffer;
+  const char* strs[MULTI_LINES + 1];
+  strs[MULTI_LINES] = NULL;
+  for (int i = 0; i < MULTI_LINES; ++i) {
+    stringStream ss;
+    ss.print_cr("nonbreakable log message line-%02d", i);
+    strs[i] = ss.as_string();
+  }
+  // check nonbreakable log messages are consecutive
+  EXPECT_TRUE(file_contains_substrings_in_order(TestLogFileName, strs));
+  EXPECT_TRUE(file_contains_substring(TestLogFileName, "a noisy message from other logger"));
+}
+
+TEST_VM_F(AsyncLogTest, droppingMessage) {
+  set_log_config(TestLogFileName, "logging=debug");
+  const size_t sz = 100;
+
+  if (AsyncLogWriter::instance() != nullptr) {
+    // shrink async buffer.
+    AutoModifyRestore<size_t> saver(AsyncLogBufferSize, sz * 1024 /*in byte*/);
+    LogMessage(logging) lm;
+
+    // write 100x more messages than its capacity in burst
+    for (size_t i = 0; i < sz * 100; ++i) {
+      lm.debug("a lot of log...");
+    }
+    lm.flush();
+    AsyncLogWriter::flush();
+    EXPECT_TRUE(file_contains_substring(TestLogFileName, "messages dropped due to async logging"));
+  }
+}
diff --git a/test/hotspot/gtest/logging/test_log.cpp b/test/hotspot/gtest/logging/test_log.cpp
index 0700e202b702d..ccd928474d438 100644
--- a/test/hotspot/gtest/logging/test_log.cpp
+++ b/test/hotspot/gtest/logging/test_log.cpp
@@ -58,6 +58,7 @@ TEST_VM_F(LogTest, large_message) {
   memset(big_msg, Xchar, sizeof(big_msg) - 1);
   log_trace(logging)("%s", big_msg);
 
+  AsyncLogWriter::flush();
   ResourceMark rm;
   FILE* fp = fopen(TestLogFileName, "r");
   ASSERT_NE((void*)NULL, fp);
diff --git a/test/hotspot/jtreg/gtest/AsyncLogGtest.java b/test/hotspot/jtreg/gtest/AsyncLogGtest.java
new file mode 100644
index 0000000000000..f85f6d5c7e211
--- /dev/null
+++ b/test/hotspot/jtreg/gtest/AsyncLogGtest.java
@@ -0,0 +1,38 @@
+/*
+ * Copyright Amazon.com Inc. or its affiliates. All Rights Reserved.
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
+ *
+ * This code is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU General Public License version 2 only, as
+ * published by the Free Software Foundation.
+ *
+ * This code is distributed in the hope that it will be useful, but WITHOUT
+ * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+ * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
+ * version 2 for more details (a copy is included in the LICENSE file that
+ * accompanied this code).
+ *
+ * You should have received a copy of the GNU General Public License version
+ * 2 along with this work; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
+ *
+ * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
+ * or visit www.oracle.com if you need additional information or have any
+ * questions.
+ *
+ */
+
+/*
+ * Note: This runs the unified logging part of gtest in async mode.
+ * The reason is that hotspot can't safely turn off asynclogging dyanmically.
+ * There's no TEST_OTHER_VM_F.
+ */
+
+/* @test
+ * @summary Run logging gtest in async mode.
+ * @library /test/lib
+ * @modules java.base/jdk.internal.misc
+ *          java.xml
+ * @run main/native GTestWrapper --gtest_filter=AsyncLogTest* -Xlog:async
+ * @run main/native GTestWrapper --gtest_filter=Log*Test* -Xlog:async
+ */