001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.regionserver.wal;
019
020import java.util.concurrent.ExecutionException;
021import java.util.concurrent.TimeUnit;
022
023import org.apache.hadoop.hbase.exceptions.TimeoutIOException;
024import org.apache.yetus.audience.InterfaceAudience;
025
026/**
027 * A Future on a filesystem sync call. It given to a client or 'Handler' for it to wait on till the
028 * sync completes.
029 * <p>
030 * Handlers coming in call append, append, append, and then do a flush/sync of the edits they have
031 * appended the WAL before returning. Since sync takes a while to complete, we give the Handlers
032 * back this sync future to wait on until the actual HDFS sync completes. Meantime this sync future
033 * goes across a queue and is handled by a background thread; when it completes, it finishes up the
034 * future, the handler get or failed check completes and the Handler can then progress.
035 * <p>
036 * This is just a partial implementation of Future; we just implement get and failure.
037 * <p>
038 * There is not a one-to-one correlation between dfs sync invocations and instances of this class. A
039 * single dfs sync call may complete and mark many SyncFutures as done; i.e. we batch up sync calls
040 * rather than do a dfs sync call every time a Handler asks for it.
041 * <p>
042 * SyncFutures are immutable but recycled. Call #reset(long, Span) before use even if it the first
043 * time, start the sync, then park the 'hitched' thread on a call to #get().
044 */
045@InterfaceAudience.Private
046class SyncFuture {
047  // Implementation notes: I tried using a cyclicbarrier in here for handler and sync threads
048  // to coordinate on but it did not give any obvious advantage and some issues with order in which
049  // events happen.
050  private static final long NOT_DONE = -1L;
051
052  /**
053   * The transaction id of this operation, monotonically increases.
054   */
055  private long txid;
056
057  /**
058   * The transaction id that was set in here when we were marked done. Should be equal or > txnId.
059   * Put this data member into the NOT_DONE state while this class is in use.
060   */
061  private long doneTxid;
062
063  /**
064   * If error, the associated throwable. Set when the future is 'done'.
065   */
066  private Throwable throwable;
067
068  private Thread t;
069
070  private boolean forceSync;
071
072  /**
073   * Call this method to clear old usage and get it ready for new deploy.
074   * @param txid the new transaction id
075   * @return this
076   */
077  synchronized SyncFuture reset(long txid) {
078    if (t != null && t != Thread.currentThread()) {
079      throw new IllegalStateException();
080    }
081    t = Thread.currentThread();
082    if (!isDone()) {
083      throw new IllegalStateException("" + txid + " " + Thread.currentThread());
084    }
085    this.doneTxid = NOT_DONE;
086    this.txid = txid;
087    this.throwable = null;
088    return this;
089  }
090
091  @Override
092  public synchronized String toString() {
093    return "done=" + isDone() + ", txid=" + this.txid;
094  }
095
096  synchronized long getTxid() {
097    return this.txid;
098  }
099
100  synchronized boolean isForceSync() {
101    return forceSync;
102  }
103
104  synchronized SyncFuture setForceSync(boolean forceSync) {
105    this.forceSync = forceSync;
106    return this;
107  }
108
109  /**
110   * @param txid the transaction id at which this future 'completed'.
111   * @param t Can be null. Set if we are 'completing' on error (and this 't' is the error).
112   * @return True if we successfully marked this outstanding future as completed/done. Returns false
113   *         if this future is already 'done' when this method called.
114   */
115  synchronized boolean done(final long txid, final Throwable t) {
116    if (isDone()) {
117      return false;
118    }
119    this.throwable = t;
120    if (txid < this.txid) {
121      // Something badly wrong.
122      if (throwable == null) {
123        this.throwable =
124            new IllegalStateException("done txid=" + txid + ", my txid=" + this.txid);
125      }
126    }
127    // Mark done.
128    this.doneTxid = txid;
129    // Wake up waiting threads.
130    notify();
131    return true;
132  }
133
134  boolean cancel(boolean mayInterruptIfRunning) {
135    throw new UnsupportedOperationException();
136  }
137
138  synchronized long get(long timeoutNs) throws InterruptedException,
139      ExecutionException, TimeoutIOException {
140    final long done = System.nanoTime() + timeoutNs;
141    while (!isDone()) {
142      wait(1000);
143      if (System.nanoTime() >= done) {
144        throw new TimeoutIOException(
145            "Failed to get sync result after " + TimeUnit.NANOSECONDS.toMillis(timeoutNs)
146                + " ms for txid=" + this.txid + ", WAL system stuck?");
147      }
148    }
149    if (this.throwable != null) {
150      throw new ExecutionException(this.throwable);
151    }
152    return this.doneTxid;
153  }
154
155  synchronized boolean isDone() {
156    return this.doneTxid != NOT_DONE;
157  }
158
159  synchronized boolean isThrowable() {
160    return isDone() && getThrowable() != null;
161  }
162
163  synchronized Throwable getThrowable() {
164    return this.throwable;
165  }
166}