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