d4d338de33
Summary: This diff adds timeout_hint_us to WriteOptions. If it's non-zero, then 1) writes associated with this options MAY be aborted when it has been waiting for longer than the specified time. If an abortion happens, associated writes will return Status::TimeOut. 2) the stall time of the associated write caused by flush or compaction will be limited by timeout_hint_us. The default value of timeout_hint_us is 0 (i.e., OFF.) The statistics of timeout writes will be recorded in WRITE_TIMEDOUT. Test Plan: export ROCKSDB_TESTS=WriteTimeoutAndDelayTest make db_test ./db_test Reviewers: igor, ljin, haobo, sdong Reviewed By: sdong Subscribers: dhruba, leveldb Differential Revision: https://reviews.facebook.net/D18837
136 lines
3.6 KiB
C++
136 lines
3.6 KiB
C++
// Copyright (c) 2013, Facebook, Inc. All rights reserved.
|
|
// This source code is licensed under the BSD-style license found in the
|
|
// LICENSE file in the root directory of this source tree. An additional grant
|
|
// of patent rights can be found in the PATENTS file in the same 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 <stdio.h>
|
|
#include <assert.h>
|
|
#include <errno.h>
|
|
#include <sys/time.h>
|
|
#include <string.h>
|
|
#include <cstdlib>
|
|
#include "util/logging.h"
|
|
|
|
namespace rocksdb {
|
|
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) {
|
|
#ifdef OS_LINUX
|
|
if (!adaptive) {
|
|
PthreadCall("init mutex", pthread_mutex_init(&mu_, NULL));
|
|
} 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 // ignore adaptive for non-linux platform
|
|
PthreadCall("init mutex", pthread_mutex_init(&mu_, NULL));
|
|
#endif // OS_LINUX
|
|
}
|
|
|
|
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_, NULL));
|
|
}
|
|
|
|
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 = abs_time_us / 1000000;
|
|
ts.tv_nsec = (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_, NULL)); }
|
|
|
|
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_)); }
|
|
|
|
void InitOnce(OnceType* once, void (*initializer)()) {
|
|
PthreadCall("once", pthread_once(once, initializer));
|
|
}
|
|
|
|
} // namespace port
|
|
} // namespace rocksdb
|