View Javadoc

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 }