BUG#27463149 - SCALABILITY ISSUE WITH XPLUGINS MESSAG_BUILDER

Description
===========
Reception of a resultset in X Protocol consists of multiple protocol
messages. X Plugin allocates CodedOutputStream on heap for each
protocol message. The allocation degrades performance:

*  PFS OFF - degradation around 3% for point_select test
*  PFS ON - degradation around 100%-300% for point_select test

Fix
===
The memory region for CodedOutputStream is allocated once, and for
each message the object is created by placement new operator.

RB:18975
Reviewed-by: Tomasz Stepniak <[email protected]>
Reviewed-by: Andrzej Religa <[email protected]>

Author: lkotula

plugin/x/ngs/include/ngs/protocol/message_builder.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016, 2017, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2016, 2018, Oracle and/or its affiliates. All rights reserved.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0,
@@ -25,8 +25,10 @@
 #ifndef _NGS_MESSAGE_BUILDER_H_
 #define _NGS_MESSAGE_BUILDER_H_
 
+#include <memory>
 #include "m_ctype.h"
 #include "my_inttypes.h"
+
 #include "plugin/x/ngs/include/ngs/memory.h"
 #include "plugin/x/ngs/include/ngs_common/protocol_protobuf.h"
 
@@ -43,15 +45,19 @@ class Message_builder {
 
  protected:
   using CodedOutputStream = ::google::protobuf::io::CodedOutputStream;
-  using CodedOutputStream_ptr =
-      ngs::Memory_instrumented<CodedOutputStream>::Unique_ptr;
+  using Stream_allocator = std::allocator<CodedOutputStream>;
 
  protected:
   void start_message(Output_buffer *out_buffer, const uint8 type);
   void end_message();
 
+  void construct_stream();
+  void reset_stream();
+
   Output_buffer *m_out_buffer;
-  CodedOutputStream_ptr m_out_stream;
+  CodedOutputStream *m_out_stream;
+  bool m_valid_out_stream{false};
+
   int m_size_addr2_size;
   int m_field_number;
 

plugin/x/ngs/src/protocol/message_builder.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2017, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0,
@@ -30,9 +30,12 @@
 
 namespace ngs {
 
-Message_builder::Message_builder() : m_out_buffer(NULL) {}
+Message_builder::Message_builder()
+    : m_out_buffer(nullptr), m_out_stream(Stream_allocator().allocate(1)) {}
 
-Message_builder::~Message_builder() {}
+Message_builder::~Message_builder() {
+  Stream_allocator().deallocate(m_out_stream, 1);
+}
 
 void Message_builder::encode_uint32(const uint32 value, const bool write) {
   ++m_field_number;
@@ -41,7 +44,7 @@ void Message_builder::encode_uint32(const uint32 value, const bool write) {
     google::protobuf::internal::WireFormatLite::WriteTag(
         m_field_number,
         google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT,
-        m_out_stream.get());
+        m_out_stream);
     m_out_stream->WriteVarint32(value);
   }
 }
@@ -53,7 +56,7 @@ void Message_builder::encode_uint64(const uint64 value, const bool write) {
     google::protobuf::internal::WireFormatLite::WriteTag(
         m_field_number,
         google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT,
-        m_out_stream.get());
+        m_out_stream);
     m_out_stream->WriteVarint64(value);
   }
 }
@@ -64,7 +67,7 @@ void Message_builder::encode_int32(const int32 value, const bool write) {
     google::protobuf::internal::WireFormatLite::WriteTag(
         m_field_number,
         google::protobuf::internal::WireFormatLite::WIRETYPE_VARINT,
-        m_out_stream.get());
+        m_out_stream);
     m_out_stream->WriteVarint32SignExtended(value);
   }
 }
@@ -77,7 +80,7 @@ void Message_builder::encode_string(const char *value, const size_t len,
     google::protobuf::internal::WireFormatLite::WriteTag(
         m_field_number,
         google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED,
-        m_out_stream.get());
+        m_out_stream);
     m_out_stream->WriteVarint32(static_cast<google::protobuf::uint32>(len));
     m_out_stream->WriteRaw(value, static_cast<int>(len));
   }
@@ -87,6 +90,18 @@ void Message_builder::encode_string(const char *value, const bool write) {
   encode_string(value, write ? strlen(value) : 0, write);
 }
 
