src/leveldb/util/env_posix.cc

   1 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   4 #if !defined(LEVELDB_PLATFORM_WINDOWS)
   5
   6 #include <deque>
   7 #include <set>
   8 #include <dirent.h>
   9 #include <errno.h>
  10 #include <fcntl.h>
  11 #include <pthread.h>
  12 #include <stdio.h>
  13 #include <stdlib.h>
  14 #include <string.h>
  15 #include <sys/mman.h>
  16 #include <sys/stat.h>
  17 #include <sys/time.h>
  18 #include <sys/types.h>
  19 #include <time.h>
  20 #include <unistd.h>
  21 #if defined(LEVELDB_PLATFORM_ANDROID)
  22 #include <sys/stat.h>
  23 #endif
  24 #include "leveldb/env.h"
  25 #include "leveldb/slice.h"
  26 #include "port/port.h"
  27 #include "util/logging.h"
  28 #include "util/mutexlock.h"
  29 #include "util/posix_logger.h"
  30
  31 namespace leveldb {
  32
  33 namespace {
  34
  35 static Status IOError(const std::string& context, int err_number) {
  36   return Status::IOError(context, strerror(err_number));
  37 }
  38
  39 class PosixSequentialFile: public SequentialFile {
  40  private:
  41   std::string filename_;
  42   FILE* file_;
  43
  44  public:
  45   PosixSequentialFile(const std::string& fname, FILE* f)
  46       : filename_(fname), file_(f) { }
  47   virtual ~PosixSequentialFile() { fclose(file_); }
  48
  49   virtual Status Read(size_t n, Slice* result, char* scratch) {
  50     Status s;
  51     size_t r = fread_unlocked(scratch, 1, n, file_);
  52     *result = Slice(scratch, r);
  53     if (r < n) {
  54       if (feof(file_)) {
  55         // We leave status as ok if we hit the end of the file
  56       } else {
  57         // A partial read with an error: return a non-ok status
  58         s = IOError(filename_, errno);
  59       }
  60     }
  61     return s;
  62   }
  63
  64   virtual Status Skip(uint64_t n) {
  65     if (fseek(file_, n, SEEK_CUR)) {
  66       return IOError(filename_, errno);
  67     }
  68     return Status::OK();
  69   }
  70 };
  71
  72 // pread() based random-access
  73 class PosixRandomAccessFile: public RandomAccessFile {
  74  private:
  75   std::string filename_;
  76   int fd_;
  77
  78  public:
  79   PosixRandomAccessFile(const std::string& fname, int fd)
  80       : filename_(fname), fd_(fd) { }
  81   virtual ~PosixRandomAccessFile() { close(fd_); }
  82
  83   virtual Status Read(uint64_t offset, size_t n, Slice* result,
  84                       char* scratch) const {
  85     Status s;
  86     ssize_t r = pread(fd_, scratch, n, static_cast<off_t>(offset));
  87     *result = Slice(scratch, (r < 0) ? 0 : r);
  88     if (r < 0) {
  89       // An error: return a non-ok status
  90       s = IOError(filename_, errno);
  91     }
  92     return s;
  93   }
  94 };
  95
  96 // Helper class to limit mmap file usage so that we do not end up
  97 // running out virtual memory or running into kernel performance
  98 // problems for very large databases.
  99 class MmapLimiter {
 100  public:
 101   // Up to 1000 mmaps for 64-bit binaries; none for smaller pointer sizes.
 102   MmapLimiter() {
 103     SetAllowed(sizeof(void*) >= 8 ? 1000 : 0);
 104   }
 105
 106   // If another mmap slot is available, acquire it and return true.
 107   // Else return false.
 108   bool Acquire() {
 109     if (GetAllowed() <= 0) {
 110       return false;
 111     }
 112     MutexLock l(&mu_);
 113     intptr_t x = GetAllowed();
 114     if (x <= 0) {
 115       return false;
 116     } else {
 117       SetAllowed(x - 1);
 118       return true;
 119     }
 120   }
 121
 122   // Release a slot acquired by a previous call to Acquire() that returned true.
 123   void Release() {
 124     MutexLock l(&mu_);
 125     SetAllowed(GetAllowed() + 1);
 126   }
 127
 128  private:
 129   port::Mutex mu_;
 130   port::AtomicPointer allowed_;
 131
 132   intptr_t GetAllowed() const {
 133     return reinterpret_cast<intptr_t>(allowed_.Acquire_Load());
 134   }
 135
 136   // REQUIRES: mu_ must be held
 137   void SetAllowed(intptr_t v) {
 138     allowed_.Release_Store(reinterpret_cast<void*>(v));
 139   }
 140
 141   MmapLimiter(const MmapLimiter&);
 142   void operator=(const MmapLimiter&);
 143 };
 144
 145 // mmap() based random-access
 146 class PosixMmapReadableFile: public RandomAccessFile {
 147  private:
 148   std::string filename_;
 149   void* mmapped_region_;
 150   size_t length_;
 151   MmapLimiter* limiter_;
 152
 153  public:
 154   // base[0,length-1] contains the mmapped contents of the file.
 155   PosixMmapReadableFile(const std::string& fname, void* base, size_t length,
 156                         MmapLimiter* limiter)
 157       : filename_(fname), mmapped_region_(base), length_(length),
 158         limiter_(limiter) {
 159   }
 160
 161   virtual ~PosixMmapReadableFile() {
 162     munmap(mmapped_region_, length_);
 163     limiter_->Release();
 164   }
 165
 166   virtual Status Read(uint64_t offset, size_t n, Slice* result,
 167                       char* scratch) const {
 168     Status s;
 169     if (offset + n > length_) {
 170       *result = Slice();
 171       s = IOError(filename_, EINVAL);
 172     } else {
 173       *result = Slice(reinterpret_cast<char*>(mmapped_region_) + offset, n);
 174     }
 175     return s;
 176   }
 177 };
 178
 179 // We preallocate up to an extra megabyte and use memcpy to append new
 180 // data to the file.  This is safe since we either properly close the
 181 // file before reading from it, or for log files, the reading code
 182 // knows enough to skip zero suffixes.
 183 class PosixMmapFile : public WritableFile {
 184  private:
 185   std::string filename_;
 186   int fd_;
 187   size_t page_size_;
 188   size_t map_size_;       // How much extra memory to map at a time
 189   char* base_;            // The mapped region
 190   char* limit_;           // Limit of the mapped region
 191   char* dst_;             // Where to write next  (in range [base_,limit_])
 192   char* last_sync_;       // Where have we synced up to
 193   uint64_t file_offset_;  // Offset of base_ in file
 194
 195   // Have we done an munmap of unsynced data?
 196   bool pending_sync_;
 197
 198   // Roundup x to a multiple of y
 199   static size_t Roundup(size_t x, size_t y) {
 200     return ((x + y - 1) / y) * y;
 201   }
 202
 203   size_t TruncateToPageBoundary(size_t s) {
 204     s -= (s & (page_size_ - 1));
 205     assert((s % page_size_) == 0);
 206     return s;
 207   }
 208
 209   bool UnmapCurrentRegion() {
 210     bool result = true;
 211     if (base_ != NULL) {
 212       if (last_sync_ < limit_) {
 213         // Defer syncing this data until next Sync() call, if any
 214         pending_sync_ = true;
 215       }
 216       if (munmap(base_, limit_ - base_) != 0) {
 217         result = false;
 218       }
 219       file_offset_ += limit_ - base_;
 220       base_ = NULL;
 221       limit_ = NULL;
 222       last_sync_ = NULL;
 223       dst_ = NULL;
 224
 225       // Increase the amount we map the next time, but capped at 1MB
 226       if (map_size_ < (1<<20)) {
 227         map_size_ *= 2;
 228       }
 229     }
 230     return result;
 231   }
 232
 233   bool MapNewRegion() {
 234     assert(base_ == NULL);
 235     if (ftruncate(fd_, file_offset_ + map_size_) < 0) {
 236       return false;
 237     }
 238     void* ptr = mmap(NULL, map_size_, PROT_READ | PROT_WRITE, MAP_SHARED,
 239                      fd_, file_offset_);
 240     if (ptr == MAP_FAILED) {
 241       return false;
 242     }
 243     base_ = reinterpret_cast<char*>(ptr);
 244     limit_ = base_ + map_size_;
 245     dst_ = base_;
 246     last_sync_ = base_;
 247     return true;
 248   }
 249
 250  public:
 251   PosixMmapFile(const std::string& fname, int fd, size_t page_size)
 252       : filename_(fname),
 253         fd_(fd),
 254         page_size_(page_size),
 255         map_size_(Roundup(65536, page_size)),
 256         base_(NULL),
 257         limit_(NULL),
 258         dst_(NULL),
 259         last_sync_(NULL),
 260         file_offset_(0),
 261         pending_sync_(false) {
 262     assert((page_size & (page_size - 1)) == 0);
 263   }
 264
 265
 266   ~PosixMmapFile() {
 267     if (fd_ >= 0) {
 268       PosixMmapFile::Close();
 269     }
 270   }
 271
 272   virtual Status Append(const Slice& data) {
 273     const char* src = data.data();
 274     size_t left = data.size();
 275     while (left > 0) {
 276       assert(base_ <= dst_);
 277       assert(dst_ <= limit_);
 278       size_t avail = limit_ - dst_;
 279       if (avail == 0) {
 280         if (!UnmapCurrentRegion() ||
 281             !MapNewRegion()) {
 282           return IOError(filename_, errno);
 283         }
 284       }
 285
 286       size_t n = (left <= avail) ? left : avail;
 287       memcpy(dst_, src, n);
 288       dst_ += n;
 289       src += n;
 290       left -= n;
 291     }
 292     return Status::OK();
 293   }
 294
 295   virtual Status Close() {
 296     Status s;
 297     size_t unused = limit_ - dst_;
 298     if (!UnmapCurrentRegion()) {
 299       s = IOError(filename_, errno);
 300     } else if (unused > 0) {
 301       // Trim the extra space at the end of the file
 302       if (ftruncate(fd_, file_offset_ - unused) < 0) {
 303         s = IOError(filename_, errno);
 304       }
 305     }
 306
 307     if (close(fd_) < 0) {
 308       if (s.ok()) {
 309         s = IOError(filename_, errno);
 310       }
 311     }
 312
 313     fd_ = -1;
 314     base_ = NULL;
 315     limit_ = NULL;
 316     return s;
 317   }
 318
 319   virtual Status Flush() {
 320     return Status::OK();
 321   }
 322
 323   virtual Status Sync() {
 324     Status s;
 325
 326     if (pending_sync_) {
 327       // Some unmapped data was not synced
 328       pending_sync_ = false;
 329       if (fdatasync(fd_) < 0) {
 330         s = IOError(filename_, errno);
 331       }
 332     }
 333
 334     if (dst_ > last_sync_) {
 335       // Find the beginnings of the pages that contain the first and last
 336       // bytes to be synced.
 337       size_t p1 = TruncateToPageBoundary(last_sync_ - base_);
 338       size_t p2 = TruncateToPageBoundary(dst_ - base_ - 1);
 339       last_sync_ = dst_;
 340       if (msync(base_ + p1, p2 - p1 + page_size_, MS_SYNC) < 0) {
 341         s = IOError(filename_, errno);
 342       }
 343     }
 344
 345     return s;
 346   }
 347 };
 348
 349 static int LockOrUnlock(int fd, bool lock) {
 350   errno = 0;
 351   struct flock f;
 352   memset(&f, 0, sizeof(f));
 353   f.l_type = (lock ? F_WRLCK : F_UNLCK);
 354   f.l_whence = SEEK_SET;
 355   f.l_start = 0;
 356   f.l_len = 0;        // Lock/unlock entire file
 357   return fcntl(fd, F_SETLK, &f);
 358 }
 359
 360 class PosixFileLock : public FileLock {
 361  public:
 362   int fd_;
 363   std::string name_;
 364 };
 365
 366 // Set of locked files.  We keep a separate set instead of just
 367 // relying on fcntrl(F_SETLK) since fcntl(F_SETLK) does not provide
 368 // any protection against multiple uses from the same process.
 369 class PosixLockTable {
 370  private:
 371   port::Mutex mu_;
 372   std::set<std::string> locked_files_;
 373  public:
 374   bool Insert(const std::string& fname) {
 375     MutexLock l(&mu_);
 376     return locked_files_.insert(fname).second;
 377   }
 378   void Remove(const std::string& fname) {
 379     MutexLock l(&mu_);
 380     locked_files_.erase(fname);
 381   }
 382 };
 383
 384 class PosixEnv : public Env {
 385  public:
 386   PosixEnv();
 387   virtual ~PosixEnv() {
 388     fprintf(stderr, "Destroying Env::Default()\n");
 389     abort();
 390   }
 391
 392   virtual Status NewSequentialFile(const std::string& fname,
 393                                    SequentialFile** result) {
 394     FILE* f = fopen(fname.c_str(), "r");
 395     if (f == NULL) {
 396       *result = NULL;
 397       return IOError(fname, errno);
 398     } else {
 399       *result = new PosixSequentialFile(fname, f);
 400       return Status::OK();
 401     }
 402   }
 403
 404   virtual Status NewRandomAccessFile(const std::string& fname,
 405                                      RandomAccessFile** result) {
 406     *result = NULL;
 407     Status s;
 408     int fd = open(fname.c_str(), O_RDONLY);
 409     if (fd < 0) {
 410       s = IOError(fname, errno);
 411     } else if (mmap_limit_.Acquire()) {
 412       uint64_t size;
 413       s = GetFileSize(fname, &size);
 414       if (s.ok()) {
 415         void* base = mmap(NULL, size, PROT_READ, MAP_SHARED, fd, 0);
 416         if (base != MAP_FAILED) {
 417           *result = new PosixMmapReadableFile(fname, base, size, &mmap_limit_);
 418         } else {
 419           s = IOError(fname, errno);
 420         }
 421       }
 422       close(fd);
 423       if (!s.ok()) {
 424         mmap_limit_.Release();
 425       }
 426     } else {
 427       *result = new PosixRandomAccessFile(fname, fd);
 428     }
 429     return s;
 430   }
 431
 432   virtual Status NewWritableFile(const std::string& fname,
 433                                  WritableFile** result) {
 434     Status s;
 435     const int fd = open(fname.c_str(), O_CREAT | O_RDWR | O_TRUNC, 0644);
 436     if (fd < 0) {
 437       *result = NULL;
 438       s = IOError(fname, errno);
 439     } else {
 440       *result = new PosixMmapFile(fname, fd, page_size_);
 441     }
 442     return s;
 443   }
 444
 445   virtual bool FileExists(const std::string& fname) {
 446     return access(fname.c_str(), F_OK) == 0;
 447   }
 448
 449   virtual Status GetChildren(const std::string& dir,
 450                              std::vector<std::string>* result) {
 451     result->clear();
 452     DIR* d = opendir(dir.c_str());
 453     if (d == NULL) {
 454       return IOError(dir, errno);
 455     }
 456     struct dirent* entry;
 457     while ((entry = readdir(d)) != NULL) {
 458       result->push_back(entry->d_name);
 459     }
 460     closedir(d);
 461     return Status::OK();
 462   }
 463
 464   virtual Status DeleteFile(const std::string& fname) {
 465     Status result;
 466     if (unlink(fname.c_str()) != 0) {
 467       result = IOError(fname, errno);
 468     }
 469     return result;
 470   }
 471
 472   virtual Status CreateDir(const std::string& name) {
 473     Status result;
 474     if (mkdir(name.c_str(), 0755) != 0) {
 475       result = IOError(name, errno);
 476     }
 477     return result;
 478   }
 479
 480   virtual Status DeleteDir(const std::string& name) {
 481     Status result;
 482     if (rmdir(name.c_str()) != 0) {
 483       result = IOError(name, errno);
 484     }
 485     return result;
 486   }
 487
 488   virtual Status GetFileSize(const std::string& fname, uint64_t* size) {
 489     Status s;
 490     struct stat sbuf;
 491     if (stat(fname.c_str(), &sbuf) != 0) {
 492       *size = 0;
 493       s = IOError(fname, errno);
 494     } else {
 495       *size = sbuf.st_size;
 496     }
 497     return s;
 498   }
 499
 500   virtual Status RenameFile(const std::string& src, const std::string& target) {
 501     Status result;
 502     if (rename(src.c_str(), target.c_str()) != 0) {
 503       result = IOError(src, errno);
 504     }
 505     return result;
 506   }
 507
 508   virtual Status LockFile(const std::string& fname, FileLock** lock) {
 509     *lock = NULL;
 510     Status result;
 511     int fd = open(fname.c_str(), O_RDWR | O_CREAT, 0644);
 512     if (fd < 0) {
 513       result = IOError(fname, errno);
 514     } else if (!locks_.Insert(fname)) {
 515       close(fd);
 516       result = Status::IOError("lock " + fname, "already held by process");
 517     } else if (LockOrUnlock(fd, true) == -1) {
 518       result = IOError("lock " + fname, errno);
 519       close(fd);
 520       locks_.Remove(fname);
 521     } else {
 522       PosixFileLock* my_lock = new PosixFileLock;
 523       my_lock->fd_ = fd;
 524       my_lock->name_ = fname;
 525       *lock = my_lock;
 526     }
 527     return result;
 528   }
 529
 530   virtual Status UnlockFile(FileLock* lock) {
 531     PosixFileLock* my_lock = reinterpret_cast<PosixFileLock*>(lock);
 532     Status result;
 533     if (LockOrUnlock(my_lock->fd_, false) == -1) {
 534       result = IOError("unlock", errno);
 535     }
 536     locks_.Remove(my_lock->name_);
 537     close(my_lock->fd_);
 538     delete my_lock;
 539     return result;
 540   }
 541
 542   virtual void Schedule(void (*function)(void*), void* arg);
 543
 544   virtual void StartThread(void (*function)(void* arg), void* arg);
 545
 546   virtual Status GetTestDirectory(std::string* result) {
 547     const char* env = getenv("TEST_TMPDIR");
 548     if (env && env[0] != '\0') {
 549       *result = env;
 550     } else {
 551       char buf[100];
 552       snprintf(buf, sizeof(buf), "/tmp/leveldbtest-%d", int(geteuid()));
 553       *result = buf;
 554     }
 555     // Directory may already exist
 556     CreateDir(*result);
 557     return Status::OK();
 558   }
 559
 560   static uint64_t gettid() {
 561     pthread_t tid = pthread_self();
 562     uint64_t thread_id = 0;
 563     memcpy(&thread_id, &tid, std::min(sizeof(thread_id), sizeof(tid)));
 564     return thread_id;
 565   }
 566
 567   virtual Status NewLogger(const std::string& fname, Logger** result) {
 568     FILE* f = fopen(fname.c_str(), "w");
 569     if (f == NULL) {
 570       *result = NULL;
 571       return IOError(fname, errno);
 572     } else {
 573       *result = new PosixLogger(f, &PosixEnv::gettid);
 574       return Status::OK();
 575     }
 576   }
 577
 578   virtual uint64_t NowMicros() {
 579     struct timeval tv;
 580     gettimeofday(&tv, NULL);
 581     return static_cast<uint64_t>(tv.tv_sec) * 1000000 + tv.tv_usec;
 582   }
 583
 584   virtual void SleepForMicroseconds(int micros) {
 585     usleep(micros);
 586   }
 587
 588  private:
 589   void PthreadCall(const char* label, int result) {
 590     if (result != 0) {
 591       fprintf(stderr, "pthread %s: %s\n", label, strerror(result));
 592       abort();
 593     }
 594   }
 595
 596   // BGThread() is the body of the background thread
 597   void BGThread();
 598   static void* BGThreadWrapper(void* arg) {
 599     reinterpret_cast<PosixEnv*>(arg)->BGThread();
 600     return NULL;
 601   }
 602
 603   size_t page_size_;
 604   pthread_mutex_t mu_;
 605   pthread_cond_t bgsignal_;
 606   pthread_t bgthread_;
 607   bool started_bgthread_;
 608
 609   // Entry per Schedule() call
 610   struct BGItem { void* arg; void (*function)(void*); };
 611   typedef std::deque<BGItem> BGQueue;
 612   BGQueue queue_;
 613
 614   PosixLockTable locks_;
 615   MmapLimiter mmap_limit_;
 616 };
 617
 618 PosixEnv::PosixEnv() : page_size_(getpagesize()),
 619                        started_bgthread_(false) {
 620   PthreadCall("mutex_init", pthread_mutex_init(&mu_, NULL));
 621   PthreadCall("cvar_init", pthread_cond_init(&bgsignal_, NULL));
 622 }
 623
 624 void PosixEnv::Schedule(void (*function)(void*), void* arg) {
 625   PthreadCall("lock", pthread_mutex_lock(&mu_));
 626
 627   // Start background thread if necessary
 628   if (!started_bgthread_) {
 629     started_bgthread_ = true;
 630     PthreadCall(
 631         "create thread",
 632         pthread_create(&bgthread_, NULL,  &PosixEnv::BGThreadWrapper, this));
 633   }
 634
 635   // If the queue is currently empty, the background thread may currently be
 636   // waiting.
 637   if (queue_.empty()) {
 638     PthreadCall("signal", pthread_cond_signal(&bgsignal_));
 639   }
 640
 641   // Add to priority queue
 642   queue_.push_back(BGItem());
 643   queue_.back().function = function;
 644   queue_.back().arg = arg;
 645
 646   PthreadCall("unlock", pthread_mutex_unlock(&mu_));
 647 }
 648
 649 void PosixEnv::BGThread() {
 650   while (true) {
 651     // Wait until there is an item that is ready to run
 652     PthreadCall("lock", pthread_mutex_lock(&mu_));
 653     while (queue_.empty()) {
 654       PthreadCall("wait", pthread_cond_wait(&bgsignal_, &mu_));
 655     }
 656
 657     void (*function)(void*) = queue_.front().function;
 658     void* arg = queue_.front().arg;
 659     queue_.pop_front();
 660
 661     PthreadCall("unlock", pthread_mutex_unlock(&mu_));
 662     (*function)(arg);
 663   }
 664 }
 665
 666 namespace {
 667 struct StartThreadState {
 668   void (*user_function)(void*);
 669   void* arg;
 670 };
 671 }
 672 static void* StartThreadWrapper(void* arg) {
 673   StartThreadState* state = reinterpret_cast<StartThreadState*>(arg);
 674   state->user_function(state->arg);
 675   delete state;
 676   return NULL;
 677 }
 678
 679 void PosixEnv::StartThread(void (*function)(void* arg), void* arg) {
 680   pthread_t t;
 681   StartThreadState* state = new StartThreadState;
 682   state->user_function = function;
 683   state->arg = arg;
 684   PthreadCall("start thread",
 685               pthread_create(&t, NULL,  &StartThreadWrapper, state));
 686 }
 687
 688 }  // namespace
 689
 690 static pthread_once_t once = PTHREAD_ONCE_INIT;
 691 static Env* default_env;
 692 static void InitDefaultEnv() { default_env = new PosixEnv; }
 693
 694 Env* Env::Default() {
 695   pthread_once(&once, InitDefaultEnv);
 696   return default_env;
 697 }
 698
 699 }  // namespace leveldb
 700
 701 #endif