Revision 48e7effa79d1ab165e83b2f96455f77ffd79cca3 authored by Siying Dong on 26 March 2019, 02:14:04 UTC, committed by Facebook Github Bot on 26 March 2019, 02:18:04 UTC
Summary: With https://github.com/facebook/rocksdb/pull/3009 we go through every CF to check whether a bottommost compaction is needed to be triggered. This is done within DB mutex. What we do within DB mutex may heavily influece the write throughput we can achieve, so we always want to minimize work there. Here we try to avoid this for-loop by first check a global threshold. In most of the time, the CF loop can be avoided. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5090 Differential Revision: D14582684 Pulled By: siying fbshipit-source-id: 968f6d9bb6affe1a5ebc4910b418300b076f166f
1 parent 52e6404
port_posix.cc
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
// This source code is licensed under both the GPLv2 (found in the
// COPYING file in the root directory) and Apache 2.0 License
// (found in the LICENSE.Apache file in the root directory).
//
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.
#include "port/port_posix.h"
#include <assert.h>
#if defined(__i386__) || defined(__x86_64__)
#include <cpuid.h>
#endif
#include <errno.h>
#include <sched.h>
#include <signal.h>
#include <stdio.h>
#include <string.h>
#include <sys/time.h>
#include <sys/resource.h>
#include <unistd.h>
#include <cstdlib>
#include "util/logging.h"
namespace rocksdb {
// We want to give users opportunity to default all the mutexes to adaptive if
// not specified otherwise. This enables a quick way to conduct various
// performance related experiements.
//
// NB! Support for adaptive mutexes is turned on by definining
// ROCKSDB_PTHREAD_ADAPTIVE_MUTEX during the compilation. If you use RocksDB
// build environment then this happens automatically; otherwise it's up to the
// consumer to define the identifier.
#ifdef ROCKSDB_DEFAULT_TO_ADAPTIVE_MUTEX
extern const bool kDefaultToAdaptiveMutex = true;
#else
extern const bool kDefaultToAdaptiveMutex = false;
#endif
namespace port {
static int PthreadCall(const char* label, int result) {
if (result != 0 && result != ETIMEDOUT) {
fprintf(stderr, "pthread %s: %s\n", label, strerror(result));
abort();
}
return result;
}
Mutex::Mutex(bool adaptive) {
(void) adaptive;
#ifdef ROCKSDB_PTHREAD_ADAPTIVE_MUTEX
if (!adaptive) {
PthreadCall("init mutex", pthread_mutex_init(&mu_, nullptr));
} else {
pthread_mutexattr_t mutex_attr;
PthreadCall("init mutex attr", pthread_mutexattr_init(&mutex_attr));
PthreadCall("set mutex attr",
pthread_mutexattr_settype(&mutex_attr,
PTHREAD_MUTEX_ADAPTIVE_NP));
PthreadCall("init mutex", pthread_mutex_init(&mu_, &mutex_attr));
PthreadCall("destroy mutex attr",
pthread_mutexattr_destroy(&mutex_attr));
}
#else
PthreadCall("init mutex", pthread_mutex_init(&mu_, nullptr));
#endif // ROCKSDB_PTHREAD_ADAPTIVE_MUTEX
}
Mutex::~Mutex() { PthreadCall("destroy mutex", pthread_mutex_destroy(&mu_)); }
void Mutex::Lock() {
PthreadCall("lock", pthread_mutex_lock(&mu_));
#ifndef NDEBUG
locked_ = true;
#endif
}
void Mutex::Unlock() {
#ifndef NDEBUG
locked_ = false;
#endif
PthreadCall("unlock", pthread_mutex_unlock(&mu_));
}
void Mutex::AssertHeld() {
#ifndef NDEBUG
assert(locked_);
#endif
}
CondVar::CondVar(Mutex* mu)
: mu_(mu) {
PthreadCall("init cv", pthread_cond_init(&cv_, nullptr));
}
CondVar::~CondVar() { PthreadCall("destroy cv", pthread_cond_destroy(&cv_)); }
void CondVar::Wait() {
#ifndef NDEBUG
mu_->locked_ = false;
#endif
PthreadCall("wait", pthread_cond_wait(&cv_, &mu_->mu_));
#ifndef NDEBUG
mu_->locked_ = true;
#endif
}
bool CondVar::TimedWait(uint64_t abs_time_us) {
struct timespec ts;
ts.tv_sec = static_cast<time_t>(abs_time_us / 1000000);
ts.tv_nsec = static_cast<suseconds_t>((abs_time_us % 1000000) * 1000);
#ifndef NDEBUG
mu_->locked_ = false;
#endif
int err = pthread_cond_timedwait(&cv_, &mu_->mu_, &ts);
#ifndef NDEBUG
mu_->locked_ = true;
#endif
if (err == ETIMEDOUT) {
return true;
}
if (err != 0) {
PthreadCall("timedwait", err);
}
return false;
}
void CondVar::Signal() {
PthreadCall("signal", pthread_cond_signal(&cv_));
}
void CondVar::SignalAll() {
PthreadCall("broadcast", pthread_cond_broadcast(&cv_));
}
RWMutex::RWMutex() {
PthreadCall("init mutex", pthread_rwlock_init(&mu_, nullptr));
}
RWMutex::~RWMutex() { PthreadCall("destroy mutex", pthread_rwlock_destroy(&mu_)); }
void RWMutex::ReadLock() { PthreadCall("read lock", pthread_rwlock_rdlock(&mu_)); }
void RWMutex::WriteLock() { PthreadCall("write lock", pthread_rwlock_wrlock(&mu_)); }
void RWMutex::ReadUnlock() { PthreadCall("read unlock", pthread_rwlock_unlock(&mu_)); }
void RWMutex::WriteUnlock() { PthreadCall("write unlock", pthread_rwlock_unlock(&mu_)); }
int PhysicalCoreID() {
#if defined(ROCKSDB_SCHED_GETCPU_PRESENT) && defined(__x86_64__) && \
(__GNUC__ > 2 || (__GNUC__ == 2 && __GNUC_MINOR__ >= 22))
// sched_getcpu uses VDSO getcpu() syscall since 2.22. I believe Linux offers VDSO
// support only on x86_64. This is the fastest/preferred method if available.
int cpuno = sched_getcpu();
if (cpuno < 0) {
return -1;
}
return cpuno;
#elif defined(__x86_64__) || defined(__i386__)
// clang/gcc both provide cpuid.h, which defines __get_cpuid(), for x86_64 and i386.
unsigned eax, ebx = 0, ecx, edx;
if (!__get_cpuid(1, &eax, &ebx, &ecx, &edx)) {
return -1;
}
return ebx >> 24;
#else
// give up, the caller can generate a random number or something.
return -1;
#endif
}
void InitOnce(OnceType* once, void (*initializer)()) {
PthreadCall("once", pthread_once(once, initializer));
}
void Crash(const std::string& srcfile, int srcline) {
fprintf(stdout, "Crashing at %s:%d\n", srcfile.c_str(), srcline);
fflush(stdout);
kill(getpid(), SIGTERM);
}
int GetMaxOpenFiles() {
#if defined(RLIMIT_NOFILE)
struct rlimit no_files_limit;
if (getrlimit(RLIMIT_NOFILE, &no_files_limit) != 0) {
return -1;
}
// protect against overflow
if (no_files_limit.rlim_cur >= std::numeric_limits<int>::max()) {
return std::numeric_limits<int>::max();
}
return static_cast<int>(no_files_limit.rlim_cur);
#endif
return -1;
}
void *cacheline_aligned_alloc(size_t size) {
#if __GNUC__ < 5 && defined(__SANITIZE_ADDRESS__)
return malloc(size);
#elif ( _POSIX_C_SOURCE >= 200112L || _XOPEN_SOURCE >= 600 || defined(__APPLE__))
void *m;
errno = posix_memalign(&m, CACHE_LINE_SIZE, size);
return errno ? nullptr : m;
#else
return malloc(size);
#endif
}
void cacheline_aligned_free(void *memblock) {
free(memblock);
}
} // namespace port
} // namespace rocksdb
![swh spinner](/static/img/swh-spinner.gif)
Computing file changes ...