Use boost::intrusive_ptr for internal Status representation

Summary:
Changed internal representation of Status from std::unique_ptr to boost::intrusive_ptr.
So status copy be copied cheaply.

Test Plan: Jenkins

Reviewers: timur, bogdan, mikhail

Reviewed By: mikhail

Subscribers: ybase

Differential Revision: https://phabricator.dev.yugabyte.com/D3752

src/yb/rpc/inbound_call.h

@@ -35,8 +35,6 @@
 #include <string>
 #include <vector>
 
-#include <boost/intrusive_ptr.hpp>
-
 #include <glog/logging.h>
 
 #include "yb/gutil/gscoped_ptr.h"

src/yb/rpc/rpc_fwd.h

@@ -18,8 +18,6 @@
 
 #include <chrono>
 
-#include <boost/intrusive_ptr.hpp>
-
 #include "yb/gutil/ref_counted.h"
 
 namespace boost {

src/yb/util/status.cc

@@ -41,16 +41,6 @@ const char* kTimeoutErrorMsgs[] = {
 
 }
 
-Status::StatePtr Status::CopyState() const {
-  if (!state_)
-    return nullptr;
-  auto state = state_.get();
-  size_t size = state->message_len + kHeaderSize;
-  StatePtr result(static_cast<State*>(malloc(size)));
-  memcpy(result.get(), state, size);
-  return result;
-}
-
 Status::Status(Code code,
                const char* file_name,
                int line_number,
@@ -65,6 +55,8 @@ Status::Status(Code code,
   state_->message_len = static_cast<uint32_t>(size);
   state_->code = static_cast<uint8_t>(code);
   state_->error_code = error_code;
+  // We aleady assigned intrusive_ptr, so counter should be one.
+  state_->counter.store(1, std::memory_order_relaxed);
   state_->file_name = file_name;
   state_->line_number = line_number;
   memcpy(state_->message, msg.data(), len1);

src/yb/util/status.h

@@ -27,11 +27,12 @@
 #ifndef YB_UTIL_STATUS_H_
 #define YB_UTIL_STATUS_H_
 
-#include <stdint.h>
-
+#include <atomic>
 #include <memory>
 #include <string>
 
+#include <boost/intrusive_ptr.hpp>
+
 #include <boost/preprocessor/cat.hpp>
 #include <boost/preprocessor/seq/for_each.hpp>
 #include <boost/preprocessor/tuple/elem.hpp>
@@ -176,14 +177,6 @@ class Status {
   // Create a success status.
   Status() {}
 
-  // Copy the specified status.
-  Status(const Status& s);
-  void operator=(const Status& s);
-
-  // Move the specified status.
-  Status(Status&& s);
-  void operator=(Status&& s);
-
   // Return a success status.
   static Status OK() { return Status(); }
 
@@ -251,13 +244,8 @@ class Status {
     return (state_ == nullptr) ? kOk : static_cast<Code>(state_->code);
   }
  private:
-  struct FreeDeleter {
-    void operator()(void* ptr) const {
-      free(ptr);
-    }
-  };
-
   struct State {
+    std::atomic<size_t> counter;
     uint32_t message_len;
     uint8_t code;
     int64_t error_code;
@@ -266,38 +254,26 @@ class Status {
     const char* file_name;
     int line_number;
     char message[1];
-  } __attribute__ ((packed));
+  };
+
+  typedef boost::intrusive_ptr<State> StatePtr;
 
-  typedef std::unique_ptr<State, FreeDeleter> StatePtr;
+  friend inline void intrusive_ptr_release(State* state) {
+    if (state->counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
+      free(state);
+    }
+  }
+
+  friend inline void intrusive_ptr_add_ref(State* state) {
+    state->counter.fetch_add(1, std::memory_order_relaxed);
+  }
 
   StatePtr state_;
   static constexpr size_t kHeaderSize = offsetof(State, message);
 
   static_assert(sizeof(Code) == 4, "Code enum size is part of abi");
-
-  StatePtr CopyState() const;
 };
 
-inline Status::Status(const Status& s)
-    : state_(s.CopyState()) {
-}
-
-inline void Status::operator=(const Status& s) {
-  // The following condition catches both aliasing (when this == &s),
-  // and the common case where both s and *this are ok.
-  if (state_ != s.state_) {
-    state_ = s.CopyState();
-  }
-}
-
-inline Status::Status(Status&& s)
-    : state_(std::move(s.state_)) {
-}
-
-inline void Status::operator=(Status&& s) {
-  state_ = std::move(s.state_);
-}
-
 inline Status&& MoveStatus(Status&& status) {
   return std::move(status);
 }