1 /**
2 *
3 * Licensed to the Apache Software Foundation (ASF) under one
4 * or more contributor license agreements. See the NOTICE file
5 * distributed with this work for additional information
6 * regarding copyright ownership. The ASF licenses this file
7 * to you under the Apache License, Version 2.0 (the
8 * "License"); you may not use this file except in compliance
9 * with the License. You may obtain a copy of the License at
10 *
11 * http://www.apache.org/licenses/LICENSE-2.0
12 *
13 * Unless required by applicable law or agreed to in writing, software
14 * distributed under the License is distributed on an "AS IS" BASIS,
15 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 * See the License for the specific language governing permissions and
17 * limitations under the License.
18 */
19 package org.apache.hadoop.hbase.regionserver;
20
21 import java.io.IOException;
22 import java.util.LinkedList;
23 import java.util.concurrent.atomic.AtomicLong;
24
25 import org.apache.hadoop.hbase.classification.InterfaceAudience;
26 import org.apache.hadoop.hbase.util.Bytes;
27 import org.apache.hadoop.hbase.util.ClassSize;
28
29 /**
30 * Manages the read/write consistency within memstore. This provides
31 * an interface for readers to determine what entries to ignore, and
32 * a mechanism for writers to obtain new write numbers, then "commit"
33 * the new writes for readers to read (thus forming atomic transactions).
34 */
35 @InterfaceAudience.Private
36 public class MultiVersionConsistencyControl {
37 static final long NO_WRITE_NUMBER = 0;
38 private volatile long memstoreRead = 0;
39 private final Object readWaiters = new Object();
40
41 // This is the pending queue of writes.
42 private final LinkedList<WriteEntry> writeQueue =
43 new LinkedList<WriteEntry>();
44
45 /**
46 * Default constructor. Initializes the memstoreRead/Write points to 0.
47 */
48 public MultiVersionConsistencyControl() {
49 }
50
51 /**
52 * Initializes the memstoreRead/Write points appropriately.
53 * @param startPoint
54 */
55 public void initialize(long startPoint) {
56 synchronized (writeQueue) {
57 writeQueue.clear();
58 memstoreRead = startPoint;
59 }
60 }
61
62 /**
63 *
64 * @param initVal The value we used initially and expected it'll be reset later
65 * @return WriteEntry instance.
66 */
67 WriteEntry beginMemstoreInsert() {
68 return beginMemstoreInsertWithSeqNum(NO_WRITE_NUMBER);
69 }
70
71 /**
72 * Get a mvcc write number before an actual one(its log sequence Id) being assigned
73 * @param sequenceId
74 * @return long a faked write number which is bigger enough not to be seen by others before a real
75 * one is assigned
76 */
77 public static long getPreAssignedWriteNumber(AtomicLong sequenceId) {
78 // the 1 billion is just an arbitrary big number to guard no scanner will reach it before
79 // current MVCC completes. Theoretically the bump only needs to be 2 * the number of handlers
80 // because each handler could increment sequence num twice and max concurrent in-flight
81 // transactions is the number of RPC handlers.
82 // We can't use Long.MAX_VALUE because we still want to maintain the ordering when multiple
83 // changes touch same row key.
84 // If for any reason, the bumped value isn't reset due to failure situations, we'll reset
85 // curSeqNum to NO_WRITE_NUMBER in order NOT to advance memstore read point at all.
86 // St.Ack 20150901 Where is the reset to NO_WRITE_NUMBER done?
87 return sequenceId.incrementAndGet() + 1000000000;
88 }
89
90 /**
91 * This function starts a MVCC transaction with current region's log change sequence number. Since
92 * we set change sequence number when flushing current change to WAL(late binding), the flush
93 * order may differ from the order to start a MVCC transaction. For example, a change begins a
94 * MVCC firstly may complete later than a change which starts MVCC at a later time. Therefore, we
95 * add a safe bumper to the passed in sequence number to start a MVCC so that no other concurrent
96 * transactions will reuse the number till current MVCC completes(success or fail). The "faked"
97 * big number is safe because we only need it to prevent current change being seen and the number
98 * will be reset to real sequence number(set in log sync) right before we complete a MVCC in order
99 * for MVCC to align with flush sequence.
100 * @param curSeqNum
101 * @return WriteEntry a WriteEntry instance with the passed in curSeqNum
102 */
103 public WriteEntry beginMemstoreInsertWithSeqNum(long curSeqNum) {
104 WriteEntry e = new WriteEntry(curSeqNum);
105 synchronized (writeQueue) {
106 writeQueue.add(e);
107 return e;
108 }
109 }
110
111 /**
112 * Complete a {@link WriteEntry} that was created by
113 * {@link #beginMemstoreInsertWithSeqNum(long)}. At the end of this call, the global read
114 * point is at least as large as the write point of the passed in WriteEntry. Thus, the write is
115 * visible to MVCC readers.
116 * @throws IOException
117 */
118 public void completeMemstoreInsertWithSeqNum(WriteEntry e, SequenceId seqId)
119 throws IOException {
120 if(e == null) return;
121 if (seqId != null) {
122 e.setWriteNumber(seqId.getSequenceId());
123 } else {
124 // set the value to NO_WRITE_NUMBER in order NOT to advance memstore readpoint inside
125 // function beginMemstoreInsertWithSeqNum in case of failures
126 e.setWriteNumber(NO_WRITE_NUMBER);
127 }
128 waitForPreviousTransactionsComplete(e);
129 }
130
131 /**
132 * Cancel a write insert that failed.
133 * Removes the write entry without advancing read point or without interfering with write
134 * entries queued behind us. It is like #advanceMemstore(WriteEntry) only this method
135 * will move the read point to the sequence id that is in WriteEntry even if it ridiculous (see
136 * the trick in HRegion where we call {@link #getPreAssignedWriteNumber(AtomicLong)} just to mark
137 * it as for special handling).
138 * @param writeEntry Failed attempt at write. Does cleanup.
139 */
140 public void cancelMemstoreInsert(WriteEntry writeEntry) {
141 // I'm not clear on how this voodoo all works but setting write number to -1 does NOT advance
142 // readpoint and gets my little writeEntry completed and removed from queue of outstanding
143 // events which seems right. St.Ack 20150901.
144 writeEntry.setWriteNumber(NO_WRITE_NUMBER);
145 advanceMemstore(writeEntry);
146 }
147
148 /**
149 * Complete a {@link WriteEntry} that was created by {@link #beginMemstoreInsert()}. At the
150 * end of this call, the global read point is at least as large as the write point of the passed
151 * in WriteEntry. Thus, the write is visible to MVCC readers.
152 */
153 public void completeMemstoreInsert(WriteEntry e) {
154 waitForPreviousTransactionsComplete(e);
155 }
156
157 /**
158 * Mark the {@link WriteEntry} as complete and advance the read point as
159 * much as possible.
160 *
161 * How much is the read point advanced?
162 * Let S be the set of all write numbers that are completed and where all previous write numbers
163 * are also completed. Then, the read point is advanced to the supremum of S.
164 *
165 * @param e
166 * @return true if e is visible to MVCC readers (that is, readpoint >= e.writeNumber)
167 */
168 boolean advanceMemstore(WriteEntry e) {
169 long nextReadValue = -1;
170 synchronized (writeQueue) {
171 e.markCompleted();
172
173 while (!writeQueue.isEmpty()) {
174 WriteEntry queueFirst = writeQueue.getFirst();
175 if (queueFirst.isCompleted()) {
176 // Using Max because Edit complete in WAL sync order not arriving order
177 nextReadValue = Math.max(nextReadValue, queueFirst.getWriteNumber());
178 writeQueue.removeFirst();
179 } else {
180 break;
181 }
182 }
183
184 if (nextReadValue > memstoreRead) {
185 memstoreRead = nextReadValue;
186 }
187
188 // notify waiters on writeQueue before return
189 writeQueue.notifyAll();
190 }
191
192 if (nextReadValue > 0) {
193 synchronized (readWaiters) {
194 readWaiters.notifyAll();
195 }
196 }
197
198 if (memstoreRead >= e.getWriteNumber()) {
199 return true;
200 }
201 return false;
202 }
203
204 /**
205 * Advances the current read point to be given seqNum if it is smaller than
206 * that.
207 */
208 void advanceMemstoreReadPointIfNeeded(long seqNum) {
209 synchronized (writeQueue) {
210 if (this.memstoreRead < seqNum) {
211 memstoreRead = seqNum;
212 }
213 }
214 }
215
216 /**
217 * Wait for all previous MVCC transactions complete
218 */
219 public void waitForPreviousTransactionsComplete() {
220 WriteEntry w = beginMemstoreInsert();
221 waitForPreviousTransactionsComplete(w);
222 }
223
224 public void waitForPreviousTransactionsComplete(WriteEntry waitedEntry) {
225 boolean interrupted = false;
226 WriteEntry w = waitedEntry;
227
228 try {
229 WriteEntry firstEntry = null;
230 do {
231 synchronized (writeQueue) {
232 // writeQueue won't be empty at this point, the following is just a safety check
233 if (writeQueue.isEmpty()) {
234 break;
235 }
236 firstEntry = writeQueue.getFirst();
237 if (firstEntry == w) {
238 // all previous in-flight transactions are done
239 break;
240 }
241 try {
242 writeQueue.wait(0);
243 } catch (InterruptedException ie) {
244 // We were interrupted... finish the loop -- i.e. cleanup --and then
245 // on our way out, reset the interrupt flag.
246 interrupted = true;
247 break;
248 }
249 }
250 } while (firstEntry != null);
251 } finally {
252 if (w != null) {
253 advanceMemstore(w);
254 }
255 }
256 if (interrupted) {
257 Thread.currentThread().interrupt();
258 }
259 }
260
261 public long memstoreReadPoint() {
262 return memstoreRead;
263 }
264
265 public static class WriteEntry {
266 private long writeNumber;
267 private volatile boolean completed = false;
268
269 WriteEntry(long writeNumber) {
270 this.writeNumber = writeNumber;
271 }
272 void markCompleted() {
273 this.completed = true;
274 }
275 boolean isCompleted() {
276 return this.completed;
277 }
278 long getWriteNumber() {
279 return this.writeNumber;
280 }
281 void setWriteNumber(long val){
282 this.writeNumber = val;
283 }
284 }
285
286 public static final long FIXED_SIZE = ClassSize.align(
287 ClassSize.OBJECT +
288 2 * Bytes.SIZEOF_LONG +
289 2 * ClassSize.REFERENCE);
290
291 }