View Javadoc

1   /**
2    * Licensed to the Apache Software Foundation (ASF) under one
3    * or more contributor license agreements.  See the NOTICE file
4    * distributed with this work for additional information
5    * regarding copyright ownership.  The ASF licenses this file
6    * to you under the Apache License, Version 2.0 (the
7    * "License"); you may not use this file except in compliance
8    * with the License.  You may obtain a copy of the License at
9    *
10   *     http://www.apache.org/licenses/LICENSE-2.0
11   *
12   * Unless required by applicable law or agreed to in writing, software
13   * distributed under the License is distributed on an "AS IS" BASIS,
14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15   * See the License for the specific language governing permissions and
16   * limitations under the License.
17   */
18  package org.apache.hadoop.hbase.regionserver.wal;
19  
20  import java.util.concurrent.ExecutionException;
21  import java.util.concurrent.TimeUnit;
22  
23  import org.apache.hadoop.hbase.classification.InterfaceAudience;
24  import org.htrace.Span;
25  
26  /**
27   * A Future on a filesystem sync call.  It given to a client or 'Handler' for it to wait on till
28   * the sync completes.
29   *
30   * <p>Handlers coming in call append, append, append, and then do a flush/sync of
31   * the edits they have appended the WAL before returning. Since sync takes a while to
32   * complete, we give the Handlers back this sync future to wait on until the
33   * actual HDFS sync completes. Meantime this sync future goes across the ringbuffer and into a
34   * sync runner thread; when it completes, it finishes up the future, the handler get or failed
35   * check completes and the Handler can then progress.
36   * <p>
37   * This is just a partial implementation of Future; we just implement get and
38   * failure.  Unimplemented methods throw {@link UnsupportedOperationException}.
39   * <p>
40   * There is not a one-to-one correlation between dfs sync invocations and
41   * instances of this class. A single dfs sync call may complete and mark many
42   * SyncFutures as done; i.e. we batch up sync calls rather than do a dfs sync
43   * call every time a Handler asks for it.
44   * <p>
45   * SyncFutures are immutable but recycled. Call {@link #reset(long, Span)} before use even
46   * if it the first time, start the sync, then park the 'hitched' thread on a call to
47   * {@link #get()}
48   */
49  @InterfaceAudience.Private
50  class SyncFuture {
51    // Implementation notes: I tried using a cyclicbarrier in here for handler and sync threads
52    // to coordinate on but it did not give any obvious advantage and some issues with order in which
53    // events happen.
54    private static final long NOT_DONE = 0;
55  
56    /**
57     * The sequence at which we were added to the ring buffer.
58     */
59    private long ringBufferSequence;
60  
61    /**
62     * The sequence that was set in here when we were marked done. Should be equal
63     * or > ringBufferSequence.  Put this data member into the NOT_DONE state while this
64     * class is in use.  But for the first position on construction, let it be -1 so we can
65     * immediately call {@link #reset(long, Span)} below and it will work.
66     */
67    private long doneSequence = -1;
68  
69    /**
70     * If error, the associated throwable. Set when the future is 'done'.
71     */
72    private Throwable throwable = null;
73  
74    private Thread t;
75  
76    /**
77     * Optionally carry a disconnected scope to the SyncRunner.
78     */
79    private Span span;
80  
81    /**
82     * Call this method to clear old usage and get it ready for new deploy. Call
83     * this method even if it is being used for the first time.
84     *
85     * @param sequence sequenceId from this Future's position in the RingBuffer
86     * @return this
87     */
88    synchronized SyncFuture reset(final long sequence) {
89      return reset(sequence, null);
90    }
91  
92    /**
93     * Call this method to clear old usage and get it ready for new deploy. Call
94     * this method even if it is being used for the first time.
95     *
96     * @param sequence sequenceId from this Future's position in the RingBuffer
97     * @param span curren span, detached from caller. Don't forget to attach it when
98     *             resuming after a call to {@link #get()}.
99     * @return this
100    */
101   synchronized SyncFuture reset(final long sequence, Span span) {
102     if (t != null && t != Thread.currentThread()) throw new IllegalStateException();
103     t = Thread.currentThread();
104     if (!isDone()) throw new IllegalStateException("" + sequence + " " + Thread.currentThread());
105     this.doneSequence = NOT_DONE;
106     this.ringBufferSequence = sequence;
107     this.span = span;
108     return this;
109   }
110 
111   @Override
112   public synchronized String toString() {
113     return "done=" + isDone() + ", ringBufferSequence=" + this.ringBufferSequence;
114   }
115 
116   synchronized long getRingBufferSequence() {
117     return this.ringBufferSequence;
118   }
119 
120   /**
121    * Retrieve the {@code span} instance from this Future. EventHandler calls
122    * this method to continue the span. Thread waiting on this Future musn't call
123    * this method until AFTER calling {@link #get()} and the future has been
124    * released back to the originating thread.
125    */
126   synchronized Span getSpan() {
127     return this.span;
128   }
129 
130   /**
131    * Used to re-attach a {@code span} to the Future. Called by the EventHandler
132    * after a it has completed processing and detached the span from its scope.
133    */
134   synchronized void setSpan(Span span) {
135     this.span = span;
136   }
137 
138   /**
139    * @param sequence Sync sequence at which this future 'completed'.
140    * @param t Can be null.  Set if we are 'completing' on error (and this 't' is the error).
141    * @return True if we successfully marked this outstanding future as completed/done.
142    * Returns false if this future is already 'done' when this method called.
143    */
144   synchronized boolean done(final long sequence, final Throwable t) {
145     if (isDone()) return false;
146     this.throwable = t;
147     if (sequence < this.ringBufferSequence) {
148       // Something badly wrong.
149       if (throwable == null) {
150         this.throwable = new IllegalStateException("sequence=" + sequence +
151           ", ringBufferSequence=" + this.ringBufferSequence);
152       }
153     }
154     // Mark done.
155     this.doneSequence = sequence;
156     // Wake up waiting threads.
157     notify();
158     return true;
159   }
160 
161   public boolean cancel(boolean mayInterruptIfRunning) {
162     throw new UnsupportedOperationException();
163   }
164 
165   public synchronized long get() throws InterruptedException, ExecutionException {
166     while (!isDone()) {
167       wait(1000);
168     }
169     if (this.throwable != null) throw new ExecutionException(this.throwable);
170     return this.doneSequence;
171   }
172 
173   public Long get(long timeout, TimeUnit unit)
174   throws InterruptedException, ExecutionException {
175     throw new UnsupportedOperationException();
176   }
177 
178   public boolean isCancelled() {
179     throw new UnsupportedOperationException();
180   }
181 
182   synchronized boolean isDone() {
183     return this.doneSequence != NOT_DONE;
184   }
185 
186   synchronized boolean isThrowable() {
187     return isDone() && getThrowable() != null;
188   }
189 
190   synchronized Throwable getThrowable() {
191     return this.throwable;
192   }
193 }