| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance with the License. You may obtain a |
| * copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License |
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
| * or implied. See the License for the specific language governing permissions and limitations under |
| * the License. |
| */ |
| |
| package org.apache.fluo.core.impl; |
| |
| import java.util.ArrayList; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Objects; |
| import java.util.Set; |
| import java.util.concurrent.CompletableFuture; |
| import java.util.concurrent.CompletionException; |
| import java.util.concurrent.Executor; |
| import java.util.function.Consumer; |
| import java.util.function.Supplier; |
| |
| import com.google.common.annotations.VisibleForTesting; |
| import com.google.common.base.Preconditions; |
| import com.google.common.collect.Iterables; |
| import com.google.common.collect.Iterators; |
| import com.google.common.collect.Sets; |
| import org.apache.accumulo.core.client.ConditionalWriter; |
| import org.apache.accumulo.core.client.ConditionalWriter.Result; |
| import org.apache.accumulo.core.client.ConditionalWriter.Status; |
| import org.apache.accumulo.core.client.IteratorSetting; |
| import org.apache.accumulo.core.client.Scanner; |
| import org.apache.accumulo.core.client.TableNotFoundException; |
| import org.apache.accumulo.core.data.ColumnUpdate; |
| import org.apache.accumulo.core.data.Condition; |
| import org.apache.accumulo.core.data.ConditionalMutation; |
| import org.apache.accumulo.core.data.Key; |
| import org.apache.accumulo.core.data.Mutation; |
| import org.apache.accumulo.core.data.Range; |
| import org.apache.accumulo.core.data.Value; |
| import org.apache.fluo.accumulo.iterators.PrewriteIterator; |
| import org.apache.fluo.accumulo.util.ColumnConstants; |
| import org.apache.fluo.accumulo.util.ColumnType; |
| import org.apache.fluo.accumulo.util.ReadLockUtil; |
| import org.apache.fluo.accumulo.values.DelLockValue; |
| import org.apache.fluo.accumulo.values.DelReadLockValue; |
| import org.apache.fluo.accumulo.values.LockValue; |
| import org.apache.fluo.accumulo.values.ReadLockValue; |
| import org.apache.fluo.api.client.AbstractTransactionBase; |
| import org.apache.fluo.api.client.Snapshot; |
| import org.apache.fluo.api.client.SnapshotBase; |
| import org.apache.fluo.api.client.scanner.ScannerBuilder; |
| import org.apache.fluo.api.data.Bytes; |
| import org.apache.fluo.api.data.Column; |
| import org.apache.fluo.api.data.RowColumn; |
| import org.apache.fluo.api.data.Span; |
| import org.apache.fluo.api.exceptions.AlreadySetException; |
| import org.apache.fluo.api.exceptions.CommitException; |
| import org.apache.fluo.api.exceptions.FluoException; |
| import org.apache.fluo.core.async.AsyncCommitObserver; |
| import org.apache.fluo.core.async.AsyncConditionalWriter; |
| import org.apache.fluo.core.async.AsyncTransaction; |
| import org.apache.fluo.core.async.SyncCommitObserver; |
| import org.apache.fluo.core.exceptions.AlreadyAcknowledgedException; |
| import org.apache.fluo.core.exceptions.StaleScanException; |
| import org.apache.fluo.core.impl.AsyncReader; |
| import org.apache.fluo.core.impl.scanner.ScannerBuilderImpl; |
| import org.apache.fluo.core.oracle.Stamp; |
| import org.apache.fluo.core.util.ByteUtil; |
| import org.apache.fluo.core.util.ColumnUtil; |
| import org.apache.fluo.core.util.ConditionalFlutation; |
| import org.apache.fluo.core.util.FluoCondition; |
| import org.apache.fluo.core.util.Flutation; |
| import org.apache.fluo.core.util.Hex; |
| import org.apache.fluo.core.util.SpanUtil; |
| import org.apache.fluo.core.util.UtilWaitThread; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import static org.apache.fluo.api.observer.Observer.NotificationType.STRONG; |
| import static org.apache.fluo.api.observer.Observer.NotificationType.WEAK; |
| |
| /** |
| * Transaction implementation |
| */ |
| public class TransactionImpl extends AbstractTransactionBase implements AsyncTransaction, Snapshot { |
| |
| public static final byte[] EMPTY = new byte[0]; |
| public static final Bytes EMPTY_BS = Bytes.of(EMPTY); |
| private static final Bytes DELETE = |
| Bytes.of("special delete object f804266bf94935edd45ae3e6c287b93c1814295c"); |
| private static final Bytes NTFY_VAL = |
| Bytes.of("special ntfy value ce0c523e6e4dc093be8a2736b82eca1b95f97ed4"); |
| private static final Bytes RLOCK_VAL = |
| Bytes.of("special rlock value 94da84e7796ff3b23b779805d820a33f1997cb8b"); |
| |
| // added to avoid findbugs false positive |
| private static final Supplier<Void> NULLS = () -> null; |
| |
| private static boolean isWrite(Bytes val) { |
| return val != NTFY_VAL && val != RLOCK_VAL; |
| } |
| |
| private static boolean isDelete(Bytes val) { |
| return val == DELETE; |
| } |
| |
| private static boolean isReadLock(Bytes val) { |
| return val == RLOCK_VAL; |
| } |
| |
| private enum TxStatus { |
| OPEN, COMMIT_STARTED, COMMITTED, CLOSED |
| } |
| |
| private final long startTs; |
| private final Map<Bytes, Map<Column, Bytes>> updates = new HashMap<>(); |
| private final Map<Bytes, Set<Column>> weakNotifications = new HashMap<>(); |
| private final Set<Column> observedColumns; |
| private final Environment env; |
| private final Map<Bytes, Set<Column>> columnsRead = new HashMap<>(); |
| // Tracks row columns that were observed to have had a read lock in the past. |
| private final Map<Bytes, Set<Column>> readLocksSeen = new HashMap<>(); |
| private final TxStats stats; |
| private Notification notification; |
| private Notification weakNotification; |
| private TransactorNode tnode = null; |
| private TxStatus status = TxStatus.OPEN; |
| private boolean commitAttempted = false; |
| private AsyncReader asyncReader = null; |
| |
| |
| public TransactionImpl(Environment env, Notification trigger, long startTs) { |
| Objects.requireNonNull(env, "environment cannot be null"); |
| Preconditions.checkArgument(startTs >= 0, "startTs cannot be negative"); |
| this.env = env; |
| this.stats = new TxStats(env); |
| this.startTs = startTs; |
| this.observedColumns = env.getConfiguredObservers().getObservedColumns(STRONG); |
| |
| if (trigger != null |
| && env.getConfiguredObservers().getObservedColumns(WEAK).contains(trigger.getColumn())) { |
| this.weakNotification = trigger; |
| } else { |
| this.notification = trigger; |
| } |
| |
| if (notification != null) { |
| Map<Column, Bytes> colUpdates = new HashMap<>(); |
| colUpdates.put(notification.getColumn(), NTFY_VAL); |
| updates.put(notification.getRow(), colUpdates); |
| } |
| } |
| |
| public TransactionImpl(Environment env, Notification trigger) { |
| this(env, trigger, allocateTimestamp(env).getTxTimestamp()); |
| } |
| |
| public TransactionImpl(Environment env) { |
| this(env, null, allocateTimestamp(env).getTxTimestamp()); |
| } |
| |
| public TransactionImpl(Environment env, long startTs) { |
| this(env, null, startTs); |
| } |
| |
| private static Stamp allocateTimestamp(Environment env) { |
| return env.getSharedResources().getTimestampTracker().allocateTimestamp(); |
| } |
| |
| @Override |
| public Bytes get(Bytes row, Column column) { |
| checkIfOpen(); |
| // TODO cache? precache? |
| return get(row, Collections.singleton(column)).get(column); |
| } |
| |
| @Override |
| public Map<Column, Bytes> get(Bytes row, Set<Column> columns) { |
| checkIfOpen(); |
| return getImpl(row, columns); |
| } |
| |
| @Override |
| public Map<Bytes, Map<Column, Bytes>> get(Collection<Bytes> rows, Set<Column> columns) { |
| checkIfOpen(); |
| |
| if (rows.isEmpty() || columns.isEmpty()) { |
| return Collections.emptyMap(); |
| } |
| |
| env.getSharedResources().getVisCache().validate(columns); |
| |
| ParallelSnapshotScanner pss = |
| new ParallelSnapshotScanner(rows, columns, env, startTs, stats, readLocksSeen, kve -> { |
| }); |
| |
| Map<Bytes, Map<Column, Bytes>> ret = pss.scan(); |
| |
| for (Entry<Bytes, Map<Column, Bytes>> entry : ret.entrySet()) { |
| updateColumnsRead(entry.getKey(), entry.getValue().keySet()); |
| } |
| |
| return ret; |
| } |
| |
| @Override |
| public Map<RowColumn, Bytes> get(Collection<RowColumn> rowColumns) { |
| checkIfOpen(); |
| return getImpl(rowColumns, kve -> { |
| }); |
| } |
| |
| private Map<Column, Bytes> getImpl(Bytes row, Set<Column> columns) { |
| |
| // TODO push visibility filtering to server side? |
| |
| env.getSharedResources().getVisCache().validate(columns); |
| |
| boolean shouldCopy = false; |
| |
| for (Column column : columns) { |
| if (column.isVisibilitySet()) { |
| shouldCopy = true; |
| } |
| } |
| |
| SnapshotScanner.Opts opts; |
| if (shouldCopy) { |
| HashSet<Column> cols = new HashSet<>(); |
| for (Column column : columns) { |
| if (column.isVisibilitySet()) { |
| cols.add(new Column(column.getFamily(), column.getQualifier())); |
| } else { |
| cols.add(column); |
| } |
| } |
| opts = new SnapshotScanner.Opts(Span.exact(row), columns, true); |
| } else { |
| opts = new SnapshotScanner.Opts(Span.exact(row), columns, true); |
| } |
| |
| Map<Column, Bytes> ret = new HashMap<>(); |
| Set<Column> readLockCols = null; |
| |
| for (Entry<Key, Value> kve : new SnapshotScanner(env, opts, startTs, stats)) { |
| |
| Column col = ColumnUtil.convert(kve.getKey()); |
| if (shouldCopy && !columns.contains(col)) { |
| continue; |
| } |
| |
| if (ColumnType.from(kve.getKey()) == ColumnType.RLOCK) { |
| if (readLockCols == null) { |
| readLockCols = readLocksSeen.computeIfAbsent(row, k -> new HashSet<>()); |
| } |
| readLockCols.add(col); |
| } else { |
| ret.put(col, Bytes.of(kve.getValue().get())); |
| } |
| } |
| |
| // only update columns read after successful read |
| updateColumnsRead(row, columns); |
| |
| return ret; |
| } |
| |
| private Map<RowColumn, Bytes> getImpl(Collection<RowColumn> rowColumns, |
| Consumer<Entry<Key, Value>> writeLocksSeen) { |
| if (rowColumns.isEmpty()) { |
| return Collections.emptyMap(); |
| } |
| |
| ParallelSnapshotScanner pss = |
| new ParallelSnapshotScanner(rowColumns, env, startTs, stats, readLocksSeen, writeLocksSeen); |
| |
| Map<Bytes, Map<Column, Bytes>> scan = pss.scan(); |
| Map<RowColumn, Bytes> ret = new HashMap<>(); |
| |
| for (Entry<Bytes, Map<Column, Bytes>> entry : scan.entrySet()) { |
| updateColumnsRead(entry.getKey(), entry.getValue().keySet()); |
| for (Entry<Column, Bytes> colVal : entry.getValue().entrySet()) { |
| ret.put(new RowColumn(entry.getKey(), colVal.getKey()), colVal.getValue()); |
| } |
| } |
| |
| return ret; |
| } |
| |
| @Override |
| public CompletableFuture<Bytes> getAsync(Bytes row, Column column) { |
| return getAsyncReader().get(row, column); |
| } |
| |
| @Override |
| public CompletableFuture<Bytes> getAsync(Bytes row, Column column, Bytes defaultValue) { |
| return getAsyncReader().get(row, column, defaultValue); |
| } |
| |
| @Override |
| public CompletableFuture<String> getsAsync(String row, Column column) { |
| return getAsyncReader().gets(row, column); |
| } |
| |
| @Override |
| public CompletableFuture<String> getsAsync(String row, Column column, String defaultValue) { |
| return getAsyncReader().gets(row, column, defaultValue); |
| } |
| |
| private AsyncReader getAsyncReader() { |
| if (asyncReader == null) { |
| asyncReader = new AsyncReader(this); |
| } |
| return asyncReader; |
| } |
| |
| @Override |
| public ScannerBuilder scanner() { |
| checkIfOpen(); |
| return new ScannerBuilderImpl(this); |
| } |
| |
| private void updateColumnsRead(Bytes row, Set<Column> columns) { |
| Set<Column> colsRead = columnsRead.get(row); |
| if (colsRead == null) { |
| colsRead = new HashSet<>(); |
| columnsRead.put(row, colsRead); |
| } |
| colsRead.addAll(columns); |
| } |
| |
| void setReadLock(Bytes row, Column col) { |
| checkIfOpen(); |
| Objects.requireNonNull(row); |
| Objects.requireNonNull(col); |
| |
| if (col.getFamily().equals(ColumnConstants.NOTIFY_CF)) { |
| throw new IllegalArgumentException(ColumnConstants.NOTIFY_CF + " is a reserved family"); |
| } |
| |
| env.getSharedResources().getVisCache().validate(col); |
| |
| Map<Column, Bytes> colUpdates = updates.computeIfAbsent(row, k -> new HashMap<>()); |
| Bytes curVal = colUpdates.get(col); |
| if (curVal != null && (isWrite(curVal) || isDelete(curVal))) { |
| throw new AlreadySetException("Attemped read lock after write lock " + row + " " + col); |
| } |
| |
| colUpdates.put(col, RLOCK_VAL); |
| } |
| |
| @Override |
| public SnapshotBase withReadLock() { |
| return new ReadLockSnapshot(this); |
| } |
| |
| @Override |
| public void set(Bytes row, Column col, Bytes value) throws AlreadySetException { |
| checkIfOpen(); |
| Objects.requireNonNull(row); |
| Objects.requireNonNull(col); |
| Objects.requireNonNull(value); |
| |
| if (col.getFamily().equals(ColumnConstants.NOTIFY_CF)) { |
| throw new IllegalArgumentException(ColumnConstants.NOTIFY_CF + " is a reserved family"); |
| } |
| |
| env.getSharedResources().getVisCache().validate(col); |
| |
| Map<Column, Bytes> colUpdates = updates.computeIfAbsent(row, k -> new HashMap<>()); |
| |
| Bytes curVal = colUpdates.get(col); |
| if (curVal != null && isWrite(curVal)) { |
| throw new AlreadySetException("Value already set " + row + " " + col); |
| } |
| colUpdates.put(col, value); |
| } |
| |
| @Override |
| public void setWeakNotification(Bytes row, Column col) { |
| checkIfOpen(); |
| Objects.requireNonNull(row); |
| Objects.requireNonNull(col); |
| |
| if (!env.getConfiguredObservers().getObservedColumns(WEAK).contains(col)) { |
| throw new IllegalArgumentException("Column not configured for weak notifications " + col); |
| } |
| |
| env.getSharedResources().getVisCache().validate(col); |
| |
| Set<Column> columns = weakNotifications.get(row); |
| if (columns == null) { |
| columns = new HashSet<>(); |
| weakNotifications.put(row, columns); |
| } |
| |
| columns.add(col); |
| } |
| |
| @Override |
| public void delete(Bytes row, Column col) throws AlreadySetException { |
| checkIfOpen(); |
| Objects.requireNonNull(row); |
| Objects.requireNonNull(col); |
| set(row, col, DELETE); |
| } |
| |
| private ConditionalFlutation prewrite(ConditionalFlutation cm, Bytes row, Column col, Bytes val, |
| Bytes primaryRow, Column primaryColumn, boolean isTriggerRow) { |
| IteratorSetting iterConf = new IteratorSetting(10, PrewriteIterator.class); |
| PrewriteIterator.setSnaptime(iterConf, startTs); |
| boolean isTrigger = isTriggerRow && col.equals(notification.getColumn()); |
| if (isTrigger) { |
| PrewriteIterator.enableAckCheck(iterConf, notification.getTimestamp()); |
| } |
| |
| if (isReadLock(val)) { |
| PrewriteIterator.setReadlock(iterConf); |
| } |
| |
| Condition cond = new FluoCondition(env, col).setIterators(iterConf); |
| |
| if (cm == null) { |
| cm = new ConditionalFlutation(env, row, cond); |
| } else { |
| cm.addCondition(cond); |
| } |
| |
| if (isWrite(val) && !isDelete(val)) { |
| cm.put(col, ColumnType.DATA.encode(startTs), val.toArray()); |
| } |
| |
| if (isReadLock(val)) { |
| cm.put(col, ColumnType.RLOCK.encode(ReadLockUtil.encodeTs(startTs, false)), |
| ReadLockValue.encode(primaryRow, primaryColumn, getTransactorID())); |
| } else { |
| cm.put(col, ColumnType.LOCK.encode(startTs), LockValue.encode(primaryRow, primaryColumn, |
| isWrite(val), isDelete(val), isTriggerRow, getTransactorID())); |
| } |
| |
| return cm; |
| } |
| |
| private ConditionalFlutation prewrite(Bytes row, Column col, Bytes val, Bytes primaryRow, |
| Column primaryColumn, boolean isTriggerRow) { |
| return prewrite(null, row, col, val, primaryRow, primaryColumn, isTriggerRow); |
| } |
| |
| private void prewrite(ConditionalFlutation cm, Column col, Bytes val, Bytes primaryRow, |
| Column primaryColumn, boolean isTriggerRow) { |
| prewrite(cm, null, col, val, primaryRow, primaryColumn, isTriggerRow); |
| } |
| |
| public static class CommitData { |
| ConditionalWriter cw; |
| private Bytes prow; |
| private Column pcol; |
| private Bytes pval; |
| |
| private HashSet<Bytes> acceptedRows; |
| private Map<Bytes, Set<Column>> rejected = null; |
| |
| private void addPrimaryToRejected() { |
| rejected = Collections.singletonMap(prow, Collections.singleton(pcol)); |
| } |
| |
| private void addToRejected(Bytes row, Set<Column> columns) { |
| if (rejected == null) { |
| rejected = new HashMap<>(); |
| } |
| |
| Set<Column> ret = rejected.put(row, columns); |
| if (ret != null) { |
| throw new IllegalStateException(); |
| } |
| } |
| |
| private Map<Bytes, Set<Column>> getRejected() { |
| if (rejected == null) { |
| return Collections.emptyMap(); |
| } |
| |
| return rejected; |
| } |
| |
| @Override |
| public String toString() { |
| return prow + " " + pcol + " " + pval + " " + getRejected().size(); |
| } |
| |
| public String getShortCollisionMessage() { |
| StringBuilder sb = new StringBuilder(); |
| if (!getRejected().isEmpty()) { |
| int numCollisions = 0; |
| for (Set<Column> cols : getRejected().values()) { |
| numCollisions += cols.size(); |
| } |
| |
| sb.append("Collisions("); |
| sb.append(numCollisions); |
| sb.append("):"); |
| |
| String sep = ""; |
| outer: for (Entry<Bytes, Set<Column>> entry : getRejected().entrySet()) { |
| Bytes row = entry.getKey(); |
| for (Column col : entry.getValue()) { |
| sb.append(sep); |
| sep = ", "; |
| Hex.encNonAscii(sb, row); |
| sb.append(" "); |
| Hex.encNonAscii(sb, col, " "); |
| if (sb.length() > 100) { |
| sb.append(" ..."); |
| break outer; |
| } |
| } |
| } |
| } |
| |
| return sb.toString(); |
| } |
| |
| // async stuff |
| private AsyncConditionalWriter acw; |
| private AsyncConditionalWriter bacw; |
| private AsyncCommitObserver commitObserver; |
| |
| } |
| |
| private boolean isTriggerRow(Bytes row) { |
| return notification != null && notification.getRow().equals(row); |
| } |
| |
| public boolean preCommit(CommitData cd) { |
| return preCommit(cd, null); |
| } |
| |
| @VisibleForTesting |
| public boolean preCommit(CommitData cd, RowColumn primary) { |
| |
| synchronized (this) { |
| checkIfOpen(); |
| status = TxStatus.COMMIT_STARTED; |
| commitAttempted = true; |
| } |
| |
| SyncCommitObserver sco = new SyncCommitObserver(); |
| cd = setUpBeginCommitAsync(cd, sco, primary); |
| if (cd != null) { |
| beginCommitAsyncTest(cd); |
| } |
| try { |
| sco.waitForCommit(); |
| } catch (AlreadyAcknowledgedException e) { |
| throw e; |
| } catch (CommitException e) { |
| return false; |
| } |
| return true; |
| } |
| |
| /** |
| * This function helps handle the following case |
| * |
| * <OL> |
| * <LI>TX1 locks r1 col1 |
| * <LI>TX1 fails before unlocking |
| * <LI>TX2 attempts to write r1:col1 w/o reading it |
| * </OL> |
| * |
| * <p> |
| * In this case TX2 would not roll back TX1, because it never read the column. This function |
| * attempts to handle this case if TX2 fails. Only doing this in case of failures is cheaper than |
| * trying to always read unread columns. |
| * |
| * @param cd Commit data |
| */ |
| private void readUnread(CommitData cd, Consumer<Entry<Key, Value>> writeLocksSeen) { |
| // TODO need to keep track of ranges read (not ranges passed in, but actual data read... user |
| // may not iterate over entire range |
| Collection<RowColumn> rowColumnsToRead = new ArrayList<>(); |
| |
| for (Entry<Bytes, Set<Column>> entry : cd.getRejected().entrySet()) { |
| Set<Column> rowColsRead = columnsRead.get(entry.getKey()); |
| if (rowColsRead == null) { |
| for (Column column : entry.getValue()) { |
| rowColumnsToRead.add(new RowColumn(entry.getKey(), column)); |
| } |
| } else { |
| HashSet<Column> colsToRead = new HashSet<>(entry.getValue()); |
| colsToRead.removeAll(rowColsRead); |
| if (!colsToRead.isEmpty()) { |
| for (Column column : colsToRead) { |
| rowColumnsToRead.add(new RowColumn(entry.getKey(), column)); |
| } |
| } |
| } |
| } |
| |
| getImpl(rowColumnsToRead, writeLocksSeen); |
| } |
| |
| private void checkForOrphanedReadLocks(CommitData cd, Map<Bytes, Set<Column>> locksResolved) |
| throws Exception { |
| |
| if (readLocksSeen.isEmpty()) { |
| return; |
| } |
| |
| Map<Bytes, Set<Column>> rowColsToCheck = new HashMap<>(); |
| |
| for (Entry<Bytes, Set<Column>> entry : cd.getRejected().entrySet()) { |
| |
| Set<Column> resolvedColumns = |
| locksResolved.getOrDefault(entry.getKey(), Collections.emptySet()); |
| |
| Set<Column> colsToCheck = null; |
| Set<Column> readLockCols = readLocksSeen.get(entry.getKey()); |
| if (readLockCols != null) { |
| for (Column candidate : Sets.intersection(readLockCols, entry.getValue())) { |
| if (resolvedColumns.contains(candidate)) { |
| // A write lock was seen and this is probably what caused the collision, no need to |
| // check this column for read locks. |
| continue; |
| } |
| |
| if (!isReadLock(updates.getOrDefault(entry.getKey(), Collections.emptyMap()) |
| .getOrDefault(candidate, EMPTY_BS))) { |
| if (colsToCheck == null) { |
| colsToCheck = new HashSet<>(); |
| } |
| colsToCheck.add(candidate); |
| } |
| } |
| |
| if (colsToCheck != null) { |
| rowColsToCheck.put(entry.getKey(), colsToCheck); |
| } |
| } |
| } |
| |
| if (!rowColsToCheck.isEmpty()) { |
| |
| long waitTime = SnapshotScanner.INITIAL_WAIT_TIME; |
| |
| boolean resolved = false; |
| |
| List<Entry<Key, Value>> openReadLocks = LockResolver.getOpenReadLocks(env, rowColsToCheck); |
| |
| long startTime = System.currentTimeMillis(); |
| |
| while (!resolved) { |
| resolved = LockResolver.resolveLocks(env, startTs, stats, openReadLocks, startTime); |
| if (!resolved) { |
| UtilWaitThread.sleep(waitTime); |
| stats.incrementLockWaitTime(waitTime); |
| waitTime = Math.min(SnapshotScanner.MAX_WAIT_TIME, waitTime * 2); |
| |
| openReadLocks = LockResolver.getOpenReadLocks(env, rowColsToCheck); |
| } |
| } |
| } |
| } |
| |
| private void checkForOrphanedLocks(CommitData cd) throws Exception { |
| |
| Map<Bytes, Set<Column>> writeLocksSeen = new HashMap<>(); |
| |
| readUnread(cd, kve -> { |
| Bytes row = ByteUtil.toBytes(kve.getKey().getRowData()); |
| Column col = ColumnUtil.convert(kve.getKey()); |
| writeLocksSeen.computeIfAbsent(row, k -> new HashSet<>()).add(col); |
| }); |
| |
| checkForOrphanedReadLocks(cd, writeLocksSeen); |
| } |
| |
| private boolean checkForAckCollision(ConditionalMutation cm) { |
| Bytes row = Bytes.of(cm.getRow()); |
| |
| if (isTriggerRow(row)) { |
| List<ColumnUpdate> updates = cm.getUpdates(); |
| |
| for (ColumnUpdate cu : updates) { |
| // TODO avoid create col vis object |
| Column col = new Column(Bytes.of(cu.getColumnFamily()), Bytes.of(cu.getColumnQualifier()), |
| Bytes.of(cu.getColumnVisibility())); |
| |
| if (notification.getColumn().equals(col)) { |
| // check to see if ACK exist after notification |
| Key startKey = SpanUtil.toKey(notification.getRowColumn()); |
| startKey.setTimestamp(ColumnType.ACK.first()); |
| |
| Key endKey = SpanUtil.toKey(notification.getRowColumn()); |
| endKey.setTimestamp(ColumnType.ACK.encode(notification.getTimestamp() + 1)); |
| |
| Range range = new Range(startKey, endKey); |
| |
| try (Scanner scanner = |
| env.getAccumuloClient().createScanner(env.getTable(), env.getAuthorizations())) { |
| scanner.setRange(range); |
| |
| // TODO could use iterator that stops after 1st ACK. thought of using versioning iter |
| // but |
| // it scans to ACK |
| if (scanner.iterator().hasNext()) { |
| env.getSharedResources().getBatchWriter() |
| .writeMutationAsync(notification.newDelete(env)); |
| return true; |
| } |
| } catch (TableNotFoundException e) { |
| // TODO proper exception handling |
| throw new RuntimeException(e); |
| } |
| } |
| } |
| } |
| |
| return false; |
| } |
| |
| public CommitData createCommitData() { |
| CommitData cd = new CommitData(); |
| cd.cw = env.getSharedResources().getConditionalWriter(); |
| cd.acw = env.getSharedResources().getAsyncConditionalWriter(); |
| cd.bacw = env.getSharedResources().getBulkAsyncConditionalWriter(); |
| return cd; |
| } |
| |
| @Override |
| public synchronized void commit() throws CommitException { |
| try { |
| SyncCommitObserver sco = new SyncCommitObserver(); |
| commitAsync(sco); |
| sco.waitForCommit(); |
| } finally { |
| updates.clear(); |
| weakNotification = null; |
| columnsRead.clear(); |
| } |
| } |
| |
| void deleteWeakRow() { |
| if (weakNotification != null) { |
| env.getSharedResources().getBatchWriter() |
| .writeMutation(weakNotification.newDelete(env, startTs)); |
| } |
| } |
| |
| @Override |
| public TxStats getStats() { |
| return stats; |
| } |
| |
| public long getStartTs() { |
| return startTs; |
| } |
| |
| /** |
| * Sets the transactor of this transaction |
| * |
| * @param tnode TransactorNode |
| * @return this Transaction |
| */ |
| @VisibleForTesting |
| public TransactionImpl setTransactor(TransactorNode tnode) { |
| this.tnode = tnode; |
| return this; |
| } |
| |
| /** |
| * Retrieves transactor ID by first getting/creating transactor (which is only done until |
| * necessary) |
| */ |
| private Long getTransactorID() { |
| if (tnode == null) { |
| tnode = env.getSharedResources().getTransactorNode(); |
| } |
| return tnode.getTransactorID().getLongID(); |
| } |
| |
| private synchronized void close(boolean checkForStaleScan) { |
| if (asyncReader != null) { |
| asyncReader.close(); |
| } |
| |
| if (status != TxStatus.CLOSED) { |
| status = TxStatus.CLOSED; |
| |
| if (checkForStaleScan && !commitAttempted) { |
| Stamp stamp = env.getSharedResources().getOracleClient().getStamp(); |
| if (startTs < stamp.getGcTimestamp()) { |
| throw new StaleScanException(); |
| } |
| } |
| |
| env.getSharedResources().getTimestampTracker().removeTimestamp(startTs); |
| } |
| } |
| |
| @Override |
| public void close() { |
| close(true); |
| } |
| |
| private synchronized void checkIfOpen() { |
| if (status != TxStatus.OPEN) { |
| throw new IllegalStateException("Transaction is no longer open! status = " + status); |
| } |
| } |
| |
| // CHECKSTYLE:OFF |
| @Override |
| protected void finalize() { |
| // CHECKSTYLE:ON |
| // TODO Log an error if transaction is not closed (See FLUO-486) |
| close(false); |
| } |
| |
| @Override |
| public long getStartTimestamp() { |
| return startTs; |
| } |
| |
| @Override |
| public int getSize() { |
| // TODO could calculate as items are added/set |
| int size = 0; |
| |
| for (Entry<Bytes, Map<Column, Bytes>> entry : updates.entrySet()) { |
| size += entry.getKey().length(); |
| for (Entry<Column, Bytes> entry2 : entry.getValue().entrySet()) { |
| Column c = entry2.getKey(); |
| size += c.getFamily().length(); |
| size += c.getQualifier().length(); |
| size += c.getVisibility().length(); |
| size += entry2.getValue().length(); |
| } |
| } |
| |
| for (Entry<Bytes, Set<Column>> entry : columnsRead.entrySet()) { |
| size += entry.getKey().length(); |
| for (Column c : entry.getValue()) { |
| size += c.getFamily().length(); |
| size += c.getQualifier().length(); |
| size += c.getVisibility().length(); |
| } |
| } |
| |
| return size; |
| } |
| |
| abstract class CommitStep { |
| private CommitStep nextStep; |
| |
| // the boolean indicates if the operation was successful. |
| abstract CompletableFuture<Boolean> getMainOp(CommitData cd); |
| |
| // create and run this op in the event that the main op was a failure |
| abstract CompletableFuture<Void> getFailureOp(CommitData cd); |
| |
| // set the next step to run if this step is successful |
| CommitStep andThen(CommitStep next) { |
| this.nextStep = next; |
| return next; |
| } |
| |
| |
| CompletableFuture<Void> compose(CommitData cd) { |
| return getMainOp(cd).thenComposeAsync(successful -> { |
| if (successful) { |
| if (nextStep != null) { |
| return nextStep.compose(cd); |
| } else { |
| return CompletableFuture.completedFuture(null); |
| } |
| } else { |
| return getFailureOp(cd); |
| } |
| }, env.getSharedResources().getAsyncCommitExecutor()); |
| } |
| |
| } |
| |
| abstract class ConditionalStep extends CommitStep { |
| |
| public abstract Collection<ConditionalMutation> createMutations(CommitData cd); |
| |
| public abstract Iterator<Result> handleUnknown(CommitData cd, Iterator<Result> results) |
| throws Exception; |
| |
| public abstract boolean processResults(CommitData cd, Iterator<Result> results) |
| throws Exception; |
| |
| public AsyncConditionalWriter getACW(CommitData cd) { |
| return cd.acw; |
| } |
| |
| @Override |
| CompletableFuture<Boolean> getMainOp(CommitData cd) { |
| // TODO not sure threading is correct |
| Executor ace = env.getSharedResources().getAsyncCommitExecutor(); |
| return getACW(cd).apply(createMutations(cd)).thenCompose(results -> { |
| // ugh icky that this is an iterator, forces copy to inspect.. could refactor async CW to |
| // return collection |
| ArrayList<Result> resultsList = new ArrayList<>(); |
| Iterators.addAll(resultsList, results); |
| boolean containsUknown = false; |
| for (Result result : resultsList) { |
| try { |
| containsUknown |= result.getStatus() == Status.UNKNOWN; |
| } catch (Exception e) { |
| throw new CompletionException(e); |
| } |
| } |
| if (containsUknown) { |
| // process unknown in sync executor |
| Executor se = env.getSharedResources().getSyncCommitExecutor(); |
| return CompletableFuture.supplyAsync(() -> { |
| try { |
| return handleUnknown(cd, resultsList.iterator()); |
| } catch (Exception e) { |
| throw new CompletionException(e); |
| } |
| }, se); |
| } else { |
| return CompletableFuture.completedFuture(resultsList.iterator()); |
| } |
| }).thenApplyAsync(results -> { |
| try { |
| return processResults(cd, results); |
| } catch (Exception e) { |
| throw new CompletionException(e); |
| } |
| }, ace); |
| } |
| |
| |
| } |
| |
| class LockPrimaryStep extends ConditionalStep { |
| |
| @Override |
| public Collection<ConditionalMutation> createMutations(CommitData cd) { |
| return Collections |
| .singleton(prewrite(cd.prow, cd.pcol, cd.pval, cd.prow, cd.pcol, isTriggerRow(cd.prow))); |
| } |
| |
| @Override |
| public Iterator<Result> handleUnknown(CommitData cd, Iterator<Result> results) |
| throws Exception { |
| |
| Result result = Iterators.getOnlyElement(results); |
| Status mutationStatus = result.getStatus(); |
| // TODO convert this code to async |
| while (mutationStatus == Status.UNKNOWN) { |
| TxInfo txInfo = TxInfo.getTransactionInfo(env, cd.prow, cd.pcol, startTs); |
| |
| switch (txInfo.status) { |
| case LOCKED: |
| return Collections |
| .singleton( |
| new Result(Status.ACCEPTED, result.getMutation(), result.getTabletServer())) |
| .iterator(); |
| case ROLLED_BACK: |
| return Collections |
| .singleton( |
| new Result(Status.REJECTED, result.getMutation(), result.getTabletServer())) |
| .iterator(); |
| case UNKNOWN: |
| // TODO async |
| Result newResult = cd.cw.write(result.getMutation()); |
| mutationStatus = newResult.getStatus(); |
| if (mutationStatus != Status.UNKNOWN) { |
| return Collections.singleton(newResult).iterator(); |
| } |
| // TODO handle case were data other tx has lock |
| break; |
| case COMMITTED: |
| default: |
| throw new IllegalStateException( |
| "unexpected tx state " + txInfo.status + " " + cd.prow + " " + cd.pcol); |
| |
| } |
| } |
| |
| // TODO |
| throw new IllegalStateException(); |
| } |
| |
| @Override |
| public boolean processResults(CommitData cd, Iterator<Result> results) throws Exception { |
| Result result = Iterators.getOnlyElement(results); |
| return result.getStatus() == Status.ACCEPTED; |
| } |
| |
| @Override |
| CompletableFuture<Void> getFailureOp(CommitData cd) { |
| // TODO can this be simplified by pushing some code to the superclass? |
| return CompletableFuture.supplyAsync(() -> { |
| final ConditionalMutation pcm = Iterables.getOnlyElement(createMutations(cd)); |
| |
| cd.addPrimaryToRejected(); |
| getStats().setRejected(cd.getRejected()); |
| // TODO do async |
| try { |
| checkForOrphanedLocks(cd); |
| } catch (Exception e) { |
| throw new CompletionException(e); |
| } |
| if (checkForAckCollision(pcm)) { |
| cd.commitObserver.alreadyAcknowledged(); |
| } else { |
| cd.commitObserver.commitFailed(cd.getShortCollisionMessage()); |
| } |
| |
| return null; |
| }, env.getSharedResources().getSyncCommitExecutor()); |
| } |
| |
| } |
| |
| class LockOtherStep extends ConditionalStep { |
| |
| @Override |
| public AsyncConditionalWriter getACW(CommitData cd) { |
| return cd.bacw; |
| } |
| |
| |
| @Override |
| public Collection<ConditionalMutation> createMutations(CommitData cd) { |
| |
| ArrayList<ConditionalMutation> mutations = new ArrayList<>(); |
| |
| for (Entry<Bytes, Map<Column, Bytes>> rowUpdates : updates.entrySet()) { |
| ConditionalFlutation cm = null; |
| |
| for (Entry<Column, Bytes> colUpdates : rowUpdates.getValue().entrySet()) { |
| if (cm == null) { |
| cm = prewrite(rowUpdates.getKey(), colUpdates.getKey(), colUpdates.getValue(), cd.prow, |
| cd.pcol, false); |
| } else { |
| prewrite(cm, colUpdates.getKey(), colUpdates.getValue(), cd.prow, cd.pcol, false); |
| } |
| } |
| |
| mutations.add(cm); |
| } |
| |
| cd.acceptedRows = new HashSet<>(); |
| |
| return mutations; |
| } |
| |
| @Override |
| public Iterator<Result> handleUnknown(CommitData cd, Iterator<Result> results) { |
| // TODO this step does not currently handle unknown |
| return results; |
| } |
| |
| @Override |
| public boolean processResults(CommitData cd, Iterator<Result> results) throws Exception { |
| |
| while (results.hasNext()) { |
| Result result = results.next(); |
| // TODO handle unknown? |
| Bytes row = Bytes.of(result.getMutation().getRow()); |
| if (result.getStatus() == Status.ACCEPTED) { |
| cd.acceptedRows.add(row); |
| } else { |
| cd.addToRejected(row, updates.get(row).keySet()); |
| } |
| } |
| |
| return cd.getRejected().isEmpty(); |
| } |
| |
| @Override |
| CompletableFuture<Void> getFailureOp(CommitData cd) { |
| return CompletableFuture.supplyAsync(() -> { |
| getStats().setRejected(cd.getRejected()); |
| try { |
| // Does this need to be async? |
| checkForOrphanedLocks(cd); |
| } catch (Exception e) { |
| throw new CompletionException(e); |
| } |
| return null; |
| }, env.getSharedResources().getSyncCommitExecutor()).thenCompose(v -> rollbackLocks(cd)); |
| } |
| } |
| |
| abstract class BatchWriterStep extends CommitStep { |
| public abstract Collection<Mutation> createMutations(CommitData cd); |
| |
| @Override |
| CompletableFuture<Boolean> getMainOp(CommitData cd) { |
| return env.getSharedResources().getBatchWriter() |
| .writeMutationsAsyncFuture(createMutations(cd)).thenApply(v -> true); |
| } |
| |
| @Override |
| CompletableFuture<Void> getFailureOp(CommitData cd) { |
| throw new IllegalStateException("Failure not expected"); |
| } |
| } |
| |
| |
| |
| private CompletableFuture<Void> rollbackLocks(CommitData cd) { |
| CommitStep firstStep = new RollbackOtherLocks(); |
| firstStep.andThen(new RollbackPrimaryLock()); |
| |
| return firstStep.compose(cd) |
| .thenRun(() -> cd.commitObserver.commitFailed(cd.getShortCollisionMessage())); |
| |
| } |
| |
| |
| class RollbackOtherLocks extends BatchWriterStep { |
| |
| @Override |
| public Collection<Mutation> createMutations(CommitData cd) { |
| // roll back locks |
| |
| // TODO let rollback be done lazily? this makes GC more difficult |
| |
| Flutation m; |
| |
| ArrayList<Mutation> mutations = new ArrayList<>(cd.acceptedRows.size()); |
| for (Bytes row : cd.acceptedRows) { |
| m = new Flutation(env, row); |
| for (Entry<Column, Bytes> entry : updates.get(row).entrySet()) { |
| if (isReadLock(entry.getValue())) { |
| m.put(entry.getKey(), ColumnType.RLOCK.encode(ReadLockUtil.encodeTs(startTs, true)), |
| DelReadLockValue.encodeRollback()); |
| } else { |
| m.put(entry.getKey(), ColumnType.DEL_LOCK.encode(startTs), |
| DelLockValue.encodeRollback(false, true)); |
| } |
| } |
| mutations.add(m); |
| } |
| |
| return mutations; |
| } |
| } |
| |
| class RollbackPrimaryLock extends BatchWriterStep { |
| |
| @Override |
| public Collection<Mutation> createMutations(CommitData cd) { |
| // mark transaction as complete for garbage collection purposes |
| Flutation m = new Flutation(env, cd.prow); |
| |
| m.put(cd.pcol, ColumnType.DEL_LOCK.encode(startTs), |
| DelLockValue.encodeRollback(startTs, true, true)); |
| m.put(cd.pcol, ColumnType.TX_DONE.encode(startTs), EMPTY); |
| |
| return Collections.singletonList(m); |
| } |
| } |
| |
| class CommittedTestStep extends CommitStep { |
| @Override |
| CompletableFuture<Boolean> getMainOp(CommitData cd) { |
| cd.commitObserver.committed(); |
| return CompletableFuture.completedFuture(true); |
| } |
| |
| @Override |
| CompletableFuture<Void> getFailureOp(CommitData cd) { |
| throw new IllegalStateException("Failure not expected"); |
| } |
| } |
| |
| @VisibleForTesting |
| public boolean commitPrimaryColumn(CommitData cd, Stamp commitStamp) { |
| |
| SyncCommitObserver sco = new SyncCommitObserver(); |
| cd.commitObserver = sco; |
| try { |
| CommitStep firstStep = new GetCommitStampStepTest(commitStamp); |
| |
| firstStep.andThen(new WriteNotificationsStep()).andThen(new CommitPrimaryStep()) |
| .andThen(new CommittedTestStep()); |
| |
| firstStep.compose(cd).exceptionally(throwable -> { |
| setFailed(cd, throwable); |
| return null; |
| }); |
| sco.waitForCommit(); |
| } catch (CommitException e) { |
| return false; |
| } catch (Exception e) { |
| throw new FluoException(e); |
| } |
| return true; |
| } |
| |
| class GetCommitStampStep extends CommitStep { |
| protected CompletableFuture<Stamp> getStampOp() { |
| return env.getSharedResources().getOracleClient().getStampAsync(); |
| } |
| |
| @Override |
| CompletableFuture<Boolean> getMainOp(CommitData cd) { |
| |
| return getStampOp().thenApply(commitStamp -> { |
| if (startTs < commitStamp.getGcTimestamp()) { |
| return false; |
| } else { |
| getStats().setCommitTs(commitStamp.getTxTimestamp()); |
| return true; |
| } |
| }); |
| } |
| |
| @Override |
| CompletableFuture<Void> getFailureOp(CommitData cd) { |
| return rollbackLocks(cd); |
| } |
| |
| } |
| |
| class GetCommitStampStepTest extends GetCommitStampStep { |
| private final Stamp testStamp; |
| |
| public GetCommitStampStepTest(Stamp testStamp) { |
| this.testStamp = testStamp; |
| } |
| |
| @Override |
| protected CompletableFuture<Stamp> getStampOp() { |
| return CompletableFuture.completedFuture(testStamp); |
| } |
| |
| } |
| |
| class WriteNotificationsStep extends BatchWriterStep { |
| |
| @Override |
| public Collection<Mutation> createMutations(CommitData cd) { |
| long commitTs = getStats().getCommitTs(); |
| HashMap<Bytes, Mutation> mutations = new HashMap<>(); |
| |
| if (observedColumns.contains(cd.pcol) && isWrite(cd.pval) && !isDelete(cd.pval)) { |
| Flutation m = new Flutation(env, cd.prow); |
| Notification.put(env, m, cd.pcol, commitTs); |
| mutations.put(cd.prow, m); |
| } |
| |
| for (Entry<Bytes, Map<Column, Bytes>> rowUpdates : updates.entrySet()) { |
| |
| for (Entry<Column, Bytes> colUpdates : rowUpdates.getValue().entrySet()) { |
| if (observedColumns.contains(colUpdates.getKey())) { |
| Bytes val = colUpdates.getValue(); |
| if (isWrite(val) && !isDelete(val)) { |
| Mutation m = mutations.get(rowUpdates.getKey()); |
| if (m == null) { |
| m = new Flutation(env, rowUpdates.getKey()); |
| mutations.put(rowUpdates.getKey(), m); |
| } |
| Notification.put(env, m, colUpdates.getKey(), commitTs); |
| } |
| } |
| } |
| } |
| |
| for (Entry<Bytes, Set<Column>> entry : weakNotifications.entrySet()) { |
| Mutation m = mutations.get(entry.getKey()); |
| if (m == null) { |
| m = new Flutation(env, entry.getKey()); |
| mutations.put(entry.getKey(), m); |
| } |
| for (Column col : entry.getValue()) { |
| Notification.put(env, m, col, commitTs); |
| } |
| } |
| return mutations.values(); |
| } |
| |
| } |
| |
| class CommitPrimaryStep extends ConditionalStep { |
| |
| @Override |
| public Collection<ConditionalMutation> createMutations(CommitData cd) { |
| long commitTs = getStats().getCommitTs(); |
| IteratorSetting iterConf = new IteratorSetting(10, PrewriteIterator.class); |
| PrewriteIterator.setSnaptime(iterConf, startTs); |
| boolean isTrigger = isTriggerRow(cd.prow) && cd.pcol.equals(notification.getColumn()); |
| |
| Condition lockCheck = |
| new FluoCondition(env, cd.pcol).setIterators(iterConf).setValue(LockValue.encode(cd.prow, |
| cd.pcol, isWrite(cd.pval), isDelete(cd.pval), isTrigger, getTransactorID())); |
| final ConditionalMutation delLockMutation = new ConditionalFlutation(env, cd.prow, lockCheck); |
| |
| ColumnUtil.commitColumn(env, isTrigger, true, cd.pcol, isWrite(cd.pval), isDelete(cd.pval), |
| isReadLock(cd.pval), startTs, commitTs, observedColumns, delLockMutation); |
| |
| return Collections.singletonList(delLockMutation); |
| } |
| |
| @Override |
| public Iterator<Result> handleUnknown(CommitData cd, Iterator<Result> results) |
| throws Exception { |
| // the code for handing this is synchronous and needs to be handled in another thread pool |
| // TODO - how do we do the above without return a CF? |
| long commitTs = getStats().getCommitTs(); |
| Result result = Iterators.getOnlyElement(results); |
| Status ms = result.getStatus(); |
| |
| while (ms == Status.UNKNOWN) { |
| |
| // TODO async |
| TxInfo txInfo = TxInfo.getTransactionInfo(env, cd.prow, cd.pcol, startTs); |
| |
| switch (txInfo.status) { |
| case COMMITTED: |
| if (txInfo.commitTs != commitTs) { |
| throw new IllegalStateException( |
| cd.prow + " " + cd.pcol + " " + txInfo.commitTs + "!=" + commitTs); |
| } |
| ms = Status.ACCEPTED; |
| break; |
| case LOCKED: |
| // TODO async |
| ConditionalMutation delLockMutation = result.getMutation(); |
| ms = cd.cw.write(delLockMutation).getStatus(); |
| break; |
| default: |
| ms = Status.REJECTED; |
| } |
| } |
| Result newResult = new Result(ms, result.getMutation(), result.getTabletServer()); |
| return Collections.singletonList(newResult).iterator(); |
| } |
| |
| @Override |
| public boolean processResults(CommitData cd, Iterator<Result> results) throws Exception { |
| Result result = Iterators.getOnlyElement(results); |
| return result.getStatus() == Status.ACCEPTED; |
| } |
| |
| @Override |
| CompletableFuture<Void> getFailureOp(CommitData cd) { |
| cd.commitObserver.commitFailed(cd.getShortCollisionMessage()); |
| return CompletableFuture.completedFuture(NULLS.get()); |
| } |
| |
| } |
| |
| @VisibleForTesting |
| public boolean finishCommit(CommitData cd, Stamp commitStamp) { |
| cd.commitObserver = new SyncCommitObserver(); |
| |
| getStats().setCommitTs(commitStamp.getTxTimestamp()); |
| |
| CommitStep firstStep = new DeleteLocksStep(); |
| firstStep.andThen(new FinishCommitStep()); |
| firstStep.compose(cd).exceptionally(throwable -> { |
| System.err.println("Unexpected exception in finish commit test method : "); |
| throwable.printStackTrace(); |
| return null; |
| }); |
| |
| return true; |
| } |
| |
| |
| class DeleteLocksStep extends BatchWriterStep { |
| |
| @Override |
| public Collection<Mutation> createMutations(CommitData cd) { |
| long commitTs = getStats().getCommitTs(); |
| ArrayList<Mutation> mutations = new ArrayList<>(updates.size() + 1); |
| for (Entry<Bytes, Map<Column, Bytes>> rowUpdates : updates.entrySet()) { |
| Flutation m = new Flutation(env, rowUpdates.getKey()); |
| boolean isTriggerRow = isTriggerRow(rowUpdates.getKey()); |
| for (Entry<Column, Bytes> colUpdates : rowUpdates.getValue().entrySet()) { |
| ColumnUtil.commitColumn(env, |
| isTriggerRow && colUpdates.getKey().equals(notification.getColumn()), false, |
| colUpdates.getKey(), isWrite(colUpdates.getValue()), isDelete(colUpdates.getValue()), |
| isReadLock(colUpdates.getValue()), startTs, commitTs, observedColumns, m); |
| } |
| |
| mutations.add(m); |
| } |
| |
| return mutations; |
| } |
| |
| } |
| |
| class FinishCommitStep extends BatchWriterStep { |
| |
| @Override |
| CompletableFuture<Boolean> getMainOp(CommitData cd) { |
| return super.getMainOp(cd).thenApply(b -> { |
| Preconditions.checkArgument(b); |
| cd.commitObserver.committed(); |
| return true; |
| }); |
| } |
| |
| @Override |
| public Collection<Mutation> createMutations(CommitData cd) { |
| long commitTs = getStats().getCommitTs(); |
| ArrayList<Mutation> afterFlushMutations = new ArrayList<>(2); |
| |
| Flutation m = new Flutation(env, cd.prow); |
| // mark transaction as complete for garbage collection purposes |
| m.put(cd.pcol, ColumnType.TX_DONE.encode(commitTs), EMPTY); |
| afterFlushMutations.add(m); |
| |
| if (weakNotification != null) { |
| afterFlushMutations.add(weakNotification.newDelete(env, startTs)); |
| } |
| |
| if (notification != null) { |
| afterFlushMutations.add(notification.newDelete(env, startTs)); |
| } |
| |
| return afterFlushMutations; |
| } |
| |
| } |
| |
| @Override |
| public synchronized void commitAsync(AsyncCommitObserver commitCallback) { |
| |
| checkIfOpen(); |
| status = TxStatus.COMMIT_STARTED; |
| commitAttempted = true; |
| |
| try { |
| CommitData cd = createCommitData(); |
| cd = setUpBeginCommitAsync(cd, commitCallback, null); |
| if (cd != null) { |
| beginCommitAsync(cd); |
| } |
| } catch (Exception e) { |
| e.printStackTrace(); |
| commitCallback.failed(e); |
| } |
| } |
| |
| private CommitData setUpBeginCommitAsync(CommitData cd, AsyncCommitObserver commitCallback, |
| RowColumn primary) { |
| if (updates.isEmpty()) { |
| // TODO do async |
| deleteWeakRow(); |
| commitCallback.committed(); |
| return null; |
| } |
| |
| for (Map<Column, Bytes> cols : updates.values()) { |
| stats.incrementEntriesSet(cols.size()); |
| } |
| |
| Bytes primRow = null; |
| Column primCol = null; |
| |
| if (primary != null) { |
| primRow = primary.getRow(); |
| primCol = primary.getColumn(); |
| if (notification != null && !primary.equals(notification.getRowColumn())) { |
| throw new IllegalArgumentException("Primary must be notification"); |
| } |
| } else if (notification != null) { |
| primRow = notification.getRow(); |
| primCol = notification.getColumn(); |
| } else { |
| |
| outer: for (Entry<Bytes, Map<Column, Bytes>> entry : updates.entrySet()) { |
| for (Entry<Column, Bytes> entry2 : entry.getValue().entrySet()) { |
| if (!isReadLock(entry2.getValue())) { |
| primRow = entry.getKey(); |
| primCol = entry2.getKey(); |
| break outer; |
| } |
| } |
| } |
| |
| if (primRow == null) { |
| // there are only read locks, so nothing to write |
| deleteWeakRow(); |
| commitCallback.committed(); |
| return null; |
| } |
| } |
| |
| // get a primary column |
| cd.prow = primRow; |
| Map<Column, Bytes> colSet = updates.get(cd.prow); |
| cd.pcol = primCol; |
| cd.pval = colSet.remove(primCol); |
| if (colSet.isEmpty()) { |
| updates.remove(cd.prow); |
| } |
| |
| cd.commitObserver = commitCallback; |
| |
| return cd; |
| } |
| |
| private void setFailed(CommitData cd, Throwable throwable) { |
| try { |
| cd.commitObserver.failed(throwable); |
| } catch (RuntimeException e) { |
| Logger log = LoggerFactory.getLogger(TransactionImpl.class); |
| log.error("Failed to set tx failure (startTs=" + startTs + ") cause ", e); |
| log.error("Failed to set tx failure (startTs=" + startTs + ") lost throwable ", throwable); |
| } |
| } |
| |
| private void beginCommitAsync(CommitData cd) { |
| |
| // Notification are written between GetCommitStampStep and CommitPrimaryStep for the following |
| // reasons : |
| // * At this point all columns are locked, this guarantees that anything triggering as a |
| // result of this transaction will see all of this transactions changes. |
| // * The transaction is not yet committed. If the process dies at this point whatever |
| // was running this transaction should rerun and recreate all of the notifications. |
| // The next transactions will rerun because this transaction will have to be rolled back. |
| // * If notifications are written in the 2nd phase of commit, then when the 2nd phase |
| // partially succeeds notifications may never be written. Because in the case of failure |
| // notifications would not be written until a column is read and it may never be read. |
| // See https://github.com/fluo-io/fluo/issues/642 |
| // |
| // Its very important the notifications which trigger an observer are deleted after the 2nd |
| // phase of commit finishes. |
| |
| CommitStep firstStep = new LockPrimaryStep(); |
| |
| firstStep.andThen(new LockOtherStep()).andThen(new GetCommitStampStep()) |
| .andThen(new WriteNotificationsStep()).andThen(new CommitPrimaryStep()) |
| .andThen(new DeleteLocksStep()).andThen(new FinishCommitStep()); |
| |
| firstStep.compose(cd).exceptionally(throwable -> { |
| setFailed(cd, throwable); |
| return null; |
| }); |
| } |
| |
| private void beginCommitAsyncTest(CommitData cd) { |
| |
| CommitStep firstStep = new LockPrimaryStep(); |
| |
| firstStep.andThen(new LockOtherStep()).andThen(new CommittedTestStep()); |
| |
| firstStep.compose(cd).exceptionally(throwable -> { |
| setFailed(cd, throwable); |
| return null; |
| }); |
| } |
| |
| public SnapshotScanner newSnapshotScanner(Span span, Collection<Column> columns) { |
| return new SnapshotScanner(env, new SnapshotScanner.Opts(span, columns, false), startTs, stats); |
| } |
| } |
