8252090: JFR: StreamWriterHost::write_unbuffered() stucks in an infinite loop OpenJDK (build 13.0.1+9)

Reviewed-by: hseigel

src/hotspot/share/jfr/instrumentation/jfrEventClassTransformer.cpp

@@ -625,7 +625,7 @@ static jlong add_method_info(JfrBigEndianWriter& writer,
   DEBUG_ONLY(assert(start_offset + 8 == writer.current_offset(), "invariant");)
   // Code attribute
   writer.write(code_index); // "Code"
-  writer.bytes(code, code_len);
+  writer.write_bytes(code, code_len);
   DEBUG_ONLY(assert((start_offset + 8 + 2 + (jlong)code_len) == writer.current_offset(), "invariant");)
   return writer.current_offset();
 }
@@ -768,8 +768,8 @@ static u2 position_stream_after_methods(JfrBigEndianWriter& writer,
       // We will need to re-create a new <clinit> in a later step.
       // For now, ensure that this method is excluded from the methods
       // being copied.
-      writer.bytes(stream->buffer() + orig_method_len_offset,
-                   method_offset - orig_method_len_offset);
+      writer.write_bytes(stream->buffer() + orig_method_len_offset,
+                         method_offset - orig_method_len_offset);
       assert(writer.is_valid(), "invariant");
 
       // Update copy position to skip copy of <clinit> method
@@ -1091,7 +1091,7 @@ static jlong insert_clinit_method(const InstanceKlass* ik,
     writer.write<u1>((u1)Bytecodes::_nop);
     writer.write<u1>((u1)Bytecodes::_nop);
     // insert original clinit code
-    writer.bytes(orig_bytecodes, orig_bytecodes_length);
+    writer.write_bytes(orig_bytecodes, orig_bytecodes_length);
   }
 
   /* END CLINIT CODE */
@@ -1290,7 +1290,7 @@ static u1* new_bytes_for_lazy_instrumentation(const InstanceKlass* ik,
   const u4 orig_access_flag_offset = orig_stream->current_offset();
   // Copy original stream from the beginning up to AccessFlags
   // This means the original constant pool contents are copied unmodified
-  writer.bytes(orig_stream->buffer(), orig_access_flag_offset);
+  writer.write_bytes(orig_stream->buffer(), orig_access_flag_offset);
   assert(writer.is_valid(), "invariant");
   assert(writer.current_offset() == (intptr_t)orig_access_flag_offset, "invariant"); // same positions
   // Our writer now sits just after the last original constant pool entry.
@@ -1324,14 +1324,14 @@ static u1* new_bytes_for_lazy_instrumentation(const InstanceKlass* ik,
   orig_stream->skip_u2_fast(iface_len);
   const u4 orig_fields_len_offset = orig_stream->current_offset();
   // Copy from AccessFlags up to and including interfaces
-  writer.bytes(orig_stream->buffer() + orig_access_flag_offset,
-               orig_fields_len_offset - orig_access_flag_offset);
+  writer.write_bytes(orig_stream->buffer() + orig_access_flag_offset,
+                     orig_fields_len_offset - orig_access_flag_offset);
   assert(writer.is_valid(), "invariant");
   const jlong new_fields_len_offset = writer.current_offset();
   const u2 orig_fields_len = position_stream_after_fields(orig_stream);
   u4 orig_method_len_offset = orig_stream->current_offset();
   // Copy up to and including fields
-  writer.bytes(orig_stream->buffer() + orig_fields_len_offset, orig_method_len_offset - orig_fields_len_offset);
+  writer.write_bytes(orig_stream->buffer() + orig_fields_len_offset, orig_method_len_offset - orig_fields_len_offset);
   assert(writer.is_valid(), "invariant");
   // We are sitting just after the original number of field_infos
   // so this is a position where we can add (append) new field_infos
@@ -1350,7 +1350,7 @@ static u1* new_bytes_for_lazy_instrumentation(const InstanceKlass* ik,
                                                             orig_method_len_offset);
   const u4 orig_attributes_count_offset = orig_stream->current_offset();
   // Copy existing methods
-  writer.bytes(orig_stream->buffer() + orig_method_len_offset, orig_attributes_count_offset - orig_method_len_offset);
+  writer.write_bytes(orig_stream->buffer() + orig_method_len_offset, orig_attributes_count_offset - orig_method_len_offset);
   assert(writer.is_valid(), "invariant");
   // We are sitting just after the original number of method_infos
   // so this is a position where we can add (append) new method_infos
@@ -1372,7 +1372,7 @@ static u1* new_bytes_for_lazy_instrumentation(const InstanceKlass* ik,
   writer.write_at_offset<u2>(orig_methods_len + number_of_new_methods, new_method_len_offset);
   assert(writer.is_valid(), "invariant");
   // Copy last remaining bytes
-  writer.bytes(orig_stream->buffer() + orig_attributes_count_offset, orig_stream_size - orig_attributes_count_offset);
+  writer.write_bytes(orig_stream->buffer() + orig_attributes_count_offset, orig_stream_size - orig_attributes_count_offset);
   assert(writer.is_valid(), "invariant");
   assert(writer.current_offset() > orig_stream->length(), "invariant");
   size_of_new_bytes = (jint)writer.current_offset();

src/hotspot/share/jfr/recorder/repository/jfrChunkWriter.cpp

@@ -62,7 +62,7 @@ class JfrChunkHeadWriter : public StackObj {
   JfrChunk* _chunk;
  public:
   void write_magic() {
-    _writer->bytes(_chunk->magic(), MAGIC_LEN);
+    _writer->write_bytes(_chunk->magic(), MAGIC_LEN);
   }
 
   void write_version() {

src/hotspot/share/jfr/recorder/storage/jfrStorageUtils.inline.hpp

@@ -31,7 +31,8 @@
 
 template <typename T>
 inline bool UnBufferedWriteToChunk<T>::write(T* t, const u1* data, size_t size) {
-  _writer.write_unbuffered(data, size);
+  assert((intptr_t)size >= 0, "invariant");
+  _writer.write_unbuffered(data, (intptr_t)size);
   ++_elements;
   _size += size;
   return true;
@@ -56,6 +57,7 @@ inline bool ConcurrentWriteOp<Operation>::process(typename Operation::Type* t) {
   // acquire_critical_section_top() must be read before pos() for stable access
   const u1* const top = is_retired ? t->top() : t->acquire_critical_section_top();
   const size_t unflushed_size = get_unflushed_size(top, t);
+  assert((intptr_t)unflushed_size >= 0, "invariant");
   if (unflushed_size == 0) {
     if (is_retired) {
       t->set_top(top);
@@ -78,6 +80,7 @@ inline bool MutexedWriteOp<Operation>::process(typename Operation::Type* t) {
   assert(t != NULL, "invariant");
   const u1* const top = t->top();
   const size_t unflushed_size = get_unflushed_size(top, t);
+  assert((intptr_t)unflushed_size >= 0, "invariant");
   if (unflushed_size == 0) {
     return true;
   }
@@ -113,6 +116,7 @@ inline bool DiscardOp<Operation>::process(typename Operation::Type* t) {
   assert(t != NULL, "invariant");
   const u1* const top = _mode == concurrent ? t->acquire_critical_section_top() : t->top();
   const size_t unflushed_size = get_unflushed_size(top, t);
+  assert((intptr_t)unflushed_size >= 0, "invariant");
   if (unflushed_size == 0) {
     if (_mode == concurrent) {
       t->release_critical_section_top(top);
@@ -141,6 +145,7 @@ inline bool EpochDispatchOp<Operation>::process(typename Operation::Type* t) {
   assert(t != NULL, "invariant");
   const u1* const current_top = _previous_epoch ? t->start() : t->top();
   const size_t unflushed_size = Atomic::load_acquire(t->pos_address()) - current_top;
+  assert((intptr_t)unflushed_size >= 0, "invariant");
   if (unflushed_size == 0) {
     return true;
   }

src/hotspot/share/jfr/utilities/jfrBlob.hpp

@@ -50,7 +50,7 @@ class JfrBlob : public JfrCHeapObj {
   static JfrBlobHandle make(const u1* data, size_t size);
   template <typename Writer>
   void write(Writer& writer) const {
-    writer.bytes(_data, _size);
+    writer.write_bytes(_data, _size);
     if (_next.valid()) {
       _next->write(writer);
     }
@@ -60,7 +60,7 @@ class JfrBlob : public JfrCHeapObj {
     if (_written) {
       return;
     }
-    writer.bytes(_data, _size);
+    writer.write_bytes(_data, _size);
     _written = true;
     if (_next.valid()) {
       _next->exclusive_write(writer);

src/hotspot/share/jfr/writers/jfrMemoryWriterHost.hpp

@@ -49,7 +49,7 @@ class MemoryWriterHost : public StorageHost<Adapter, AP> {
  public:
   typedef typename Adapter::StorageType StorageType;
  protected:
-  void bytes(void* dest, const void* buf, size_t len);
+  void write_bytes(void* dest, const void* buf, intptr_t len);
   MemoryWriterHost(StorageType* storage, Thread* thread);
   MemoryWriterHost(StorageType* storage, size_t size);
   MemoryWriterHost(Thread* thread);

src/hotspot/share/jfr/writers/jfrMemoryWriterHost.inline.hpp

@@ -28,9 +28,10 @@
 #include "jfr/writers/jfrMemoryWriterHost.hpp"
 
 template <typename Adapter, typename AP, typename AccessAssert>
-inline void MemoryWriterHost<Adapter, AP, AccessAssert>::bytes(void* dest, const void* buf, size_t len) {
+inline void MemoryWriterHost<Adapter, AP, AccessAssert>::write_bytes(void* dest, const void* buf, intptr_t len) {
   assert(dest != NULL, "invariant");
-  memcpy(dest, buf, len); // no encoding
+  assert(len >= 0, "invariant");
+  memcpy(dest, buf, (size_t)len); // no encoding
   this->set_current_pos(len);
 }
 

src/hotspot/share/jfr/writers/jfrStreamWriterHost.hpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2020, 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
@@ -37,12 +37,14 @@ class StreamWriterHost : public MemoryWriterHost<Adapter, AP> {
   fio_fd _fd;
   int64_t current_stream_position() const;
 
+  void write_bytes(const u1* buf, intptr_t len);
+
  protected:
   StreamWriterHost(StorageType* storage, Thread* thread);
   StreamWriterHost(StorageType* storage, size_t size);
   StreamWriterHost(Thread* thread);
   bool accommodate(size_t used, size_t requested);
-  void bytes(void* dest, const void* src, size_t len);
+  void write_bytes(void* dest, const void* src, intptr_t len);
   void flush(size_t size);
   bool has_valid_fd() const;
 
@@ -50,7 +52,7 @@ class StreamWriterHost : public MemoryWriterHost<Adapter, AP> {
   int64_t current_offset() const;
   void seek(int64_t offset);
   void flush();
-  void write_unbuffered(const void* src, size_t len);
+  void write_unbuffered(const void* src, intptr_t len);
   bool is_valid() const;
   void close_fd();
   void reset(fio_fd fd);

src/hotspot/share/jfr/writers/jfrStreamWriterHost.inline.hpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2020, 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
@@ -61,19 +61,33 @@ inline bool StreamWriterHost<Adapter, AP>::accommodate(size_t used, size_t reque
 }
 
 template <typename Adapter, typename AP>
-inline void StreamWriterHost<Adapter, AP>::bytes(void* dest, const void* buf, size_t len) {
-  if (len > this->available_size()) {
+inline void StreamWriterHost<Adapter, AP>::write_bytes(void* dest, const void* buf, intptr_t len) {
+  assert(len >= 0, "invariant");
+  if (len > (intptr_t)this->available_size()) {
     this->write_unbuffered(buf, len);
     return;
   }
-  MemoryWriterHost<Adapter, AP>::bytes(dest, buf, len);
+  MemoryWriterHost<Adapter, AP>::write_bytes(dest, buf, len);
+}
+
+template <typename Adapter, typename AP>
+inline void StreamWriterHost<Adapter, AP>::write_bytes(const u1* buf, intptr_t len) {
+  assert(len >= 0, "invariant");
+  while (len > 0) {
+    const unsigned int nBytes = len > INT_MAX ? INT_MAX : (unsigned int)len;
+    const ssize_t num_written = (ssize_t)os::write(_fd, buf, nBytes);
+    guarantee(num_written > 0, "Nothing got written, or os::write() failed");
+    _stream_pos += num_written;
+    len -= num_written;
+    buf += num_written;
+  }
 }
 
 template <typename Adapter, typename AP>
 inline void StreamWriterHost<Adapter, AP>::flush(size_t size) {
   assert(size > 0, "invariant");
   assert(this->is_valid(), "invariant");
-  _stream_pos += os::write(_fd, this->start_pos(), (unsigned int)size);
+  this->write_bytes(this->start_pos(), (intptr_t)size);
   StorageHost<Adapter, AP>::reset();
   assert(0 == this->used_offset(), "invariant");
 }
@@ -106,14 +120,10 @@ void StreamWriterHost<Adapter, AP>::flush() {
 }
 
 template <typename Adapter, typename AP>
-void StreamWriterHost<Adapter, AP>::write_unbuffered(const void* buf, size_t len) {
+void StreamWriterHost<Adapter, AP>::write_unbuffered(const void* buf, intptr_t len) {
   this->flush();
   assert(0 == this->used_offset(), "can only seek from beginning");
-  while (len > 0) {
-    const unsigned int n = MIN2((unsigned int)len, (unsigned int)INT_MAX);
-    _stream_pos += os::write(_fd, buf, n);
-    len -= n;
-  }
+  this->write_bytes((const u1*)buf, len);
 }
 
 template <typename Adapter, typename AP>

src/hotspot/share/jfr/writers/jfrWriterHost.hpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2019, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2020, 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
@@ -88,7 +88,7 @@ class WriterHost : public WriterPolicyImpl {
   void write(const Tickspan& time);
   void write(const JfrTicks& time);
   void write(const JfrTickspan& time);
-  void bytes(const void* buf, size_t len);
+  void write_bytes(const void* buf, intptr_t len);
   void write_utf8_u2_len(const char* value);
   template <typename T>
   void write_padded_at_offset(T value, int64_t offset);

src/hotspot/share/jfr/writers/jfrWriterHost.inline.hpp

@@ -293,10 +293,11 @@ void WriterHost<BE, IE, WriterPolicyImpl>::write(const JfrTickspan& time) {
 }
 
 template <typename BE, typename IE, typename WriterPolicyImpl>
-void WriterHost<BE, IE, WriterPolicyImpl>::bytes(const void* buf, size_t len) {
-  u1* const pos = this->ensure_size(len);
+void WriterHost<BE, IE, WriterPolicyImpl>::write_bytes(const void* buf, intptr_t len) {
+  assert(len >= 0, "invariant");
+  u1* const pos = this->ensure_size((size_t)len);
   if (pos != NULL) {
-    WriterPolicyImpl::bytes(pos, buf, len); // WriterPolicyImpl responsible for position update
+    WriterPolicyImpl::write_bytes(pos, buf, len); // WriterPolicyImpl responsible for position update
   }
 }