Skip to content

Commit 944ff67

Browse files
committed
Merge branch 'master' into columnfamilies
2 parents 8895526 + b2795b7 commit 944ff67

File tree

8 files changed

+878
-6
lines changed

8 files changed

+878
-6
lines changed

HISTORY.md

+1
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
* Removed arena.h from public header files.
1414
* By default, checksums are verified on every read from database
1515
* Added is_manual_compaction to CompactionFilter::Context
16+
* Added "virtual void WaitForJoin() = 0" in class Env
1617

1718
## 2.7.0 (01/28/2014)
1819

Makefile

+6-2
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,8 @@ TESTS = \
8888
version_set_test \
8989
write_batch_test\
9090
deletefile_test \
91-
table_test
91+
table_test \
92+
thread_local_test
9293

9394
TOOLS = \
9495
sst_dump \
@@ -148,7 +149,7 @@ all: $(LIBRARY) $(PROGRAMS)
148149

149150
dbg: $(LIBRARY) $(PROGRAMS)
150151

151-
# Will also generate shared libraries.
152+
# Will also generate shared libraries.
152153
release:
153154
$(MAKE) clean
154155
OPT="-DNDEBUG -O2" $(MAKE) all -j32
@@ -280,6 +281,9 @@ redis_test: utilities/redis/redis_lists_test.o $(LIBOBJECTS) $(TESTHARNESS)
280281
histogram_test: util/histogram_test.o $(LIBOBJECTS) $(TESTHARNESS)
281282
$(CXX) util/histogram_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o$@ $(LDFLAGS) $(COVERAGEFLAGS)
282283

284+
thread_local_test: util/thread_local_test.o $(LIBOBJECTS) $(TESTHARNESS)
285+
$(CXX) util/thread_local_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS) $(COVERAGEFLAGS)
286+
283287
corruption_test: db/corruption_test.o $(LIBOBJECTS) $(TESTHARNESS)
284288
$(CXX) db/corruption_test.o $(LIBOBJECTS) $(TESTHARNESS) $(EXEC_LDFLAGS) -o $@ $(LDFLAGS) $(COVERAGEFLAGS)
285289

hdfs/env_hdfs.h

