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}