/Users/deen/code/yugabyte-db/src/yb/util/striped64.cc
Line | Count | Source (jump to first uncovered line) |
1 | | // Licensed to the Apache Software Foundation (ASF) under one |
2 | | // or more contributor license agreements. See the NOTICE file |
3 | | // distributed with this work for additional information |
4 | | // regarding copyright ownership. The ASF licenses this file |
5 | | // to you under the Apache License, Version 2.0 (the |
6 | | // "License"); you may not use this file except in compliance |
7 | | // with the License. You may obtain a copy of the License at |
8 | | // |
9 | | // http://www.apache.org/licenses/LICENSE-2.0 |
10 | | // |
11 | | // Unless required by applicable law or agreed to in writing, |
12 | | // software distributed under the License is distributed on an |
13 | | // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
14 | | // KIND, either express or implied. See the License for the |
15 | | // specific language governing permissions and limitations |
16 | | // under the License. |
17 | | // |
18 | | // The following only applies to changes made to this file as part of YugaByte development. |
19 | | // |
20 | | // Portions Copyright (c) YugaByte, Inc. |
21 | | // |
22 | | // Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except |
23 | | // in compliance with the License. You may obtain a copy of the License at |
24 | | // |
25 | | // http://www.apache.org/licenses/LICENSE-2.0 |
26 | | // |
27 | | // Unless required by applicable law or agreed to in writing, software distributed under the License |
28 | | // is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
29 | | // or implied. See the License for the specific language governing permissions and limitations |
30 | | // under the License. |
31 | | // |
32 | | |
33 | | #include "yb/util/monotime.h" |
34 | | #include "yb/util/random.h" |
35 | | #include "yb/util/striped64.h" |
36 | | #include "yb/util/threadlocal.h" |
37 | | |
38 | | using yb::striped64::internal::HashCode; |
39 | | using yb::striped64::internal::Cell; |
40 | | |
41 | | namespace yb { |
42 | | |
43 | | namespace striped64 { |
44 | | namespace internal { |
45 | | // |
46 | | // HashCode |
47 | | // |
48 | | |
49 | 1.11M | HashCode::HashCode() { |
50 | 1.11M | std::mt19937_64 random(MonoTime::Now().GetDeltaSince(MonoTime::Min()).ToNanoseconds()); |
51 | 1.11M | const uint64_t hash = random(); |
52 | 1.11M | code_ = (hash == 0) ? 10 : hash; // Avoid zero to allow xorShift rehash |
53 | 1.11M | } |
54 | | |
55 | | // |
56 | | // Cell |
57 | | // |
58 | | |
59 | | Cell::Cell() |
60 | 13.7M | : value_(0) { |
61 | 13.7M | } |
62 | | } // namespace internal |
63 | | } // namespace striped64 |
64 | | |
65 | | // |
66 | | // Striped64 |
67 | | // |
68 | | namespace { |
69 | | |
70 | | const int64_t kNumCpus = sysconf(_SC_NPROCESSORS_ONLN); |
71 | | thread_local std::unique_ptr<HashCode> hashcode_; |
72 | | |
73 | | } |
74 | | |
75 | | Striped64::Striped64() |
76 | | : busy_(false), |
77 | | cell_buffer_(nullptr), |
78 | | cells_(nullptr), |
79 | 11.9M | num_cells_(0) { |
80 | 11.9M | } |
81 | | |
82 | 574k | Striped64::~Striped64() { |
83 | | // Cell is a POD, so no need to destruct each one. |
84 | 574k | free(cell_buffer_); |
85 | 574k | } |
86 | | |
87 | 151k | void Striped64::RetryUpdate(int64_t x, Rehash contention) { |
88 | 151k | uint64_t h = hashcode_->code_; |
89 | | // There are three operations in this loop. |
90 | | // |
91 | | // 1. Try to add to the Cell hash table entry for the thread if the table exists. |
92 | | // When there's contention, rehash to try a different Cell. |
93 | | // 2. Try to initialize the hash table. |
94 | | // 3. Try to update the base counter. |
95 | | // |
96 | | // These are predicated on successful CAS operations, which is why it's all wrapped in an |
97 | | // infinite retry loop. |
98 | 300k | while (true299k ) { |
99 | 300k | int32_t n = base::subtle::Acquire_Load(&num_cells_); |
100 | 300k | if (n > 0) { |
101 | 182k | if (contention == kRehash) { |
102 | | // CAS failed already, rehash before trying to increment. |
103 | 32.7k | contention = kNoRehash; |
104 | 149k | } else { |
105 | 149k | Cell *cell = &(cells_[(n - 1) & h]); |
106 | 149k | int64_t v = cell->value_.Load(); |
107 | 149k | if (cell->CompareAndSet(v, Fn(v, x))) { |
108 | | // Successfully CAS'd the corresponding cell, done. |
109 | 149k | break; |
110 | 149k | } |
111 | 149k | } |
112 | | // Rehash since we failed to CAS, either previously or just now. |
113 | 32.9k | h ^= h << 13; |
114 | 32.9k | h ^= h >> 17; |
115 | 32.9k | h ^= h << 5; |
116 | 119k | } else if (118k n == 0118k && CasBusy()) { |
117 | | // We think table hasn't been initialized yet, try to do so. |
118 | | // Recheck preconditions, someone else might have init'd in the meantime. |
119 | 116k | n = base::subtle::Acquire_Load(&num_cells_); |
120 | 116k | if (n == 0) { |
121 | 116k | n = 1; |
122 | | // Calculate the size. Nearest power of two >= NCPU. |
123 | | // Also handle a negative NCPU, can happen if sysconf name is unknown |
124 | 583k | while (kNumCpus > n) { |
125 | 466k | n <<= 1; |
126 | 466k | } |
127 | | // Allocate cache-aligned memory for use by the cells_ table. |
128 | 116k | int err = posix_memalign(&cell_buffer_, CACHELINE_SIZE, sizeof(Cell)*n); |
129 | 116k | CHECK_EQ(0, err) << "error calling posix_memalign" << std::endl0 ; |
130 | | // Initialize the table |
131 | 116k | cells_ = new (cell_buffer_) Cell[n]; |
132 | 116k | base::subtle::Release_Store(&num_cells_, n); |
133 | 116k | } |
134 | | // End critical section |
135 | 116k | busy_.Store(0); |
136 | 116k | } else { |
137 | | // Fallback to adding to the base value. |
138 | | // Means the table wasn't initialized or we failed to init it. |
139 | 1.65k | int64_t v = base_.value_.Load(); |
140 | 2.88k | if (CasBase(v, Fn(v, x))1.65k ) { |
141 | 2.88k | break; |
142 | 2.88k | } |
143 | 1.65k | } |
144 | 300k | } |
145 | | // Record index for next time |
146 | 151k | hashcode_->code_ = h; |
147 | 151k | } |
148 | | |
149 | 1 | void Striped64::InternalReset(int64_t initialValue) { |
150 | 1 | const int32_t n = base::subtle::Acquire_Load(&num_cells_); |
151 | 1 | base_.value_.Store(initialValue); |
152 | 1 | for (int i = 0; i < n; i++0 ) { |
153 | 0 | cells_[i].value_.Store(initialValue); |
154 | 0 | } |
155 | 1 | } |
156 | | |
157 | 998M | void LongAdder::IncrementBy(int64_t x) { |
158 | 998M | if (!hashcode_) { |
159 | 1.11M | hashcode_ = std::make_unique<HashCode>(); |
160 | 1.11M | } |
161 | | // Use hash table if present. If that fails, call RetryUpdate to rehash and retry. |
162 | | // If no hash table, try to CAS the base counter. If that fails, RetryUpdate to init the table. |
163 | 998M | const int32_t n = base::subtle::Acquire_Load(&num_cells_); |
164 | 998M | if (n > 0) { |
165 | 710M | Cell *cell = &(cells_[(n - 1) & hashcode_->code_]); |
166 | 710M | DCHECK_EQ(0, reinterpret_cast<const uintptr_t>(cell) & (sizeof(Cell) - 1)) |
167 | 0 | << " unaligned Cell not allowed for Striped64" << std::endl; |
168 | 710M | const int64_t old = cell->value_.Load(); |
169 | 710M | if (!cell->CompareAndSet(old, old + x)) { |
170 | | // When we hit a hash table contention, signal RetryUpdate to rehash. |
171 | 32.7k | RetryUpdate(x, kRehash); |
172 | 32.7k | } |
173 | 710M | } else { |
174 | 288M | int64_t b = base_.value_.Load(); |
175 | 288M | if (!base_.CompareAndSet(b, b + x)) { |
176 | | // Attempt to initialize the table. No need to rehash since the contention was for the |
177 | | // base counter, not the hash table. |
178 | 118k | RetryUpdate(x, kNoRehash); |
179 | 118k | } |
180 | 288M | } |
181 | 998M | } |
182 | | |
183 | | // |
184 | | // LongAdder |
185 | | // |
186 | | |
187 | 16.0M | int64_t LongAdder::Value() const { |
188 | 16.0M | int64_t sum = base_.value_.Load(); |
189 | 16.0M | const int32_t n = base::subtle::Acquire_Load(&num_cells_); |
190 | 18.6M | for (int i = 0; i < n; i++2.58M ) { |
191 | 2.58M | sum += cells_[i].value_.Load(); |
192 | 2.58M | } |
193 | 16.0M | return sum; |
194 | 16.0M | } |
195 | | |
196 | | } // namespace yb |