use mmap on 64-bit machines to speed-up reads; small build fixes

6 files changed, 35 insertions, 17 deletions

diff --git a/table/block.cc b/table/block.cc
index 40aa318..06eb6f8 100644
--- a/table/block.cc
+++ b/table/block.cc
@@ -19,9 +19,10 @@ inline uint32_t Block::NumRestarts() const {
   return DecodeFixed32(data_ + size_ - sizeof(uint32_t));
 }
 
-Block::Block(const char* data, size_t size)
+Block::Block(const char* data, size_t size, bool take_ownership)
     : data_(data),
-      size_(size) {
+      size_(size),
+      owned_(take_ownership) {
   if (size_ < sizeof(uint32_t)) {
     size_ = 0;  // Error marker
   } else {
@@ -35,7 +36,9 @@ Block::Block(const char* data, size_t size)
 }
 
 Block::~Block() {
-  delete[] data_;
+  if (owned_) {
+    delete[] data_;
+  }
 }
 
 // Helper routine: decode the next block entry starting at "p",
diff --git a/table/block.h b/table/block.h
index 9eb6f02..76088a4 100644
--- a/table/block.h
+++ b/table/block.h
@@ -16,8 +16,9 @@ class Comparator;
 class Block {
  public:
   // Initialize the block with the specified contents.
-  // Takes ownership of data[] and will delete[] it when done.
-  Block(const char* data, size_t size);
+  // Takes ownership of data[] and will delete[] it when done iff
+  // "take_ownership is true.
+  Block(const char* data, size_t size, bool take_ownership);
 
   ~Block();
 
@@ -30,6 +31,7 @@ class Block {
   const char* data_;
   size_t size_;
   uint32_t restart_offset_;     // Offset in data_ of restart array
+  bool owned_;                  // Block owns data_[]
 
   // No copying allowed
   Block(const Block&);
diff --git a/table/format.cc b/table/format.cc
index 23728d8..25b85a2 100644
--- a/table/format.cc
+++ b/table/format.cc
@@ -66,8 +66,10 @@ Status Footer::DecodeFrom(Slice* input) {
 Status ReadBlock(RandomAccessFile* file,
                  const ReadOptions& options,
                  const BlockHandle& handle,
-                 Block** block) {
+                 Block** block,
+                 bool* may_cache) {
   *block = NULL;
+  *may_cache = false;
 
   // Read the block contents as well as the type/crc footer.
   // See table_builder.cc for the code that built this structure.
@@ -100,8 +102,14 @@ Status ReadBlock(RandomAccessFile* file,
     case kNoCompression:
       if (data != buf) {
         // File implementation gave us pointer to some other data.
-        // Copy into buf[].
-        memcpy(buf, data, n + kBlockTrailerSize);
+        // Use it directly under the assumption that it will be live
+        // while the file is open.
+        delete[] buf;
+        *block = new Block(data, n, false /* do not take ownership */);
+        *may_cache = false;  // Do not double-cache
+      } else {
+        *block = new Block(buf, n, true /* take ownership */);
+        *may_cache = true;
       }
 
       // Ok
@@ -119,8 +127,8 @@ Status ReadBlock(RandomAccessFile* file,
         return Status::Corruption("corrupted compressed block contents");
       }
       delete[] buf;
-      buf = ubuf;
-      n = ulength;
+      *block = new Block(ubuf, ulength, true /* take ownership */);
+      *may_cache = true;
       break;
     }
     default:
@@ -128,7 +136,6 @@ Status ReadBlock(RandomAccessFile* file,
       return Status::Corruption("bad block type");
   }
 
-  *block = new Block(buf, n);  // Block takes ownership of buf[]
   return Status::OK();
 }
 
diff --git a/table/format.h b/table/format.h
index 2a3e1ac..66a15da 100644
--- a/table/format.h
+++ b/table/format.h
@@ -86,10 +86,13 @@ static const size_t kBlockTrailerSize = 5;
 // Read the block identified by "handle" from "file".  On success,
 // store a pointer to the heap-allocated result in *block and return
 // OK.  On failure store NULL in *block and return non-OK.
+// On success, stores true in *may_cache if the result may be
+// cached, false if it must not be cached.
 extern Status ReadBlock(RandomAccessFile* file,
                         const ReadOptions& options,
                         const BlockHandle& handle,
-                        Block** block);
+                        Block** block,
+                        bool* may_cache);
 
 // Implementation details follow.  Clients should ignore,
 
diff --git a/table/table.cc b/table/table.cc
index 5f9238e..07dcffd 100644
--- a/table/table.cc
+++ b/table/table.cc
@@ -49,7 +49,9 @@ Status Table::Open(const Options& options,
   // Read the index block
   Block* index_block = NULL;
   if (s.ok()) {
-    s = ReadBlock(file, ReadOptions(), footer.index_handle(), &index_block);
+    bool may_cache;  // Ignored result
+    s = ReadBlock(file, ReadOptions(), footer.index_handle(), &index_block,
+                  &may_cache);
   }
 
   if (s.ok()) {
@@ -105,6 +107,7 @@ Iterator* Table::BlockReader(void* arg,
   // can add more features in the future.
 
   if (s.ok()) {
+    bool may_cache;
     if (block_cache != NULL) {
       char cache_key_buffer[16];
       EncodeFixed64(cache_key_buffer, table->rep_->cache_id);
@@ -114,14 +117,14 @@ Iterator* Table::BlockReader(void* arg,
       if (cache_handle != NULL) {
         block = reinterpret_cast<Block*>(block_cache->Value(cache_handle));
       } else {
-        s = ReadBlock(table->rep_->file, options, handle, &block);
-        if (s.ok() && options.fill_cache) {
+        s = ReadBlock(table->rep_->file, options, handle, &block, &may_cache);
+        if (s.ok() && may_cache && options.fill_cache) {
           cache_handle = block_cache->Insert(
               key, block, block->size(), &DeleteCachedBlock);
         }
       }
     } else {
-      s = ReadBlock(table->rep_->file, options, handle, &block);
+      s = ReadBlock(table->rep_->file, options, handle, &block, &may_cache);
     }
   }
 
diff --git a/table/table_test.cc b/table/table_test.cc
index cd85b4b..0c8e676 100644
--- a/table/table_test.cc
+++ b/table/table_test.cc
@@ -205,7 +205,7 @@ class BlockConstructor: public Constructor {
     block_size_ = block_data.size();
     char* block_data_copy = new char[block_size_];
     memcpy(block_data_copy, block_data.data(), block_size_);
-    block_ = new Block(block_data_copy, block_size_);
+    block_ = new Block(block_data_copy, block_size_, true /* take ownership */);
     return Status::OK();
   }
   virtual size_t NumBytes() const { return block_size_; }