4 #if !defined(LEVELDB_PLATFORM_WINDOWS)
18 #include <sys/types.h>
21 #if defined(LEVELDB_PLATFORM_ANDROID)
35 static Status IOError(
const std::string& context,
int err_number) {
39 class PosixSequentialFile:
public SequentialFile {
45 PosixSequentialFile(
const std::string& fname, FILE* f)
46 : filename_(fname), file_(f) { }
47 virtual ~PosixSequentialFile() { fclose(file_); }
49 virtual Status Read(
size_t n, Slice* result,
char* scratch) {
51 size_t r = fread_unlocked(scratch, 1, n, file_);
52 *result = Slice(scratch, r);
58 s = IOError(filename_, errno);
65 if (fseek(file_, n, SEEK_CUR)) {
66 return IOError(filename_, errno);
73 class PosixRandomAccessFile:
public RandomAccessFile {
79 PosixRandomAccessFile(
const std::string& fname,
int fd)
80 : filename_(fname), fd_(fd) { }
81 virtual ~PosixRandomAccessFile() { close(fd_); }
83 virtual Status Read(
uint64_t offset,
size_t n, Slice* result,
84 char* scratch)
const {
86 ssize_t r = pread(fd_, scratch, n, static_cast<off_t>(offset));
87 *result = Slice(scratch, (r < 0) ? 0 : r);
90 s = IOError(filename_, errno);
103 SetAllowed(
sizeof(
void*) >= 8 ? 1000 : 0);
109 if (GetAllowed() <= 0) {
113 intptr_t x = GetAllowed();
125 SetAllowed(GetAllowed() + 1);
132 intptr_t GetAllowed()
const {
133 return reinterpret_cast<intptr_t
>(allowed_.Acquire_Load());
137 void SetAllowed(intptr_t v) {
138 allowed_.Release_Store(reinterpret_cast<void*>(v));
141 MmapLimiter(
const MmapLimiter&);
142 void operator=(
const MmapLimiter&);
146 class PosixMmapReadableFile:
public RandomAccessFile {
155 PosixMmapReadableFile(
const std::string& fname,
void*
base,
size_t length,
156 MmapLimiter* limiter)
157 : filename_(fname), mmapped_region_(base), length_(length),
161 virtual ~PosixMmapReadableFile() {
162 munmap(mmapped_region_, length_);
166 virtual Status Read(
uint64_t offset,
size_t n, Slice* result,
167 char* scratch)
const {
169 if (offset + n > length_) {
171 s = IOError(filename_, EINVAL);
173 *result = Slice(reinterpret_cast<char*>(mmapped_region_) + offset, n);
183 class PosixMmapFile :
public WritableFile {
199 static size_t Roundup(
size_t x,
size_t y) {
200 return ((x + y - 1) / y) * y;
203 size_t TruncateToPageBoundary(
size_t s) {
204 s -= (s & (page_size_ - 1));
205 assert((s % page_size_) == 0);
209 bool UnmapCurrentRegion() {
212 #if defined(OS_MACOSX)
213 if (msync(base_, limit_ - base_, MS_SYNC) != 0) {
217 if (last_sync_ < limit_) {
219 pending_sync_ =
true;
221 if (munmap(base_, limit_ - base_) != 0) {
224 file_offset_ += limit_ -
base_;
231 if (map_size_ < (1<<20)) {
238 bool MapNewRegion() {
239 assert(base_ == NULL);
240 if (ftruncate(fd_, file_offset_ + map_size_) < 0) {
243 void* ptr = mmap(NULL, map_size_, PROT_READ | PROT_WRITE, MAP_SHARED,
245 if (ptr == MAP_FAILED) {
248 base_ =
reinterpret_cast<char*
>(ptr);
256 PosixMmapFile(
const std::string& fname,
int fd,
size_t page_size)
259 page_size_(page_size),
260 map_size_(Roundup(65536, page_size)),
266 pending_sync_(false) {
267 assert((page_size & (page_size - 1)) == 0);
273 PosixMmapFile::Close();
277 virtual Status Append(
const Slice& data) {
278 const char* src = data.data();
279 size_t left = data.size();
281 assert(base_ <= dst_);
282 assert(dst_ <= limit_);
283 size_t avail = limit_ -
dst_;
285 if (!UnmapCurrentRegion() ||
287 return IOError(filename_, errno);
291 size_t n = (left <= avail) ? left : avail;
292 memcpy(dst_, src, n);
300 virtual Status Close() {
302 size_t unused = limit_ -
dst_;
303 if (!UnmapCurrentRegion()) {
304 s = IOError(filename_, errno);
305 }
else if (unused > 0) {
307 if (ftruncate(fd_, file_offset_ - unused) < 0) {
308 s = IOError(filename_, errno);
312 if (close(fd_) < 0) {
314 s = IOError(filename_, errno);
324 virtual Status Flush() {
328 Status SyncDirIfManifest() {
329 const char* f = filename_.c_str();
330 const char* sep = strrchr(f,
'/');
337 dir = std::string(f, sep - f);
341 if (basename.starts_with(
"MANIFEST")) {
342 int fd = open(dir.c_str(), O_RDONLY);
344 s = IOError(dir, errno);
347 s = IOError(dir, errno);
355 virtual Status Sync() {
357 Status s = SyncDirIfManifest();
364 pending_sync_ =
false;
365 if (fdatasync(fd_) < 0) {
366 s = IOError(filename_, errno);
370 if (dst_ > last_sync_) {
373 size_t p1 = TruncateToPageBoundary(last_sync_ - base_);
374 size_t p2 = TruncateToPageBoundary(dst_ - base_ - 1);
376 if (msync(base_ + p1, p2 - p1 + page_size_, MS_SYNC) < 0) {
377 s = IOError(filename_, errno);
385 class PosixWriteableFile :
public WritableFile {
390 PosixWriteableFile(
const std::string& fname,
int fd)
396 ~PosixWriteableFile() {
398 PosixWriteableFile::Close();
402 virtual Status Append(
const Slice& data) {
405 ret =
write(fd_, data.data(), data.size());
407 s = IOError(filename_, errno);
408 }
else if (ret < data.size()) {
415 virtual Status Close() {
417 if (close(fd_) < 0) {
418 s = IOError(filename_, errno);
424 virtual Status Flush() {
428 Status SyncDirIfManifest() {
429 const char* f = filename_.c_str();
430 const char* sep = strrchr(f,
'/');
437 dir = std::string(f, sep - f);
441 if (basename.starts_with(
"MANIFEST")) {
442 int fd = open(dir.c_str(), O_RDONLY);
444 s = IOError(dir, errno);
447 s = IOError(dir, errno);
455 virtual Status Sync() {
457 Status s = SyncDirIfManifest();
462 if (fdatasync(fd_) < 0) {
463 s = IOError(filename_, errno);
470 static int LockOrUnlock(
int fd,
bool lock) {
473 memset(&f, 0,
sizeof(f));
474 f.l_type = (lock ? F_WRLCK : F_UNLCK);
475 f.l_whence = SEEK_SET;
478 return fcntl(fd, F_SETLK, &f);
481 class PosixFileLock :
public FileLock {
490 class PosixLockTable {
495 bool Insert(
const std::string& fname) {
497 return locked_files_.insert(fname).second;
499 void Remove(
const std::string& fname) {
501 locked_files_.erase(fname);
505 class PosixEnv :
public Env {
508 virtual ~PosixEnv() {
509 fprintf(stderr,
"Destroying Env::Default()\n");
513 virtual Status NewSequentialFile(
const std::string& fname,
514 SequentialFile** result) {
515 FILE* f = fopen(fname.c_str(),
"r");
518 return IOError(fname, errno);
520 *result =
new PosixSequentialFile(fname, f);
525 virtual Status NewRandomAccessFile(
const std::string& fname,
526 RandomAccessFile** result) {
529 int fd = open(fname.c_str(), O_RDONLY);
531 s = IOError(fname, errno);
533 *result =
new PosixRandomAccessFile(fname, fd);
538 virtual Status NewWritableFile(
const std::string& fname,
539 WritableFile** result) {
541 const int fd = open(fname.c_str(), O_CREAT | O_RDWR | O_TRUNC, 0644);
544 s = IOError(fname, errno);
546 *result =
new PosixWriteableFile(fname, fd);
551 virtual bool FileExists(
const std::string& fname) {
552 return access(fname.c_str(), F_OK) == 0;
555 virtual Status GetChildren(
const std::string& dir,
556 std::vector<std::string>* result) {
558 DIR* d = opendir(dir.c_str());
560 return IOError(dir, errno);
562 struct dirent* entry;
563 while ((entry = readdir(d)) != NULL) {
564 result->push_back(entry->d_name);
570 virtual Status DeleteFile(
const std::string& fname) {
572 if (unlink(fname.c_str()) != 0) {
573 result = IOError(fname, errno);
578 virtual Status CreateDir(
const std::string&
name) {
580 if (mkdir(name.c_str(), 0755) != 0) {
581 result = IOError(name, errno);
586 virtual Status DeleteDir(
const std::string& name) {
588 if (rmdir(name.c_str()) != 0) {
589 result = IOError(name, errno);
594 virtual Status GetFileSize(
const std::string& fname,
uint64_t* size) {
597 if (stat(fname.c_str(), &sbuf) != 0) {
599 s = IOError(fname, errno);
601 *size = sbuf.st_size;
606 virtual Status RenameFile(
const std::string& src,
const std::string& target) {
608 if (rename(src.c_str(), target.c_str()) != 0) {
609 result = IOError(src, errno);
614 virtual Status LockFile(
const std::string& fname, FileLock** lock) {
617 int fd = open(fname.c_str(), O_RDWR | O_CREAT, 0644);
619 result = IOError(fname, errno);
620 }
else if (!
locks_.Insert(fname)) {
623 }
else if (LockOrUnlock(fd,
true) == -1) {
624 result = IOError(
"lock " + fname, errno);
628 PosixFileLock* my_lock =
new PosixFileLock;
630 my_lock->name_ = fname;
636 virtual Status UnlockFile(FileLock* lock) {
637 PosixFileLock* my_lock =
reinterpret_cast<PosixFileLock*
>(lock);
639 if (LockOrUnlock(my_lock->fd_,
false) == -1) {
640 result = IOError(
"unlock", errno);
642 locks_.Remove(my_lock->name_);
648 virtual void Schedule(
void (*
function)(
void*),
void*
arg);
650 virtual void StartThread(
void (*
function)(
void*
arg),
void* arg);
652 virtual Status GetTestDirectory(std::string* result) {
653 const char* env = getenv(
"TEST_TMPDIR");
654 if (env && env[0] !=
'\0') {
658 snprintf(buf,
sizeof(buf),
"/tmp/leveldbtest-%d",
int(geteuid()));
667 pthread_t
tid = pthread_self();
669 memcpy(&thread_id, &tid, std::min(
sizeof(thread_id),
sizeof(tid)));
673 virtual Status NewLogger(
const std::string& fname, Logger** result) {
674 FILE* f = fopen(fname.c_str(),
"w");
677 return IOError(fname, errno);
679 *result =
new PosixLogger(f, &PosixEnv::gettid);
686 gettimeofday(&tv, NULL);
687 return static_cast<uint64_t>(tv.tv_sec) * 1000000 + tv.tv_usec;
690 virtual void SleepForMicroseconds(
int micros) {
695 void PthreadCall(
const char* label,
int result) {
697 fprintf(stderr,
"pthread %s: %s\n", label, strerror(result));
704 static void* BGThreadWrapper(
void*
arg) {
705 reinterpret_cast<PosixEnv*
>(
arg)->BGThread();
716 struct BGItem {
void*
arg; void (*
function)(
void*); };
717 typedef std::deque<BGItem> BGQueue;
724 PosixEnv::PosixEnv() : page_size_(getpagesize()),
726 PthreadCall(
"mutex_init", pthread_mutex_init(&mu_, NULL));
727 PthreadCall(
"cvar_init", pthread_cond_init(&
bgsignal_, NULL));
730 void PosixEnv::Schedule(
void (*
function)(
void*),
void*
arg) {
731 PthreadCall(
"lock", pthread_mutex_lock(&mu_));
738 pthread_create(&
bgthread_, NULL, &PosixEnv::BGThreadWrapper,
this));
744 PthreadCall(
"signal", pthread_cond_signal(&
bgsignal_));
748 queue_.push_back(BGItem());
749 queue_.back().function =
function;
752 PthreadCall(
"unlock", pthread_mutex_unlock(&mu_));
755 void PosixEnv::BGThread() {
758 PthreadCall(
"lock", pthread_mutex_lock(&mu_));
760 PthreadCall(
"wait", pthread_cond_wait(&
bgsignal_, &mu_));
763 void (*
function)(
void*) =
queue_.front().function;
764 void* arg =
queue_.front().arg;
767 PthreadCall(
"unlock", pthread_mutex_unlock(&mu_));
773 struct StartThreadState {
778 static void* StartThreadWrapper(
void* arg) {
779 StartThreadState*
state =
reinterpret_cast<StartThreadState*
>(
arg);
780 state->user_function(state->arg);
785 void PosixEnv::StartThread(
void (*
function)(
void* arg),
void* arg) {
787 StartThreadState*
state =
new StartThreadState;
788 state->user_function =
function;
790 PthreadCall(
"start thread",
791 pthread_create(&t, NULL, &StartThreadWrapper, state));
796 static pthread_once_t once = PTHREAD_ONCE_INIT;
797 static Env* default_env;
798 static void InitDefaultEnv() { default_env =
new PosixEnv; }
801 pthread_once(&once, InitDefaultEnv);
void(* user_function)(void *)
std::set< std::string > locked_files_
port::AtomicPointer allowed_
unsigned long long uint64_t
void write(const Value &value, std::ostream &os)
static Status IOError(const Slice &msg, const Slice &msg2=Slice())