+8-4
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ class HdfsFatalException : public std::exception {
4747
class HdfsEnv : public Env {
4848

4949
public:
50-
HdfsEnv(const std::string& fsname) : fsname_(fsname) {
50+
explicit HdfsEnv(const std::string& fsname) : fsname_(fsname) {
5151
posixEnv = Env::Default();
5252
fileSys_ = connectToPath(fsname_);
5353
}
@@ -108,6 +108,8 @@ class HdfsEnv : public Env {
108108
posixEnv->StartThread(function, arg);
109109
}
110110

111+
virtual void WaitForJoin() { posixEnv->WaitForJoin(); }
112+
111113
virtual Status GetTestDirectory(std::string* path) {
112114
return posixEnv->GetTestDirectory(path);
113115
}
@@ -161,7 +163,7 @@ class HdfsEnv : public Env {
161163
*/
162164
hdfsFS connectToPath(const std::string& uri) {
163165
if (uri.empty()) {
164-
return NULL;
166+
return nullptr;
165167
}
166168
if (uri.find(kProto) != 0) {
167169
// uri doesn't start with hdfs:// -> use default:0, which is special
@@ -218,10 +220,10 @@ static const Status notsup;
218220
class HdfsEnv : public Env {
219221

220222
public:
221-
HdfsEnv(const std::string& fsname) {
223+
explicit HdfsEnv(const std::string& fsname) {
222224
fprintf(stderr, "You have not build rocksdb with HDFS support\n");
223225
fprintf(stderr, "Please see hdfs/README for details\n");
224-
throw new std::exception();
226+
throw std::exception();
225227
}
226228

227229
virtual ~HdfsEnv() {
@@ -288,6 +290,8 @@ class HdfsEnv : public Env {
288290

289291
virtual void StartThread(void (*function)(void* arg), void* arg) {}
290292

293+
virtual void WaitForJoin() {}
294+
291295
virtual Status GetTestDirectory(std::string* path) {return notsup;}
292296

293297
virtual uint64_t NowMicros() {return 0;}

include/rocksdb/env.h

+4
Original file line numberDiff line numberDiff line change
@@ -205,6 +205,9 @@ class Env {
205205
// When "function(arg)" returns, the thread will be destroyed.
206206
virtual void StartThread(void (*function)(void* arg), void* arg) = 0;
207207

208+
// Wait for all threads started by StartThread to terminate.
209+
virtual void WaitForJoin() = 0;
210+
208211
// *path is set to a temporary directory that can be used for testing. It may
209212
// or many not have just been created. The directory may or may not differ
210213
// between runs of the same process, but subsequent calls will return the
@@ -634,6 +637,7 @@ class EnvWrapper : public Env {
634637
void StartThread(void (*f)(void*), void* a) {
635638
return target_->StartThread(f, a);
636639
}
640+
void WaitForJoin() { return target_->WaitForJoin(); }
637641
virtual Status GetTestDirectory(std::string* path) {
638642
return target_->GetTestDirectory(path);
639643
}

util/env_posix.cc

+9
Original file line numberDiff line numberDiff line change
@@ -1194,6 +1194,8 @@ class PosixEnv : public Env {
11941194

11951195
virtual void StartThread(void (*function)(void* arg), void* arg);
11961196

1197+
virtual void WaitForJoin();
1198+
11971199
virtual Status GetTestDirectory(std::string* result) {
11981200
const char* env = getenv("TEST_TMPDIR");
11991201
if (env && env[0] != '\0') {
@@ -1511,6 +1513,13 @@ void PosixEnv::StartThread(void (*function)(void* arg), void* arg) {
15111513
PthreadCall("unlock", pthread_mutex_unlock(&mu_));
15121514
}
15131515

1516+
void PosixEnv::WaitForJoin() {
1517+
for (const auto tid : threads_to_join_) {
1518+
pthread_join(tid, nullptr);
1519+
}
1520+
threads_to_join_.clear();
1521+
}
1522+
15141523
} // namespace
15151524

15161525
std::string Env::GenerateUniqueId() {

util/thread_local.cc

+236
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,236 @@
1+
// Copyright (c) 2013, Facebook, Inc. All rights reserved.
2+
// This source code is licensed under the BSD-style license found in the
3+
// LICENSE file in the root directory of this source tree. An additional grant
4+
// of patent rights can be found in the PATENTS file in the same directory.
5+
//
6+
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
7+
// Use of this source code is governed by a BSD-style license that can be
8+
// found in the LICENSE file. See the AUTHORS file for names of contributors.
9+
10+
#include "util/thread_local.h"
11+
#include "util/mutexlock.h"
12+
13+
#if defined(__GNUC__) && __GNUC__ >= 4
14+
#define UNLIKELY(x) (__builtin_expect((x), 0))
15+
#else
16+
#define UNLIKELY(x) (x)
17+
#endif
18+
19+
namespace rocksdb {
20+
21+
std::unique_ptr<ThreadLocalPtr::StaticMeta> ThreadLocalPtr::StaticMeta::inst_;
22+
port::Mutex ThreadLocalPtr::StaticMeta::mutex_;
23+
#if !defined(OS_MACOSX)
24+
__thread ThreadLocalPtr::ThreadData* ThreadLocalPtr::StaticMeta::tls_ = nullptr;
25+
#endif
26+
27+
ThreadLocalPtr::StaticMeta* ThreadLocalPtr::StaticMeta::Instance() {
28+
if (UNLIKELY(inst_ == nullptr)) {
29+
MutexLock l(&mutex_);
30+
if (inst_ == nullptr) {
31+
inst_.reset(new StaticMeta());
32+
}
33+
}
34+
return inst_.get();
35+
}
36+
37+
void ThreadLocalPtr::StaticMeta::OnThreadExit(void* ptr) {
38+
auto* tls = static_cast<ThreadData*>(ptr);
39+
assert(tls != nullptr);
40+
41+
auto* inst = Instance();
42+
pthread_setspecific(inst->pthread_key_, nullptr);
43+
44+
MutexLock l(&mutex_);
45+
inst->RemoveThreadData(tls);
46+
// Unref stored pointers of current thread from all instances
47+
uint32_t id = 0;
48+
for (auto& e : tls->entries) {
49+
void* raw = e.ptr.load(std::memory_order_relaxed);
50+
if (raw != nullptr) {
51+
auto unref = inst->GetHandler(id);
52+
if (unref != nullptr) {
53+
unref(raw);
54+
}
55+
}
56+
++id;
57+
}
58+
// Delete thread local structure no matter if it is Mac platform
59+
delete tls;
60+
}
61+
62+
ThreadLocalPtr::StaticMeta::StaticMeta() : next_instance_id_(0) {
63+
if (pthread_key_create(&pthread_key_, &OnThreadExit) != 0) {
64+
throw std::runtime_error("pthread_key_create failed");
65+
}
66+
head_.next = &head_;
67+
head_.prev = &head_;
68+
}
69+
70+
void ThreadLocalPtr::StaticMeta::AddThreadData(ThreadLocalPtr::ThreadData* d) {
71+
mutex_.AssertHeld();
72+
d->next = &head_;
73+
d->prev = head_.prev;
74+
head_.prev->next = d;
75+
head_.prev = d;
76+
}
77+
78+
void ThreadLocalPtr::StaticMeta::RemoveThreadData(
79+
ThreadLocalPtr::ThreadData* d) {
80+
mutex_.AssertHeld();
81+
d->next->prev = d->prev;
82+
d->prev->next = d->next;
83+
d->next = d->prev = d;
84+
}
85+
86+
ThreadLocalPtr::ThreadData* ThreadLocalPtr::StaticMeta::GetThreadLocal() {
87+
#if defined(OS_MACOSX)
88+
// Make this local variable name look like a member variable so that we
89+
// can share all the code below
90+
ThreadData* tls_ =
91+
static_cast<ThreadData*>(pthread_getspecific(Instance()->pthread_key_));
92+
#endif
93+
94+
if (UNLIKELY(tls_ == nullptr)) {
95+
auto* inst = Instance();
96+
tls_ = new ThreadData();
97+
{
98+
// Register it in the global chain, needs to be done before thread exit
99+
// handler registration
100+
MutexLock l(&mutex_);
101+
inst->AddThreadData(tls_);
102+
}
103+
// Even it is not OS_MACOSX, need to register value for pthread_key_ so that
104+
// its exit handler will be triggered.
105+
if (pthread_setspecific(inst->pthread_key_, tls_) != 0) {
106+
{
107+
MutexLock l(&mutex_);
108+
inst->RemoveThreadData(tls_);
109+
}
110+
delete tls_;
111+
throw std::runtime_error("pthread_setspecific failed");
112+
}
113+
}
114+
return tls_;
115+
}
116+
117+
void* ThreadLocalPtr::StaticMeta::Get(uint32_t id) const {
118+
auto* tls = GetThreadLocal();
119+
if (UNLIKELY(id >= tls->entries.size())) {
120+
return nullptr;
121+
}
122+
return tls->entries[id].ptr.load(std::memory_order_relaxed);
123+
}
124+
125+
void ThreadLocalPtr::StaticMeta::Reset(uint32_t id, void* ptr) {
126+
auto* tls = GetThreadLocal();
127+
if (UNLIKELY(id >= tls->entries.size())) {
128+
// Need mutex to protect entries access within ReclaimId
129+
MutexLock l(&mutex_);
130+
tls->entries.resize(id + 1);
131+
}
132+
tls->entries[id].ptr.store(ptr, std::memory_order_relaxed);
133+
}
134+
135+
void* ThreadLocalPtr::StaticMeta::Swap(uint32_t id, void* ptr) {
136+
auto* tls = GetThreadLocal();
137+
if (UNLIKELY(id >= tls->entries.size())) {
138+
// Need mutex to protect entries access within ReclaimId
139+
MutexLock l(&mutex_);
140+
tls->entries.resize(id + 1);
141+
}
142+
return tls->entries[id].ptr.exchange(ptr, std::memory_order_relaxed);
143+
}
144+
145+
void ThreadLocalPtr::StaticMeta::Scrape(uint32_t id, autovector<void*>* ptrs) {
146+
MutexLock l(&mutex_);
147+
for (ThreadData* t = head_.next; t != &head_; t = t->next) {
148+
if (id < t->entries.size()) {
149+
void* ptr =
150+
t->entries[id].ptr.exchange(nullptr, std::memory_order_relaxed);
151+
if (ptr != nullptr) {
152+
ptrs->push_back(ptr);
153+
}
154+
}
155+
}
156+
}
157+
158+
void ThreadLocalPtr::StaticMeta::SetHandler(uint32_t id, UnrefHandler handler) {
159+
MutexLock l(&mutex_);
160+
handler_map_[id] = handler;
161+
}
162+
163+
UnrefHandler ThreadLocalPtr::StaticMeta::GetHandler(uint32_t id) {
164+
mutex_.AssertHeld();
165+
auto iter = handler_map_.find(id);
166+
if (iter == handler_map_.end()) {
167+
return nullptr;
168+
}
169+
return iter->second;
170+
}
171+
172+
uint32_t ThreadLocalPtr::StaticMeta::GetId() {
173+
MutexLock l(&mutex_);
174+
if (free_instance_ids_.empty()) {
175+
return next_instance_id_++;
176+
}
177+
178+
uint32_t id = free_instance_ids_.back();
179+
free_instance_ids_.pop_back();
180+
return id;
181+
}
182+
183+
uint32_t ThreadLocalPtr::StaticMeta::PeekId() const {
184+
MutexLock l(&mutex_);
185+
if (!free_instance_ids_.empty()) {
186+
return free_instance_ids_.back();
187+
}
188+
return next_instance_id_;
189+
}
190+
191+
void ThreadLocalPtr::StaticMeta::ReclaimId(uint32_t id) {
192+
// This id is not used, go through all thread local data and release
193+
// corresponding value
194+
MutexLock l(&mutex_);
195+
auto unref = GetHandler(id);
196+
for (ThreadData* t = head_.next; t != &head_; t = t->next) {
197+
if (id < t->entries.size()) {
198+
void* ptr =
199+
t->entries[id].ptr.exchange(nullptr, std::memory_order_relaxed);
200+
if (ptr != nullptr && unref != nullptr) {
201+
unref(ptr);
202+
}
203+
}
204+
}
205+
handler_map_[id] = nullptr;
206+
free_instance_ids_.push_back(id);
207+
}
208+
209+
ThreadLocalPtr::ThreadLocalPtr(UnrefHandler handler)
210+
: id_(StaticMeta::Instance()->GetId()) {
211+
if (handler != nullptr) {
212+
StaticMeta::Instance()->SetHandler(id_, handler);
213+
}
214+
}
215+
216+
ThreadLocalPtr::~ThreadLocalPtr() {
217+
StaticMeta::Instance()->ReclaimId(id_);
218+
}
219+
220+
void* ThreadLocalPtr::Get() const {
221+
return StaticMeta::Instance()->Get(id_);
222+
}
223+
224+
void ThreadLocalPtr::Reset(void* ptr) {
225+
StaticMeta::Instance()->Reset(id_, ptr);
226+
}
227+
228+
void* ThreadLocalPtr::Swap(void* ptr) {
229+
return StaticMeta::Instance()->Swap(id_, ptr);
230+
}
231+
232+
void ThreadLocalPtr::Scrape(autovector<void*>* ptrs) {
233+
StaticMeta::Instance()->Scrape(id_, ptrs);
234+
}
235+
236+
} // namespace rocksdb

0 commit comments

Comments
 (0)