+ * ---------------------- + * | modified | deleted | meaning + * | 0 | 0 | proc exists, but hasn't been updated since last resetUpdates(). + * | 1 | 0 | proc was updated (but not deleted). + * | 1 | 1 | proc was deleted. + * | 0 | 1 | proc doesn't exist (maybe never created, maybe deleted in past). + * ---------------------- + *+ * + * The meaning of modified is that, we have modified the state of the procedure, no matter insert, + * update, or delete. And if it is an insert or update, we will set the deleted to 0, if not we will + * set the delete to 1. + * + * For a non-partial BitSetNode, the initial modified value is 0 and deleted value is 1. For the + * partial one, the initial modified value is 0 and the initial deleted value is also 0. In + * {@link #unsetPartialFlag()} we will reset the deleted to 1 if it is not modified. + */ +@InterfaceAudience.Private +class BitSetNode { + private static final long WORD_MASK = 0xffffffffffffffffL; + private static final int ADDRESS_BITS_PER_WORD = 6; + private static final int BITS_PER_WORD = 1 << ADDRESS_BITS_PER_WORD; + private static final int MAX_NODE_SIZE = 1 << ADDRESS_BITS_PER_WORD; + + /** + * Mimics {@link ProcedureStoreTracker#partial}. It will effect how we fill the new deleted bits + * when growing. + */ + private boolean partial; + + /** + * Set of procedures which have been modified since last {@link #resetModified()}. Useful to track + * procedures which have been modified since last WAL write. + */ + private long[] modified; + + /** + * Keeps track of procedure ids which belong to this bitmap's range and have been deleted. This + * represents global state since it's not reset on WAL rolls. + */ + private long[] deleted; + /** + * Offset of bitmap i.e. procedure id corresponding to first bit. + */ + private long start; + + public void dump() { + System.out.printf("%06d:%06d min=%d max=%d%n", getStart(), getEnd(), getActiveMinProcId(), + getActiveMaxProcId()); + System.out.println("Modified:"); + for (int i = 0; i < modified.length; ++i) { + for (int j = 0; j < BITS_PER_WORD; ++j) { + System.out.print((modified[i] & (1L << j)) != 0 ? "1" : "0"); + } + System.out.println(" " + i); + } + System.out.println(); + System.out.println("Delete:"); + for (int i = 0; i < deleted.length; ++i) { + for (int j = 0; j < BITS_PER_WORD; ++j) { + System.out.print((deleted[i] & (1L << j)) != 0 ? "1" : "0"); + } + System.out.println(" " + i); + } + System.out.println(); + } + + public BitSetNode(long procId, boolean partial) { + start = alignDown(procId); + + int count = 1; + modified = new long[count]; + deleted = new long[count]; + if (!partial) { + Arrays.fill(deleted, WORD_MASK); + } + + this.partial = partial; + updateState(procId, false); + } + + public BitSetNode(ProcedureProtos.ProcedureStoreTracker.TrackerNode data) { + start = data.getStartId(); + int size = data.getUpdatedCount(); + assert size == data.getDeletedCount(); + modified = new long[size]; + deleted = new long[size]; + for (int i = 0; i < size; ++i) { + modified[i] = data.getUpdated(i); + deleted[i] = data.getDeleted(i); + } + partial = false; + } + + public BitSetNode(BitSetNode other, boolean resetDelete) { + this.start = other.start; + this.partial = other.partial; + this.modified = other.modified.clone(); + // The resetDelete will be set to true when building cleanup tracker. + // The intention here is that, if a procedure is not modified in this tracker, then we do not + // need to take care of it, so we will set deleted to true for these bits, i.e, if modified is + // 0, then we set deleted to 1, otherwise keep it as is. So here, the equation is + // deleted |= ~modified, i.e, + if (resetDelete) { + this.deleted = new long[other.deleted.length]; + for (int i = 0; i < this.deleted.length; ++i) { + this.deleted[i] |= ~(other.modified[i]); + } + } else { + this.deleted = other.deleted.clone(); + } + } + + public void insertOrUpdate(final long procId) { + updateState(procId, false); + } + + public void delete(final long procId) { + updateState(procId, true); + } + + public long getStart() { + return start; + } + + public long getEnd() { + return start + (modified.length << ADDRESS_BITS_PER_WORD) - 1; + } + + public boolean contains(final long procId) { + return start <= procId && procId <= getEnd(); + } + + public DeleteState isDeleted(final long procId) { + int bitmapIndex = getBitmapIndex(procId); + int wordIndex = bitmapIndex >> ADDRESS_BITS_PER_WORD; + if (wordIndex >= deleted.length) { + return DeleteState.MAYBE; + } + return (deleted[wordIndex] & (1L << bitmapIndex)) != 0 ? DeleteState.YES : DeleteState.NO; + } + + public boolean isModified(long procId) { + int bitmapIndex = getBitmapIndex(procId); + int wordIndex = bitmapIndex >> ADDRESS_BITS_PER_WORD; + if (wordIndex >= modified.length) { + return false; + } + return (modified[wordIndex] & (1L << bitmapIndex)) != 0; + } + + /** + * @return true, if all the procedures has been modified. + */ + public boolean isAllModified() { + // TODO: cache the value + for (int i = 0; i < modified.length; ++i) { + if ((modified[i] | deleted[i]) != WORD_MASK) { + return false; + } + } + return true; + } + + /** + * @return true, if there are no active procedures in this BitSetNode, else false. + */ + public boolean isEmpty() { + // TODO: cache the value + for (int i = 0; i < deleted.length; ++i) { + if (deleted[i] != WORD_MASK) { + return false; + } + } + return true; + } + + public void resetModified() { + Arrays.fill(modified, 0); + } + + public void unsetPartialFlag() { + partial = false; + for (int i = 0; i < modified.length; ++i) { + for (int j = 0; j < BITS_PER_WORD; ++j) { + if ((modified[i] & (1L << j)) == 0) { + deleted[i] |= (1L << j); + } + } + } + } + + /** + * Convert to + * org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos.ProcedureStoreTracker.TrackerNode + * protobuf. + */ + public ProcedureProtos.ProcedureStoreTracker.TrackerNode convert() { + ProcedureProtos.ProcedureStoreTracker.TrackerNode.Builder builder = + ProcedureProtos.ProcedureStoreTracker.TrackerNode.newBuilder(); + builder.setStartId(start); + for (int i = 0; i < modified.length; ++i) { + builder.addUpdated(modified[i]); + builder.addDeleted(deleted[i]); + } + return builder.build(); + } + + // ======================================================================== + // Grow/Merge Helpers + // ======================================================================== + public boolean canGrow(final long procId) { + return Math.abs(procId - start) < MAX_NODE_SIZE; + } + + public boolean canMerge(final BitSetNode rightNode) { + // Can just compare 'starts' since boundaries are aligned to multiples of BITS_PER_WORD. + assert start < rightNode.start; + return (rightNode.getEnd() - start) < MAX_NODE_SIZE; + } + + public void grow(final long procId) { + int delta, offset; + + if (procId < start) { + // add to head + long newStart = alignDown(procId); + delta = (int) (start - newStart) >> ADDRESS_BITS_PER_WORD; + offset = delta; + start = newStart; + } else { + // Add to tail + long newEnd = alignUp(procId + 1); + delta = (int) (newEnd - getEnd()) >> ADDRESS_BITS_PER_WORD; + offset = 0; + } + + long[] newBitmap; + int oldSize = modified.length; + + newBitmap = new long[oldSize + delta]; + for (int i = 0; i < newBitmap.length; ++i) { + newBitmap[i] = 0; + } + System.arraycopy(modified, 0, newBitmap, offset, oldSize); + modified = newBitmap; + + newBitmap = new long[deleted.length + delta]; + for (int i = 0; i < newBitmap.length; ++i) { + newBitmap[i] = partial ? 0 : WORD_MASK; + } + System.arraycopy(deleted, 0, newBitmap, offset, oldSize); + deleted = newBitmap; + } + + public void merge(final BitSetNode rightNode) { + int delta = (int) (rightNode.getEnd() - getEnd()) >> ADDRESS_BITS_PER_WORD; + + long[] newBitmap; + int oldSize = modified.length; + int newSize = (delta - rightNode.modified.length); + int offset = oldSize + newSize; + + newBitmap = new long[oldSize + delta]; + System.arraycopy(modified, 0, newBitmap, 0, oldSize); + System.arraycopy(rightNode.modified, 0, newBitmap, offset, rightNode.modified.length); + modified = newBitmap; + + newBitmap = new long[oldSize + delta]; + System.arraycopy(deleted, 0, newBitmap, 0, oldSize); + System.arraycopy(rightNode.deleted, 0, newBitmap, offset, rightNode.deleted.length); + deleted = newBitmap; + + for (int i = 0; i < newSize; ++i) { + modified[offset + i] = 0; + deleted[offset + i] = partial ? 0 : WORD_MASK; + } + } + + @Override + public String toString() { + return "BitSetNode(" + getStart() + "-" + getEnd() + ")"; + } + + // ======================================================================== + // Min/Max Helpers + // ======================================================================== + public long getActiveMinProcId() { + long minProcId = start; + for (int i = 0; i < deleted.length; ++i) { + if (deleted[i] == 0) { + return (minProcId); + } + + if (deleted[i] != WORD_MASK) { + for (int j = 0; j < BITS_PER_WORD; ++j) { + if ((deleted[i] & (1L << j)) != 0) { + return minProcId + j; + } + } + } + + minProcId += BITS_PER_WORD; + } + return minProcId; + } + + public long getActiveMaxProcId() { + long maxProcId = getEnd(); + for (int i = deleted.length - 1; i >= 0; --i) { + if (deleted[i] == 0) { + return maxProcId; + } + + if (deleted[i] != WORD_MASK) { + for (int j = BITS_PER_WORD - 1; j >= 0; --j) { + if ((deleted[i] & (1L << j)) == 0) { + return maxProcId - (BITS_PER_WORD - 1 - j); + } + } + } + maxProcId -= BITS_PER_WORD; + } + return maxProcId; + } + + // ======================================================================== + // Bitmap Helpers + // ======================================================================== + private int getBitmapIndex(final long procId) { + return (int) (procId - start); + } + + void updateState(long procId, boolean isDeleted) { + int bitmapIndex = getBitmapIndex(procId); + int wordIndex = bitmapIndex >> ADDRESS_BITS_PER_WORD; + long value = (1L << bitmapIndex); + + modified[wordIndex] |= value; + if (isDeleted) { + deleted[wordIndex] |= value; + } else { + deleted[wordIndex] &= ~value; + } + } + + // ======================================================================== + // Helpers + // ======================================================================== + /** + * @return upper boundary (aligned to multiple of BITS_PER_WORD) of bitmap range x belongs to. + */ + private static long alignUp(final long x) { + return (x + (BITS_PER_WORD - 1)) & -BITS_PER_WORD; + } + + /** + * @return lower boundary (aligned to multiple of BITS_PER_WORD) of bitmap range x belongs to. + */ + private static long alignDown(final long x) { + return x & -BITS_PER_WORD; + } +} \ No newline at end of file diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/NoopProcedureStore.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/NoopProcedureStore.java index 9c6176d4bb8..8fbc1473ed7 100644 --- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/NoopProcedureStore.java +++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/NoopProcedureStore.java @@ -15,7 +15,6 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.hadoop.hbase.procedure2.store; import java.io.IOException; @@ -64,17 +63,17 @@ public class NoopProcedureStore extends ProcedureStoreBase { } @Override - public void insert(Procedure proc, Procedure[] subprocs) { + public void insert(Procedure proc, Procedure[] subprocs) { // no-op } @Override - public void insert(Procedure[] proc) { + public void insert(Procedure[] proc) { // no-op } @Override - public void update(Procedure proc) { + public void update(Procedure proc) { // no-op } @@ -84,7 +83,7 @@ public class NoopProcedureStore extends ProcedureStoreBase { } @Override - public void delete(Procedure proc, long[] subprocs) { + public void delete(Procedure proc, long[] subprocs) { // no-op } diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStore.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStore.java index 72883405d7c..8063b125ba5 100644 --- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStore.java +++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStore.java @@ -81,6 +81,7 @@ public interface ProcedureStore { * @throws IOException if there was an error fetching/deserializing the procedure * @return the next procedure in the iteration. */ + @SuppressWarnings("rawtypes") Procedure next() throws IOException; } @@ -173,7 +174,7 @@ public interface ProcedureStore { * @param proc the procedure to serialize and write to the store. * @param subprocs the newly created child of the proc. */ - void insert(Procedure proc, Procedure[] subprocs); + void insert(Procedure proc, Procedure[] subprocs); /** * Serialize a set of new procedures. @@ -182,14 +183,14 @@ public interface ProcedureStore { * * @param procs the procedures to serialize and write to the store. */ - void insert(Procedure[] procs); + void insert(Procedure[] procs); /** * The specified procedure was executed, * and the new state should be written to the store. * @param proc the procedure to serialize and write to the store. */ - void update(Procedure proc); + void update(Procedure proc); /** * The specified procId was removed from the executor, @@ -205,7 +206,7 @@ public interface ProcedureStore { * @param parentProc the parent procedure to serialize and write to the store. * @param subProcIds the IDs of the sub-procedure to remove. */ - void delete(Procedure parentProc, long[] subProcIds); + void delete(Procedure parentProc, long[] subProcIds); /** * The specified procIds were removed from the executor, diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStoreTracker.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStoreTracker.java index 2dad5ac72ce..361419ab48f 100644 --- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStoreTracker.java +++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/ProcedureStoreTracker.java @@ -53,383 +53,14 @@ public class ProcedureStoreTracker { * It's set to true only when recovering from old logs. See {@link #isDeleted(long)} docs to * understand it's real use. */ - private boolean partial = false; + boolean partial = false; - private long minUpdatedProcId = Long.MAX_VALUE; - private long maxUpdatedProcId = Long.MIN_VALUE; + private long minModifiedProcId = Long.MAX_VALUE; + private long maxModifiedProcId = Long.MIN_VALUE; public enum DeleteState { YES, NO, MAYBE } - /** - * A bitmap which can grow/merge with other {@link BitSetNode} (if certain conditions are met). - * Boundaries of bitmap are aligned to multiples of {@link BitSetNode#BITS_PER_WORD}. So the - * range of a {@link BitSetNode} is from [x * K, y * K) where x and y are integers, y > x and K - * is BITS_PER_WORD. - */ - public static class BitSetNode { - private final static long WORD_MASK = 0xffffffffffffffffL; - private final static int ADDRESS_BITS_PER_WORD = 6; - private final static int BITS_PER_WORD = 1 << ADDRESS_BITS_PER_WORD; - private final static int MAX_NODE_SIZE = 1 << ADDRESS_BITS_PER_WORD; - - /** - * Mimics {@link ProcedureStoreTracker#partial}. - */ - private final boolean partial; - - /* ---------------------- - * | updated | deleted | meaning - * | 0 | 0 | proc exists, but hasn't been updated since last resetUpdates(). - * | 1 | 0 | proc was updated (but not deleted). - * | 1 | 1 | proc was deleted. - * | 0 | 1 | proc doesn't exist (maybe never created, maybe deleted in past). - /* ---------------------- - */ - - /** - * Set of procedures which have been updated since last {@link #resetUpdates()}. - * Useful to track procedures which have been updated since last WAL write. - */ - private long[] updated; - /** - * Keeps track of procedure ids which belong to this bitmap's range and have been deleted. - * This represents global state since it's not reset on WAL rolls. - */ - private long[] deleted; - /** - * Offset of bitmap i.e. procedure id corresponding to first bit. - */ - private long start; - - public void dump() { - System.out.printf("%06d:%06d min=%d max=%d%n", getStart(), getEnd(), - getActiveMinProcId(), getActiveMaxProcId()); - System.out.println("Update:"); - for (int i = 0; i < updated.length; ++i) { - for (int j = 0; j < BITS_PER_WORD; ++j) { - System.out.print((updated[i] & (1L << j)) != 0 ? "1" : "0"); - } - System.out.println(" " + i); - } - System.out.println(); - System.out.println("Delete:"); - for (int i = 0; i < deleted.length; ++i) { - for (int j = 0; j < BITS_PER_WORD; ++j) { - System.out.print((deleted[i] & (1L << j)) != 0 ? "1" : "0"); - } - System.out.println(" " + i); - } - System.out.println(); - } - - public BitSetNode(final long procId, final boolean partial) { - start = alignDown(procId); - - int count = 1; - updated = new long[count]; - deleted = new long[count]; - for (int i = 0; i < count; ++i) { - updated[i] = 0; - deleted[i] = partial ? 0 : WORD_MASK; - } - - this.partial = partial; - updateState(procId, false); - } - - protected BitSetNode(final long start, final long[] updated, final long[] deleted) { - this.start = start; - this.updated = updated; - this.deleted = deleted; - this.partial = false; - } - - public BitSetNode(ProcedureProtos.ProcedureStoreTracker.TrackerNode data) { - start = data.getStartId(); - int size = data.getUpdatedCount(); - updated = new long[size]; - deleted = new long[size]; - for (int i = 0; i < size; ++i) { - updated[i] = data.getUpdated(i); - deleted[i] = data.getDeleted(i); - } - partial = false; - } - - public BitSetNode(final BitSetNode other, final boolean resetDelete) { - this.start = other.start; - this.partial = other.partial; - this.updated = other.updated.clone(); - if (resetDelete) { - this.deleted = new long[other.deleted.length]; - for (int i = 0; i < this.deleted.length; ++i) { - this.deleted[i] = ~(other.updated[i]); - } - } else { - this.deleted = other.deleted.clone(); - } - } - - public void update(final long procId) { - updateState(procId, false); - } - - public void delete(final long procId) { - updateState(procId, true); - } - - public long getStart() { - return start; - } - - public long getEnd() { - return start + (updated.length << ADDRESS_BITS_PER_WORD) - 1; - } - - public boolean contains(final long procId) { - return start <= procId && procId <= getEnd(); - } - - public DeleteState isDeleted(final long procId) { - int bitmapIndex = getBitmapIndex(procId); - int wordIndex = bitmapIndex >> ADDRESS_BITS_PER_WORD; - if (wordIndex >= deleted.length) { - return DeleteState.MAYBE; - } - return (deleted[wordIndex] & (1L << bitmapIndex)) != 0 ? DeleteState.YES : DeleteState.NO; - } - - private boolean isUpdated(final long procId) { - int bitmapIndex = getBitmapIndex(procId); - int wordIndex = bitmapIndex >> ADDRESS_BITS_PER_WORD; - if (wordIndex >= updated.length) { - return false; - } - return (updated[wordIndex] & (1L << bitmapIndex)) != 0; - } - - public boolean isUpdated() { - // TODO: cache the value - for (int i = 0; i < updated.length; ++i) { - if ((updated[i] | deleted[i]) != WORD_MASK) { - return false; - } - } - return true; - } - - /** - * @return true, if there are no active procedures in this BitSetNode, else false. - */ - public boolean isEmpty() { - // TODO: cache the value - for (int i = 0; i < deleted.length; ++i) { - if (deleted[i] != WORD_MASK) { - return false; - } - } - return true; - } - - public void resetUpdates() { - for (int i = 0; i < updated.length; ++i) { - updated[i] = 0; - } - } - - /** - * Clears the {@link #deleted} bitmaps. - */ - public void undeleteAll() { - for (int i = 0; i < updated.length; ++i) { - deleted[i] = 0; - } - } - - public void unsetPartialFlag() { - for (int i = 0; i < updated.length; ++i) { - for (int j = 0; j < BITS_PER_WORD; ++j) { - if ((updated[i] & (1L << j)) == 0) { - deleted[i] |= (1L << j); - } - } - } - } - - /** - * Convert to - * org.apache.hadoop.hbase.protobuf.generated.ProcedureProtos.ProcedureStoreTracker.TrackerNode - * protobuf. - */ - public ProcedureProtos.ProcedureStoreTracker.TrackerNode convert() { - ProcedureProtos.ProcedureStoreTracker.TrackerNode.Builder builder = - ProcedureProtos.ProcedureStoreTracker.TrackerNode.newBuilder(); - builder.setStartId(start); - for (int i = 0; i < updated.length; ++i) { - builder.addUpdated(updated[i]); - builder.addDeleted(deleted[i]); - } - return builder.build(); - } - - // ======================================================================== - // Grow/Merge Helpers - // ======================================================================== - public boolean canGrow(final long procId) { - return Math.abs(procId - start) < MAX_NODE_SIZE; - } - - public boolean canMerge(final BitSetNode rightNode) { - // Can just compare 'starts' since boundaries are aligned to multiples of BITS_PER_WORD. - assert start < rightNode.start; - return (rightNode.getEnd() - start) < MAX_NODE_SIZE; - } - - public void grow(final long procId) { - int delta, offset; - - if (procId < start) { - // add to head - long newStart = alignDown(procId); - delta = (int)(start - newStart) >> ADDRESS_BITS_PER_WORD; - offset = delta; - start = newStart; - } else { - // Add to tail - long newEnd = alignUp(procId + 1); - delta = (int)(newEnd - getEnd()) >> ADDRESS_BITS_PER_WORD; - offset = 0; - } - - long[] newBitmap; - int oldSize = updated.length; - - newBitmap = new long[oldSize + delta]; - for (int i = 0; i < newBitmap.length; ++i) { - newBitmap[i] = 0; - } - System.arraycopy(updated, 0, newBitmap, offset, oldSize); - updated = newBitmap; - - newBitmap = new long[deleted.length + delta]; - for (int i = 0; i < newBitmap.length; ++i) { - newBitmap[i] = partial ? 0 : WORD_MASK; - } - System.arraycopy(deleted, 0, newBitmap, offset, oldSize); - deleted = newBitmap; - } - - public void merge(final BitSetNode rightNode) { - int delta = (int)(rightNode.getEnd() - getEnd()) >> ADDRESS_BITS_PER_WORD; - - long[] newBitmap; - int oldSize = updated.length; - int newSize = (delta - rightNode.updated.length); - int offset = oldSize + newSize; - - newBitmap = new long[oldSize + delta]; - System.arraycopy(updated, 0, newBitmap, 0, oldSize); - System.arraycopy(rightNode.updated, 0, newBitmap, offset, rightNode.updated.length); - updated = newBitmap; - - newBitmap = new long[oldSize + delta]; - System.arraycopy(deleted, 0, newBitmap, 0, oldSize); - System.arraycopy(rightNode.deleted, 0, newBitmap, offset, rightNode.deleted.length); - deleted = newBitmap; - - for (int i = 0; i < newSize; ++i) { - updated[offset + i] = 0; - deleted[offset + i] = partial ? 0 : WORD_MASK; - } - } - - @Override - public String toString() { - return "BitSetNode(" + getStart() + "-" + getEnd() + ")"; - } - - // ======================================================================== - // Min/Max Helpers - // ======================================================================== - public long getActiveMinProcId() { - long minProcId = start; - for (int i = 0; i < deleted.length; ++i) { - if (deleted[i] == 0) { - return(minProcId); - } - - if (deleted[i] != WORD_MASK) { - for (int j = 0; j < BITS_PER_WORD; ++j) { - if ((deleted[i] & (1L << j)) != 0) { - return minProcId + j; - } - } - } - - minProcId += BITS_PER_WORD; - } - return minProcId; - } - - public long getActiveMaxProcId() { - long maxProcId = getEnd(); - for (int i = deleted.length - 1; i >= 0; --i) { - if (deleted[i] == 0) { - return maxProcId; - } - - if (deleted[i] != WORD_MASK) { - for (int j = BITS_PER_WORD - 1; j >= 0; --j) { - if ((deleted[i] & (1L << j)) == 0) { - return maxProcId - (BITS_PER_WORD - 1 - j); - } - } - } - maxProcId -= BITS_PER_WORD; - } - return maxProcId; - } - - // ======================================================================== - // Bitmap Helpers - // ======================================================================== - private int getBitmapIndex(final long procId) { - return (int)(procId - start); - } - - private void updateState(final long procId, final boolean isDeleted) { - int bitmapIndex = getBitmapIndex(procId); - int wordIndex = bitmapIndex >> ADDRESS_BITS_PER_WORD; - long value = (1L << bitmapIndex); - - updated[wordIndex] |= value; - if (isDeleted) { - deleted[wordIndex] |= value; - } else { - deleted[wordIndex] &= ~value; - } - } - - - // ======================================================================== - // Helpers - // ======================================================================== - /** - * @return upper boundary (aligned to multiple of BITS_PER_WORD) of bitmap range x belongs to. - */ - private static long alignUp(final long x) { - return (x + (BITS_PER_WORD - 1)) & -BITS_PER_WORD; - } - - /** - * @return lower boundary (aligned to multiple of BITS_PER_WORD) of bitmap range x belongs to. - */ - private static long alignDown(final long x) { - return x & -BITS_PER_WORD; - } - } - - public void resetToProto(final ProcedureProtos.ProcedureStoreTracker trackerProtoBuf) { + public void resetToProto(ProcedureProtos.ProcedureStoreTracker trackerProtoBuf) { reset(); for (ProcedureProtos.ProcedureStoreTracker.TrackerNode protoNode: trackerProtoBuf.getNodeList()) { final BitSetNode node = new BitSetNode(protoNode); @@ -440,14 +71,23 @@ public class ProcedureStoreTracker { /** * Resets internal state to same as given {@code tracker}. Does deep copy of the bitmap. */ - public void resetTo(final ProcedureStoreTracker tracker) { + public void resetTo(ProcedureStoreTracker tracker) { resetTo(tracker, false); } - public void resetTo(final ProcedureStoreTracker tracker, final boolean resetDelete) { + /** + * Resets internal state to same as given {@code tracker}, and change the deleted flag according + * to the modified flag if {@code resetDelete} is true. Does deep copy of the bitmap. + * + * The {@code resetDelete} will be set to true when building cleanup tracker, please see the + * comments in {@link BitSetNode#BitSetNode(BitSetNode, boolean)} to learn how we change the + * deleted flag if {@code resetDelete} is true. + */ + public void resetTo(ProcedureStoreTracker tracker, boolean resetDelete) { + reset(); this.partial = tracker.partial; - this.minUpdatedProcId = tracker.minUpdatedProcId; - this.maxUpdatedProcId = tracker.maxUpdatedProcId; + this.minModifiedProcId = tracker.minModifiedProcId; + this.maxModifiedProcId = tracker.maxModifiedProcId; this.keepDeletes = tracker.keepDeletes; for (Map.Entry
+ * procedureMap = | A | | E | | C | | | | | G | | | + * D B + * replayOrderHead = C <-> B <-> E <-> D <-> A <-> G + * + * We also have a lazy grouping by "root procedure", and a list of + * unlinked procedures. If after reading all the WALs we have unlinked + * procedures it means that we had a missing WAL or a corruption. + * rootHead = A <-> D <-> G + * B E + * C + * unlinkFromLinkList = None + *+ */ +class WALProcedureMap { + + private static final Logger LOG = LoggerFactory.getLogger(WALProcedureMap.class); + + private static class Entry { + // For bucketed linked lists in hash-table. + private Entry hashNext; + // child head + private Entry childHead; + // double-link for rootHead or childHead + private Entry linkNext; + private Entry linkPrev; + // replay double-linked-list + private Entry replayNext; + private Entry replayPrev; + // procedure-infos + private Procedure procedure; + private ProcedureProtos.Procedure proto; + private boolean ready = false; + + public Entry(Entry hashNext) { + this.hashNext = hashNext; + } + + public long getProcId() { + return proto.getProcId(); + } + + public long getParentId() { + return proto.getParentId(); + } + + public boolean hasParent() { + return proto.hasParentId(); + } + + public boolean isReady() { + return ready; + } + + public boolean isFinished() { + if (!hasParent()) { + // we only consider 'root' procedures. because for the user 'finished' + // means when everything up to the 'root' is finished. + switch (proto.getState()) { + case ROLLEDBACK: + case SUCCESS: + return true; + default: + break; + } + } + return false; + } + + public Procedure convert() throws IOException { + if (procedure == null) { + procedure = ProcedureUtil.convertToProcedure(proto); + } + return procedure; + } + + @Override + public String toString() { + final StringBuilder sb = new StringBuilder(); + sb.append("Entry("); + sb.append(getProcId()); + sb.append(", parentId="); + sb.append(getParentId()); + sb.append(", class="); + sb.append(proto.getClassName()); + sb.append(")"); + return sb.toString(); + } + } + + private static class EntryIterator implements ProcedureIterator { + private final Entry replayHead; + private Entry current; + + public EntryIterator(Entry replayHead) { + this.replayHead = replayHead; + this.current = replayHead; + } + + @Override + public void reset() { + this.current = replayHead; + } + + @Override + public boolean hasNext() { + return current != null; + } + + @Override + public boolean isNextFinished() { + return current != null && current.isFinished(); + } + + @Override + public void skipNext() { + current = current.replayNext; + } + + @Override + public Procedure next() throws IOException { + try { + return current.convert(); + } finally { + current = current.replayNext; + } + } + } + + // procedure hash table + private Entry[] procedureMap; + + // replay-order double-linked-list + private Entry replayOrderHead; + private Entry replayOrderTail; + + // root linked-list + private Entry rootHead; + + // pending unlinked children (root not present yet) + private Entry childUnlinkedHead; + + // Track ProcId range + private long minModifiedProcId = Long.MAX_VALUE; + private long maxModifiedProcId = Long.MIN_VALUE; + + public WALProcedureMap(int size) { + procedureMap = new Entry[size]; + replayOrderHead = null; + replayOrderTail = null; + rootHead = null; + childUnlinkedHead = null; + } + + public void add(ProcedureProtos.Procedure procProto) { + trackProcIds(procProto.getProcId()); + Entry entry = addToMap(procProto.getProcId(), procProto.hasParentId()); + boolean newEntry = entry.proto == null; + // We have seen procedure WALs where the entries are out of order; see HBASE-18152. + // To compensate, only replace the Entry procedure if for sure this new procedure + // is indeed an entry that came later. + // TODO: Fix the writing of procedure info so it does not violate basic expectation, that WALs + // contain procedure changes goingfrom start to finish in sequence. + if (newEntry || isIncreasing(entry.proto, procProto)) { + entry.proto = procProto; + } + addToReplayList(entry); + if (newEntry) { + if (procProto.hasParentId()) { + childUnlinkedHead = addToLinkList(entry, childUnlinkedHead); + } else { + rootHead = addToLinkList(entry, rootHead); + } + } + } + + /** + * @return True if this new procedure is 'richer' than the current one else false and we log this + * incidence where it appears that the WAL has older entries appended after newer ones. + * See HBASE-18152. + */ + private static boolean isIncreasing(ProcedureProtos.Procedure current, + ProcedureProtos.Procedure candidate) { + // Check that the procedures we see are 'increasing'. We used to compare + // procedure id first and then update time but it can legitimately go backwards if the + // procedure is failed or rolled back so that was unreliable. Was going to compare + // state but lets see if comparing update time enough (unfortunately this issue only + // seen under load...) + boolean increasing = current.getLastUpdate() <= candidate.getLastUpdate(); + if (!increasing) { + LOG.warn("NOT INCREASING! current=" + current + ", candidate=" + candidate); + } + return increasing; + } + + public boolean remove(long procId) { + trackProcIds(procId); + Entry entry = removeFromMap(procId); + if (entry != null) { + unlinkFromReplayList(entry); + unlinkFromLinkList(entry); + return true; + } + return false; + } + + private void trackProcIds(long procId) { + minModifiedProcId = Math.min(minModifiedProcId, procId); + maxModifiedProcId = Math.max(maxModifiedProcId, procId); + } + + public long getMinModifiedProcId() { + return minModifiedProcId; + } + + public long getMaxModifiedProcId() { + return maxModifiedProcId; + } + + public boolean contains(long procId) { + return getProcedure(procId) != null; + } + + public boolean isEmpty() { + return replayOrderHead == null; + } + + public void clear() { + for (int i = 0; i < procedureMap.length; ++i) { + procedureMap[i] = null; + } + replayOrderHead = null; + replayOrderTail = null; + rootHead = null; + childUnlinkedHead = null; + minModifiedProcId = Long.MAX_VALUE; + maxModifiedProcId = Long.MIN_VALUE; + } + + /* + * Merges two WalProcedureMap, the target is the "global" map, the source is the "local" map. - + * The entries in the hashtables are guaranteed to be unique. On replay we don't load procedures + * that already exist in the "global" map (the one we are merging the "local" in to). - The + * replayOrderList of the "local" nao will be appended to the "global" map replay list. - The + * "local" map will be cleared at the end of the operation. + */ + public void mergeTail(WALProcedureMap other) { + for (Entry p = other.replayOrderHead; p != null; p = p.replayNext) { + int slotIndex = getMapSlot(p.getProcId()); + p.hashNext = procedureMap[slotIndex]; + procedureMap[slotIndex] = p; + } + + if (replayOrderHead == null) { + replayOrderHead = other.replayOrderHead; + replayOrderTail = other.replayOrderTail; + rootHead = other.rootHead; + childUnlinkedHead = other.childUnlinkedHead; + } else { + // append replay list + assert replayOrderTail.replayNext == null; + assert other.replayOrderHead.replayPrev == null; + replayOrderTail.replayNext = other.replayOrderHead; + other.replayOrderHead.replayPrev = replayOrderTail; + replayOrderTail = other.replayOrderTail; + + // merge rootHead + if (rootHead == null) { + rootHead = other.rootHead; + } else if (other.rootHead != null) { + Entry otherTail = findLinkListTail(other.rootHead); + otherTail.linkNext = rootHead; + rootHead.linkPrev = otherTail; + rootHead = other.rootHead; + } + + // merge childUnlinkedHead + if (childUnlinkedHead == null) { + childUnlinkedHead = other.childUnlinkedHead; + } else if (other.childUnlinkedHead != null) { + Entry otherTail = findLinkListTail(other.childUnlinkedHead); + otherTail.linkNext = childUnlinkedHead; + childUnlinkedHead.linkPrev = otherTail; + childUnlinkedHead = other.childUnlinkedHead; + } + } + maxModifiedProcId = Math.max(maxModifiedProcId, other.maxModifiedProcId); + minModifiedProcId = Math.max(minModifiedProcId, other.minModifiedProcId); + + other.clear(); + } + + /** + * Returns an EntryIterator with the list of procedures ready to be added to the executor. A + * Procedure is ready if its children and parent are ready. + */ + public ProcedureIterator fetchReady() { + buildGraph(); + + Entry readyHead = null; + Entry readyTail = null; + Entry p = replayOrderHead; + while (p != null) { + Entry next = p.replayNext; + if (p.isReady()) { + unlinkFromReplayList(p); + if (readyTail != null) { + readyTail.replayNext = p; + p.replayPrev = readyTail; + } else { + p.replayPrev = null; + readyHead = p; + } + readyTail = p; + p.replayNext = null; + } + p = next; + } + // we need the hash-table lookups for parents, so this must be done + // out of the loop where we check isReadyToRun() + for (p = readyHead; p != null; p = p.replayNext) { + removeFromMap(p.getProcId()); + unlinkFromLinkList(p); + } + return readyHead != null ? new EntryIterator(readyHead) : null; + } + + /** + * Drain this map and return all procedures in it. + */ + public ProcedureIterator fetchAll() { + Entry head = replayOrderHead; + for (Entry p = head; p != null; p = p.replayNext) { + removeFromMap(p.getProcId()); + } + for (int i = 0; i < procedureMap.length; ++i) { + assert procedureMap[i] == null : "map not empty i=" + i; + } + replayOrderHead = null; + replayOrderTail = null; + childUnlinkedHead = null; + rootHead = null; + return head != null ? new EntryIterator(head) : null; + } + + private void buildGraph() { + Entry p = childUnlinkedHead; + while (p != null) { + Entry next = p.linkNext; + Entry rootProc = getRootProcedure(p); + if (rootProc != null) { + rootProc.childHead = addToLinkList(p, rootProc.childHead); + } + p = next; + } + + for (p = rootHead; p != null; p = p.linkNext) { + checkReadyToRun(p); + } + } + + private Entry getRootProcedure(Entry entry) { + while (entry != null && entry.hasParent()) { + entry = getProcedure(entry.getParentId()); + } + return entry; + } + + /** + * (see the comprehensive explanation in the beginning of {@link ProcedureWALFormatReader}). A + * Procedure is ready when parent and children are ready. "ready" means that we all the + * information that we need in-memory. + * + * Example-1:
+ * wal-2 | C B | + * wal-1 | A B C | + *+ * + * If C and B don't depend on A (A is not the parent), we can start them before reading wal-1. If + * B is the only one with parent A we can start C. We have to read one more WAL before being able + * to start B. + * + * How do we know with the only information in B that we are not ready. + *
+ * wal-2: B has stackId = [1, 2] + * wal-1: B has stackId = [1] + * wal-1: A has stackId = [0] + *+ * + * Since we know that the Stack-IDs are incremental for a Procedure, we notice that there is a gap + * in the stackIds of B, so something was executed before. + * + * To identify when a Procedure is ready we do the sum of the stackIds of the procedure and the + * parent. if the stackIdSum is equal to the sum of {1..maxStackId} then everything we need is + * available. + * + * Example-2 + * + *
+ * wal-2 | A | A stackIds = [0, 2] + * wal-1 | A B | B stackIds = [1] + *+ * + * There is a gap between A stackIds so something was executed in between. + */ + private boolean checkReadyToRun(Entry rootEntry) { + assert !rootEntry.hasParent() : "expected root procedure, got " + rootEntry; + + if (rootEntry.isFinished()) { + // If the root procedure is finished, sub-procedures should be gone + if (rootEntry.childHead != null) { + LOG.error("unexpected active children for root-procedure: {}", rootEntry); + for (Entry p = rootEntry.childHead; p != null; p = p.linkNext) { + LOG.error("unexpected active children: {}", p); + } + } + + assert rootEntry.childHead == null : "unexpected children on root completion. " + rootEntry; + rootEntry.ready = true; + return true; + } + + int stackIdSum = 0; + int maxStackId = 0; + for (int i = 0; i < rootEntry.proto.getStackIdCount(); ++i) { + int stackId = 1 + rootEntry.proto.getStackId(i); + maxStackId = Math.max(maxStackId, stackId); + stackIdSum += stackId; + LOG.trace("stackId={} stackIdSum={} maxStackid={} {}", stackId, stackIdSum, maxStackId, + rootEntry); + } + + for (Entry p = rootEntry.childHead; p != null; p = p.linkNext) { + for (int i = 0; i < p.proto.getStackIdCount(); ++i) { + int stackId = 1 + p.proto.getStackId(i); + maxStackId = Math.max(maxStackId, stackId); + stackIdSum += stackId; + LOG.trace("stackId={} stackIdSum={} maxStackid={} {}", stackId, stackIdSum, maxStackId, p); + } + } + // The cmpStackIdSum is this formula for finding the sum of a series of numbers: + // http://www.wikihow.com/Sum-the-Integers-from-1-to-N#/Image:Sum-the-Integers-from-1-to-N-Step-2-Version-3.jpg + final int cmpStackIdSum = (maxStackId * (maxStackId + 1) / 2); + if (cmpStackIdSum == stackIdSum) { + rootEntry.ready = true; + for (Entry p = rootEntry.childHead; p != null; p = p.linkNext) { + p.ready = true; + } + return true; + } + return false; + } + + private void unlinkFromReplayList(Entry entry) { + if (replayOrderHead == entry) { + replayOrderHead = entry.replayNext; + } + if (replayOrderTail == entry) { + replayOrderTail = entry.replayPrev; + } + if (entry.replayPrev != null) { + entry.replayPrev.replayNext = entry.replayNext; + } + if (entry.replayNext != null) { + entry.replayNext.replayPrev = entry.replayPrev; + } + } + + private void addToReplayList(final Entry entry) { + unlinkFromReplayList(entry); + entry.replayNext = replayOrderHead; + entry.replayPrev = null; + if (replayOrderHead != null) { + replayOrderHead.replayPrev = entry; + } else { + replayOrderTail = entry; + } + replayOrderHead = entry; + } + + private void unlinkFromLinkList(Entry entry) { + if (entry == rootHead) { + rootHead = entry.linkNext; + } else if (entry == childUnlinkedHead) { + childUnlinkedHead = entry.linkNext; + } + if (entry.linkPrev != null) { + entry.linkPrev.linkNext = entry.linkNext; + } + if (entry.linkNext != null) { + entry.linkNext.linkPrev = entry.linkPrev; + } + } + + private Entry addToLinkList(Entry entry, Entry linkHead) { + unlinkFromLinkList(entry); + entry.linkNext = linkHead; + entry.linkPrev = null; + if (linkHead != null) { + linkHead.linkPrev = entry; + } + return entry; + } + + private Entry findLinkListTail(Entry linkHead) { + Entry tail = linkHead; + while (tail.linkNext != null) { + tail = tail.linkNext; + } + return tail; + } + + private Entry addToMap(long procId, boolean hasParent) { + int slotIndex = getMapSlot(procId); + Entry entry = getProcedure(slotIndex, procId); + if (entry != null) { + return entry; + } + + entry = new Entry(procedureMap[slotIndex]); + procedureMap[slotIndex] = entry; + return entry; + } + + private Entry removeFromMap(final long procId) { + int slotIndex = getMapSlot(procId); + Entry prev = null; + Entry entry = procedureMap[slotIndex]; + while (entry != null) { + if (procId == entry.getProcId()) { + if (prev != null) { + prev.hashNext = entry.hashNext; + } else { + procedureMap[slotIndex] = entry.hashNext; + } + entry.hashNext = null; + return entry; + } + prev = entry; + entry = entry.hashNext; + } + return null; + } + + private Entry getProcedure(long procId) { + return getProcedure(getMapSlot(procId), procId); + } + + private Entry getProcedure(int slotIndex, long procId) { + Entry entry = procedureMap[slotIndex]; + while (entry != null) { + if (procId == entry.getProcId()) { + return entry; + } + entry = entry.hashNext; + } + return null; + } + + private int getMapSlot(long procId) { + return (int) (Procedure.getProcIdHashCode(procId) % procedureMap.length); + } +} \ No newline at end of file diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java index 7d5d6d24488..00dfe85c1ca 100644 --- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java +++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java @@ -15,14 +15,12 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.apache.hadoop.hbase.procedure2.store.wal; import java.io.FileNotFoundException; import java.io.IOException; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collections; import java.util.Comparator; import java.util.HashSet; import java.util.Iterator; @@ -35,7 +33,6 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReentrantLock; - import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileAlreadyExistsException; @@ -52,25 +49,60 @@ import org.apache.hadoop.hbase.procedure2.store.ProcedureStoreBase; import org.apache.hadoop.hbase.procedure2.store.ProcedureStoreTracker; import org.apache.hadoop.hbase.procedure2.util.ByteSlot; import org.apache.hadoop.hbase.procedure2.util.StringUtils; -import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos.ProcedureWALHeader; import org.apache.hadoop.hbase.util.CommonFSUtils; import org.apache.hadoop.hbase.util.Threads; import org.apache.hadoop.ipc.RemoteException; import org.apache.yetus.audience.InterfaceAudience; -import org.apache.yetus.audience.InterfaceStability; import org.slf4j.Logger; import org.slf4j.LoggerFactory; + import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; import org.apache.hbase.thirdparty.org.apache.commons.collections4.queue.CircularFifoQueue; +import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos.ProcedureWALHeader; /** * WAL implementation of the ProcedureStore. + * + * When starting, the upper layer will first call {@link #start(int)}, then {@link #recoverLease()}, + * then {@link #load(ProcedureLoader)}. + * + * In {@link #recoverLease()}, we will get the lease by closing all the existing wal files(by + * calling recoverFileLease), and creating a new wal writer. And we will also get the list of all + * the old wal files. + * + * FIXME: notice that the current recover lease implementation is problematic, it can not deal with + * the races if there are two master both wants to acquire the lease... + * + * In {@link #load(ProcedureLoader)} method, we will load all the active procedures. See the + * comments of this method for more details. + * + * The actual logging way is a bit like our FileSystem based WAL implementation as RS side. There is + * a {@link #slots}, which is more like the ring buffer, and in the insert, update and delete + * methods we will put thing into the {@link #slots} and wait. And there is a background sync + * thread(see the {@link #syncLoop()} method) which get data from the {@link #slots} and write them + * to the FileSystem, and notify the caller that we have finished. + * + * TODO: try using disruptor to increase performance and simplify the logic? + * + * The {@link #storeTracker} keeps track of the modified procedures in the newest wal file, which is + * also the one being written currently. And the deleted bits in it are for all the procedures, not + * only the ones in the newest wal file. And when rolling a log, we will first store it in the + * trailer of the current wal file, and then reset its modified bits, so that it can start to track + * the modified procedures for the new wal file. + * + * The {@link #holdingCleanupTracker} is used to test whether we are safe to delete the oldest wal + * file. When there are log rolling and there are more than 1 wal files, we will make use of it. It + * will first be initialized to the oldest file's tracker(which is stored in the trailer), using the + * method {@link ProcedureStoreTracker#resetTo(ProcedureStoreTracker, boolean)}, and then merge it + * with the tracker of every newer wal files, using the + * {@link ProcedureStoreTracker#setDeletedIfModifiedInBoth(ProcedureStoreTracker)}. If we find out + * that all the modified procedures for the oldest wal file are modified or deleted in newer wal + * files, then we can delete it. * @see ProcedureWALPrettyPrinter for printing content of a single WAL. * @see #main(String[]) to parse a directory of MasterWALProcs. */ @InterfaceAudience.Private -@InterfaceStability.Evolving public class WALProcedureStore extends ProcedureStoreBase { private static final Logger LOG = LoggerFactory.getLogger(WALProcedureStore.class); public static final String LOG_PREFIX = "pv2-"; @@ -166,7 +198,7 @@ public class WALProcedureStore extends ProcedureStoreBase { private int syncWaitMsec; // Variables used for UI display - private CircularFifoQueue syncMetricsQueue; + private CircularFifoQueue
To preserve test data directories, pass the system property "hbase.testing.preserve.testdir" * setting it to true. + * For triggering test. */ @InterfaceAudience.Public @SuppressWarnings("deprecation")