Avoid malloc in NotFound key status if no message is given.

Summary:
In some places we have NotFound status created with empty message, but it doesn't avoid a malloc. With this patch, the malloc is avoided for that case.

The motivation of it is that I found in db_bench readrandom test when all keys are not existing, about 4% of the total running time is spent on malloc of Status, plus a similar amount of CPU spent on free of them, which is not necessary.

Test Plan: make all check

Reviewers: dhruba, haobo, igor

Reviewed By: haobo

CC: leveldb

Differential Revision: https://reviews.facebook.net/D14691

db/memtable.cc

@@ -225,7 +225,7 @@ bool MemTable::Get(const LookupKey& key, std::string* value, Status* s,
               *s = Status::Corruption("Error: Could not perform merge.");
             }
           } else {
-            *s = Status::NotFound(Slice());
+            *s = Status::NotFound();
           }
           return true;
         }

db/version_set.cc

@@ -545,7 +545,7 @@ void Version::Get(const ReadOptions& options,
         case kFound:
           return;
         case kDeleted:
-          *status = Status::NotFound(Slice());  // Use empty error message for speed
+          *status = Status::NotFound();  // Use empty error message for speed
           return;
         case kCorrupt:
           *status = Status::Corruption("corrupted key for ", user_key);
@@ -570,7 +570,7 @@ void Version::Get(const ReadOptions& options,
                                    user_key);
     }
   } else {
-    *status = Status::NotFound(Slice()); // Use an empty error message for speed
+    *status = Status::NotFound(); // Use an empty error message for speed
   }
 }
 

include/rocksdb/status.h

@@ -25,7 +25,7 @@ namespace rocksdb {
 class Status {
  public:
   // Create a success status.
-  Status() : state_(nullptr) { }
+  Status() : code_(kOk), state_(nullptr) { }
   ~Status() { delete[] state_; }
 
   // Copy the specified status.
@@ -39,6 +39,10 @@ class Status {
   static Status NotFound(const Slice& msg, const Slice& msg2 = Slice()) {
     return Status(kNotFound, msg, msg2);
   }
+  // Fast path for not found without malloc;
+  static Status NotFound() {
+    return Status(kNotFound);
+  }
   static Status Corruption(const Slice& msg, const Slice& msg2 = Slice()) {
     return Status(kCorruption, msg, msg2);
   }
@@ -59,7 +63,7 @@ class Status {
   }
 
   // Returns true iff the status indicates success.
-  bool ok() const { return (state_ == nullptr); }
+  bool ok() const { return code() == kOk; }
 
   // Returns true iff the status indicates a NotFound error.
   bool IsNotFound() const { return code() == kNotFound; }
@@ -87,13 +91,6 @@ class Status {
   std::string ToString() const;
 
  private:
-  // OK status has a nullptr state_.  Otherwise, state_ is a new[] array
-  // of the following form:
-  //    state_[0..3] == length of message
-  //    state_[4]    == code
-  //    state_[5..]  == message
-  const char* state_;
-
   enum Code {
     kOk = 0,
     kNotFound = 1,
@@ -105,20 +102,30 @@ class Status {
     kIncomplete = 7
   };
 
+  // A nullptr state_ (which is always the case for OK) means the message
+  // is empty.
+  // of the following form:
+  //    state_[0..3] == length of message
+  //    state_[4..]  == message
+  Code code_;
+  const char* state_;
+
   Code code() const {
-    return (state_ == nullptr) ? kOk : static_cast<Code>(state_[4]);
+    return code_;
   }
-
+  explicit Status(Code code) : code_(code), state_(nullptr) { }
   Status(Code code, const Slice& msg, const Slice& msg2);
   static const char* CopyState(const char* s);
 };
 
 inline Status::Status(const Status& s) {
+  code_ = s.code_;
   state_ = (s.state_ == nullptr) ? nullptr : CopyState(s.state_);
 }
 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.
+  code_ = s.code_;
   if (state_ != s.state_) {
     delete[] state_;
     state_ = (s.state_ == nullptr) ? nullptr : CopyState(s.state_);

util/status.cc

@@ -16,38 +16,34 @@ namespace rocksdb {
 const char* Status::CopyState(const char* state) {
   uint32_t size;
   memcpy(&size, state, sizeof(size));
-  char* result = new char[size + 5];
+  char* result = new char[size + 4];
-  memcpy(result, state, size + 5);
+  memcpy(result, state, size + 4);
   return result;
 }
 
-Status::Status(Code code, const Slice& msg, const Slice& msg2) {
+Status::Status(Code code, const Slice& msg, const Slice& msg2) :
+    code_(code) {
   assert(code != kOk);
   const uint32_t len1 = msg.size();
   const uint32_t len2 = msg2.size();
   const uint32_t size = len1 + (len2 ? (2 + len2) : 0);
-  char* result = new char[size + 5];
+  char* result = new char[size + 4];
   memcpy(result, &size, sizeof(size));
-  result[4] = static_cast<char>(code);
+  memcpy(result + 4, msg.data(), len1);
-  memcpy(result + 5, msg.data(), len1);
   if (len2) {
-    result[5 + len1] = ':';
+    result[4 + len1] = ':';
-    result[6 + len1] = ' ';
+    result[5 + len1] = ' ';
-    memcpy(result + 7 + len1, msg2.data(), len2);
+    memcpy(result + 6 + len1, msg2.data(), len2);
   }
   state_ = result;
 }
 
 std::string Status::ToString() const {
-  if (state_ == nullptr) {
-    return "OK";
-  } else {
   char tmp[30];
   const char* type;
-    switch (code()) {
+  switch (code_) {
     case kOk:
-        type = "OK";
+      return "OK";
-        break;
     case kNotFound:
       type = "NotFound: ";
       break;
@@ -73,11 +69,12 @@ std::string Status::ToString() const {
       break;
   }
   std::string result(type);
+  if (state_ != nullptr) {
     uint32_t length;
     memcpy(&length, state_, sizeof(length));
-    result.append(state_ + 5, length);
+    result.append(state_ + 4, length);
-    return result;
   }
+  return result;
 }
 
 }  // namespace rocksdb