+void Message_builder::construct_stream() {
+  if (m_valid_out_stream) reset_stream();
+
+  Stream_allocator().construct(m_out_stream, m_out_buffer);
+  m_valid_out_stream = true;
+}
+
+void Message_builder::reset_stream() {
+  Stream_allocator().destroy(m_out_stream);
+  m_valid_out_stream = false;
+}
+
 void Message_builder::start_message(Output_buffer *out_buffer,
                                     const uint8 type) {
   m_field_number = 0;
@@ -95,7 +110,8 @@ void Message_builder::start_message(Output_buffer *out_buffer,
   m_out_buffer->save_state();
   m_out_buffer->reserve(5);
   m_start_from = static_cast<uint32>(m_out_buffer->ByteCount());
-  m_out_stream.reset(ngs::allocate_object<CodedOutputStream>(m_out_buffer));
+
+  construct_stream();
 
   // at this point we don't know the size but we need to reserve the space for
   // it it is possible that the size which is stored on 4-bytes will be split
@@ -128,7 +144,7 @@ void Message_builder::end_message() {
   // here we already know the buffer size, so write it at the beginning of the
   // buffer the order is important here as the stream's destructor calls
   // buffer's BackUp() validating ByteCount
-  m_out_stream.reset();
+  reset_stream();
 
   uint32 msg_size = static_cast<uint32>(m_out_buffer->ByteCount()) -
                     m_start_from - sizeof(google::protobuf::uint32);

plugin/x/ngs/src/protocol/notice_builder.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2017, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0,
@@ -58,7 +58,7 @@ void Notice_builder::encode_rows_affected(Output_buffer *out_buffer,
   // 3) Payload
   google::protobuf::internal::WireFormatLite::WriteTag(
       3, google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED,
-      m_out_stream.get());
+      m_out_stream);
   uint32 size_scalar = CodedOutputStream::VarintSize32SignExtended(type) +
                        CodedOutputStream::VarintSize64(value) + 2 /*tags*/;
   uint32 size_payload =
@@ -74,7 +74,7 @@ void Notice_builder::encode_rows_affected(Output_buffer *out_buffer,
     google::protobuf::internal::WireFormatLite::WriteTag(
         2,
         google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED,
-        m_out_stream.get());
+        m_out_stream);
     m_out_stream->WriteVarint32(size_scalar);
     {
       m_field_number = 0;

plugin/x/ngs/src/protocol/row_builder.cc

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2015, 2017, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2015, 2018, Oracle and/or its affiliates. All rights reserved.
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License, version 2.0,
@@ -46,7 +46,7 @@ using namespace ngs;
   google::protobuf::internal::WireFormatLite::WriteTag(                      \
       1,                                                                     \
       google::protobuf::internal::WireFormatLite::WIRETYPE_LENGTH_DELIMITED, \
-      m_out_stream.get());                                                   \
+      m_out_stream);                                                         \
   ++m_num_fields;
 
 Row_builder::Row_builder() : m_row_processing(false) {}
@@ -55,7 +55,7 @@ Row_builder::~Row_builder() { abort_row(); }
 
 void Row_builder::abort_row() {
   if (m_row_processing) {
-    m_out_stream.reset();
+    reset_stream();
     m_out_buffer->rollback();
     m_row_processing = false;
   }
@@ -368,7 +368,7 @@ void Row_builder::add_time_field(const MYSQL_TIME *value, uint) {
   google::protobuf::uint8 neg = (value->neg) ? 0x01 : 0x00;
   m_out_stream->WriteRaw(&neg, 1);
 
-  append_time_values(value, m_out_stream.get());
+  append_time_values(value, m_out_stream);
 }
 
 void Row_builder::add_datetime_field(const MYSQL_TIME *value, uint) {
@@ -386,15 +386,15 @@ void Row_builder::add_datetime_field(const MYSQL_TIME *value, uint) {
   m_out_stream->WriteVarint64(value->month);
   m_out_stream->WriteVarint64(value->day);
 
-  append_time_values(value, m_out_stream.get());
+  append_time_values(value, m_out_stream);
 }
 
 void Row_builder::add_string_field(const char *const value, size_t length,
                                    const CHARSET_INFO *const) {
   ADD_FIELD_HEADER();
 
   m_out_stream->WriteVarint32(static_cast<google::protobuf::uint32>(
-      length + 1));  // 1 byte for thre trailing '\0'
+      length + 1));  // 1 byte for the trailing '\0'
 
   m_out_stream->WriteRaw(value, static_cast<int>(length));
   char zero = '\0';