http://git-wip-us.apache.org/repos/asf/incubator-tephra/blob/9c693743/tephra-core/src/main/java/org/apache/tephra/TransactionManager.java ---------------------------------------------------------------------- diff --git a/tephra-core/src/main/java/org/apache/tephra/TransactionManager.java b/tephra-core/src/main/java/org/apache/tephra/TransactionManager.java new file mode 100644 index 0000000..cfefc83 --- /dev/null +++ b/tephra-core/src/main/java/org/apache/tephra/TransactionManager.java @@ -0,0 +1,1398 @@ +/* + * 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.tephra; + +import com.google.common.base.Objects; +import com.google.common.base.Preconditions; +import com.google.common.base.Stopwatch; +import com.google.common.base.Throwables; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.common.collect.Sets; +import com.google.common.util.concurrent.AbstractService; +import com.google.inject.Inject; +import it.unimi.dsi.fastutil.longs.LongArrayList; +import org.apache.hadoop.conf.Configuration; +import org.apache.tephra.metrics.DefaultMetricsCollector; +import org.apache.tephra.metrics.MetricsCollector; +import org.apache.tephra.persist.NoOpTransactionStateStorage; +import org.apache.tephra.persist.TransactionEdit; +import org.apache.tephra.persist.TransactionLog; +import org.apache.tephra.persist.TransactionLogReader; +import org.apache.tephra.persist.TransactionSnapshot; +import org.apache.tephra.persist.TransactionStateStorage; +import org.apache.tephra.snapshot.SnapshotCodecProvider; +import org.apache.tephra.util.TxUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.io.OutputStream; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.NavigableMap; +import java.util.Set; +import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; + +/** + * This is the central place to manage all active transactions in the system. + * + * A transaction consists of + * <ul> + * <li>A write pointer: This is the version used for all writes of that transaction.</li> + * <li>A read pointer: All reads under the transaction use this as an upper bound for the version.</li> + * <li>A set of excluded versions: These are the write versions of other transactions that must be excluded from + * reads, because those transactions are still in progress, or they failed but couldn't be properly rolled back.</li> + * </ul> + * To use the transaction system, a client must follow this sequence of steps: + * <ol> + * <li>Request a new transaction.</li> + * <li>Use the transaction to read and write datasets. Datasets are encouraged to cache the writes of the + * transaction in memory, to reduce the cost of rollback in case the transaction fails. </li> + * <li>Check whether the transaction has conflicts. For this, the set of change keys are submitted via canCommit(), + * and the transaction manager verifies that none of these keys are in conflict with other transactions that + * committed since the start of this transaction.</li> + * <li>If the transaction has conflicts: + * <ol> + * <li>Roll back the changes in every dataset that was changed. This can happen in-memory if the + * changes were cached.</li> + * <li>Abort the transaction to remove it from the active transactions.</li> + * </ol> + * <li>If the transaction has no conflicts:</li> + * <ol> + * <li>Persist all datasets changes to storage.</li> + * <li>Commit the transaction. This will repeat the conflict detection, because more overlapping transactions + * may have committed since the first conflict check.</li> + * <li>If the transaction has conflicts:</li> + * <ol> + * <li>Roll back the changes in every dataset that was changed. This is more expensive because + * changes must be undone in persistent storage.</li> + * <li>Abort the transaction to remove it from the active transactions.</li> + * </ol> + * </ol> + * </ol> + * Transactions may be short or long-running. A short transaction is started with a timeout, and if it is not + * committed before that timeout, it is invalidated and excluded from future reads. A long-running transaction has + * no timeout and will remain active until it is committed or aborted. Long transactions are typically used in + * map/reduce jobs and can produce enormous amounts of changes. Therefore, long transactions do not participate in + * conflict detection (they would almost always have conflicts). We also assume that the changes of long transactions + * are not tracked, and therefore cannot be rolled back. Hence, when a long transaction is aborted, it remains in the + * list of excluded transactions to make its writes invisible. + */ +public class TransactionManager extends AbstractService { + // todo: optimize heavily + + private static final Logger LOG = LoggerFactory.getLogger(TransactionManager.class); + + // poll every 1 second to check whether a snapshot is needed + private static final long SNAPSHOT_POLL_INTERVAL = 1000L; + + //poll every 10 second to emit metrics + private static final long METRICS_POLL_INTERVAL = 10000L; + + private static final long[] NO_INVALID_TX = { }; + + // Transactions that are in progress, with their info. + private final NavigableMap<Long, InProgressTx> inProgress = new ConcurrentSkipListMap<Long, InProgressTx>(); + + // the list of transactions that are invalid (not properly committed/aborted, or timed out) + // TODO: explain usage of two arrays + private final LongArrayList invalid = new LongArrayList(); + private long[] invalidArray = NO_INVALID_TX; + + // todo: use moving array instead (use Long2ObjectMap<byte[]> in fastutil) + // todo: should this be consolidated with inProgress? + // commit time next writePointer -> changes made by this tx + private final NavigableMap<Long, Set<ChangeId>> committedChangeSets = + new ConcurrentSkipListMap<Long, Set<ChangeId>>(); + // not committed yet + private final Map<Long, Set<ChangeId>> committingChangeSets = Maps.newConcurrentMap(); + + private long readPointer; + private long lastWritePointer; + private MetricsCollector txMetricsCollector; + + private final TransactionStateStorage persistor; + + private final int cleanupInterval; + private final int defaultTimeout; + private final int defaultLongTimeout; + private DaemonThreadExecutor cleanupThread = null; + + private volatile TransactionLog currentLog; + + // timestamp of the last completed snapshot + private long lastSnapshotTime; + // frequency in millis to perform snapshots + private final long snapshotFrequencyInSeconds; + // number of most recent snapshots to retain + private final int snapshotRetainCount; + private DaemonThreadExecutor snapshotThread; + private DaemonThreadExecutor metricsThread; + + // lock guarding change of the current transaction log + private final ReentrantReadWriteLock logLock = new ReentrantReadWriteLock(); + private final Lock logReadLock = logLock.readLock(); + private final Lock logWriteLock = logLock.writeLock(); + + // fudge factor (in milliseconds) used when interpreting transactions as LONG based on expiration + // TODO: REMOVE WITH txnBackwardsCompatCheck() + private final long longTimeoutTolerance; + + public TransactionManager(Configuration config) { + this(config, new NoOpTransactionStateStorage(new SnapshotCodecProvider(config)), new DefaultMetricsCollector()); + } + + @Inject + public TransactionManager(Configuration conf, @Nonnull TransactionStateStorage persistor, + MetricsCollector txMetricsCollector) { + this.persistor = persistor; + cleanupInterval = conf.getInt(TxConstants.Manager.CFG_TX_CLEANUP_INTERVAL, + TxConstants.Manager.DEFAULT_TX_CLEANUP_INTERVAL); + defaultTimeout = conf.getInt(TxConstants.Manager.CFG_TX_TIMEOUT, + TxConstants.Manager.DEFAULT_TX_TIMEOUT); + defaultLongTimeout = conf.getInt(TxConstants.Manager.CFG_TX_LONG_TIMEOUT, + TxConstants.Manager.DEFAULT_TX_LONG_TIMEOUT); + snapshotFrequencyInSeconds = conf.getLong(TxConstants.Manager.CFG_TX_SNAPSHOT_INTERVAL, + TxConstants.Manager.DEFAULT_TX_SNAPSHOT_INTERVAL); + // must always keep at least 1 snapshot + snapshotRetainCount = Math.max(conf.getInt(TxConstants.Manager.CFG_TX_SNAPSHOT_RETAIN, + TxConstants.Manager.DEFAULT_TX_SNAPSHOT_RETAIN), 1); + + // intentionally not using a constant, as this config should not be exposed + // TODO: REMOVE WITH txnBackwardsCompatCheck() + longTimeoutTolerance = conf.getLong("data.tx.long.timeout.tolerance", 10000); + + // + this.txMetricsCollector = txMetricsCollector; + this.txMetricsCollector.configure(conf); + clear(); + } + + private void clear() { + invalid.clear(); + invalidArray = NO_INVALID_TX; + inProgress.clear(); + committedChangeSets.clear(); + committingChangeSets.clear(); + lastWritePointer = 0; + readPointer = 0; + lastSnapshotTime = 0; + } + + private boolean isStopping() { + return State.STOPPING.equals(state()); + } + + @Override + public synchronized void doStart() { + LOG.info("Starting transaction manager."); + txMetricsCollector.start(); + // start up the persistor + persistor.startAndWait(); + try { + persistor.setupStorage(); + } catch (IOException e) { + Throwables.propagate(e); + } + // establish defaults in case there is no persistence + clear(); + // attempt to recover state from last run + recoverState(); + // start the periodic cleanup thread + startCleanupThread(); + startSnapshotThread(); + startMetricsThread(); + // initialize the WAL if we did not force a snapshot in recoverState() + initLog(); + // initialize next write pointer if needed + if (lastWritePointer == 0) { + lastWritePointer = getNextWritePointer(); + readPointer = lastWritePointer; + } + + notifyStarted(); + } + + private void initLog() { + if (currentLog == null) { + try { + currentLog = persistor.createLog(System.currentTimeMillis()); + } catch (IOException ioe) { + throw Throwables.propagate(ioe); + } + } + } + + private void startCleanupThread() { + if (cleanupInterval <= 0 || defaultTimeout <= 0) { + return; + } + LOG.info("Starting periodic timed-out transaction cleanup every " + cleanupInterval + + " seconds with default timeout of " + defaultTimeout + " seconds."); + this.cleanupThread = new DaemonThreadExecutor("tx-clean-timeout") { + @Override + public void doRun() { + cleanupTimedOutTransactions(); + } + + @Override + public long getSleepMillis() { + return cleanupInterval * 1000; + } + }; + cleanupThread.start(); + } + + private void startSnapshotThread() { + if (snapshotFrequencyInSeconds > 0) { + LOG.info("Starting periodic snapshot thread, frequency = " + snapshotFrequencyInSeconds + + " seconds, location = " + persistor.getLocation()); + this.snapshotThread = new DaemonThreadExecutor("tx-snapshot") { + @Override + public void doRun() { + long currentTime = System.currentTimeMillis(); + if (lastSnapshotTime < (currentTime - snapshotFrequencyInSeconds * 1000)) { + try { + doSnapshot(false); + } catch (IOException ioe) { + LOG.error("Periodic snapshot failed!", ioe); + } + } + } + + @Override + protected void onShutdown() { + // perform a final snapshot + try { + LOG.info("Writing final snapshot prior to shutdown"); + doSnapshot(true); + } catch (IOException ioe) { + LOG.error("Failed performing final snapshot on shutdown", ioe); + } + } + + @Override + public long getSleepMillis() { + return SNAPSHOT_POLL_INTERVAL; + } + }; + snapshotThread.start(); + } + } + + // Emits Transaction Data structures size as metrics + private void startMetricsThread() { + LOG.info("Starting periodic Metrics Emitter thread, frequency = " + METRICS_POLL_INTERVAL); + this.metricsThread = new DaemonThreadExecutor("tx-metrics") { + @Override + public void doRun() { + txMetricsCollector.gauge("committing.size", committingChangeSets.size()); + txMetricsCollector.gauge("committed.size", committedChangeSets.size()); + txMetricsCollector.gauge("inprogress.size", inProgress.size()); + txMetricsCollector.gauge("invalid.size", invalidArray.length); + } + + @Override + protected void onShutdown() { + // perform a final metrics emit + txMetricsCollector.gauge("committing.size", committingChangeSets.size()); + txMetricsCollector.gauge("committed.size", committedChangeSets.size()); + txMetricsCollector.gauge("inprogress.size", inProgress.size()); + txMetricsCollector.gauge("invalid.size", invalidArray.length); + } + + @Override + public long getSleepMillis() { + return METRICS_POLL_INTERVAL; + } + }; + metricsThread.start(); + } + + private void cleanupTimedOutTransactions() { + List<TransactionEdit> invalidEdits = null; + this.logReadLock.lock(); + try { + synchronized (this) { + if (!isRunning()) { + return; + } + + long currentTime = System.currentTimeMillis(); + List<Long> timedOut = Lists.newArrayList(); + for (Map.Entry<Long, InProgressTx> tx : inProgress.entrySet()) { + long expiration = tx.getValue().getExpiration(); + if (expiration >= 0L && currentTime > expiration) { + // timed out, remember tx id (can't remove while iterating over entries) + timedOut.add(tx.getKey()); + LOG.info("Tx invalid list: added tx {} because of timeout", tx.getKey()); + } else if (expiration < 0) { + LOG.warn("Transaction {} has negative expiration time {}. Likely cause is the transaction was not " + + "migrated correctly, this transaction will be expired immediately", + tx.getKey(), expiration); + timedOut.add(tx.getKey()); + } + } + if (!timedOut.isEmpty()) { + invalidEdits = Lists.newArrayListWithCapacity(timedOut.size()); + invalid.addAll(timedOut); + for (long tx : timedOut) { + committingChangeSets.remove(tx); + inProgress.remove(tx); + invalidEdits.add(TransactionEdit.createInvalid(tx)); + } + + // todo: find a more efficient way to keep this sorted. Could it just be an array? + Collections.sort(invalid); + invalidArray = invalid.toLongArray(); + LOG.info("Invalidated {} transactions due to timeout.", timedOut.size()); + } + } + if (invalidEdits != null) { + appendToLog(invalidEdits); + } + } finally { + this.logReadLock.unlock(); + } + } + + public synchronized TransactionSnapshot getSnapshot() throws IOException { + TransactionSnapshot snapshot = null; + if (!isRunning() && !isStopping()) { + return null; + } + + long now = System.currentTimeMillis(); + // avoid duplicate snapshots at same timestamp + if (now == lastSnapshotTime || (currentLog != null && now == currentLog.getTimestamp())) { + try { + TimeUnit.MILLISECONDS.sleep(1); + } catch (InterruptedException ie) { } + } + // copy in memory state + snapshot = getCurrentState(); + + LOG.debug("Starting snapshot of transaction state with timestamp {}", snapshot.getTimestamp()); + LOG.debug("Returning snapshot of state: " + snapshot); + return snapshot; + } + + /** + * Take a snapshot of the transaction state and serialize it into the given output stream. + * @return whether a snapshot was taken. + */ + public boolean takeSnapshot(OutputStream out) throws IOException { + TransactionSnapshot snapshot = getSnapshot(); + if (snapshot != null) { + persistor.writeSnapshot(out, snapshot); + return true; + } else { + return false; + } + } + + private void doSnapshot(boolean closing) throws IOException { + long snapshotTime = 0L; + TransactionSnapshot snapshot = null; + TransactionLog oldLog = null; + try { + this.logWriteLock.lock(); + try { + synchronized (this) { + snapshot = getSnapshot(); + if (snapshot == null && !closing) { + return; + } + if (snapshot != null) { + snapshotTime = snapshot.getTimestamp(); + } + + // roll WAL + oldLog = currentLog; + if (!closing) { + currentLog = persistor.createLog(snapshot.getTimestamp()); + } + } + // there may not be an existing log on startup + if (oldLog != null) { + oldLog.close(); + } + } finally { + this.logWriteLock.unlock(); + } + + // save snapshot + if (snapshot != null) { + persistor.writeSnapshot(snapshot); + lastSnapshotTime = snapshotTime; + + // clean any obsoleted snapshots and WALs + long oldestRetainedTimestamp = persistor.deleteOldSnapshots(snapshotRetainCount); + persistor.deleteLogsOlderThan(oldestRetainedTimestamp); + } + } catch (IOException ioe) { + abortService("Snapshot (timestamp " + snapshotTime + ") failed due to: " + ioe.getMessage(), ioe); + } + } + + public synchronized TransactionSnapshot getCurrentState() { + return TransactionSnapshot.copyFrom(System.currentTimeMillis(), readPointer, lastWritePointer, + invalid, inProgress, committingChangeSets, committedChangeSets); + } + + public synchronized void recoverState() { + try { + TransactionSnapshot lastSnapshot = persistor.getLatestSnapshot(); + // if we failed before a snapshot could complete, we might not have one to restore + if (lastSnapshot != null) { + restoreSnapshot(lastSnapshot); + } + // replay any WALs since the last snapshot + Collection<TransactionLog> logs = persistor.getLogsSince(lastSnapshotTime); + if (logs != null) { + replayLogs(logs); + } + } catch (IOException e) { + LOG.error("Unable to read back transaction state:", e); + throw Throwables.propagate(e); + } + } + + /** + * Restore the initial in-memory transaction state from a snapshot. + */ + private void restoreSnapshot(TransactionSnapshot snapshot) { + LOG.info("Restoring transaction state from snapshot at " + snapshot.getTimestamp()); + Preconditions.checkState(lastSnapshotTime == 0, "lastSnapshotTime has been set!"); + Preconditions.checkState(readPointer == 0, "readPointer has been set!"); + Preconditions.checkState(lastWritePointer == 0, "lastWritePointer has been set!"); + Preconditions.checkState(invalid.isEmpty(), "invalid list should be empty!"); + Preconditions.checkState(inProgress.isEmpty(), "inProgress map should be empty!"); + Preconditions.checkState(committingChangeSets.isEmpty(), "committingChangeSets should be empty!"); + Preconditions.checkState(committedChangeSets.isEmpty(), "committedChangeSets should be empty!"); + LOG.info("Restoring snapshot of state: " + snapshot); + + lastSnapshotTime = snapshot.getTimestamp(); + readPointer = snapshot.getReadPointer(); + lastWritePointer = snapshot.getWritePointer(); + invalid.addAll(snapshot.getInvalid()); + inProgress.putAll(txnBackwardsCompatCheck(defaultLongTimeout, longTimeoutTolerance, snapshot.getInProgress())); + committingChangeSets.putAll(snapshot.getCommittingChangeSets()); + committedChangeSets.putAll(snapshot.getCommittedChangeSets()); + } + + /** + * Check if in-progress transactions need to be migrated to have expiration time and type, if so do the migration. + * This is required for backwards compatibility, when long running transactions were represented + * with expiration time -1. This can be removed when we stop supporting SnapshotCodec version 1. + */ + public static Map<Long, InProgressTx> txnBackwardsCompatCheck(int defaultLongTimeout, long longTimeoutTolerance, + Map<Long, InProgressTx> inProgress) { + for (Map.Entry<Long, InProgressTx> entry : inProgress.entrySet()) { + long writePointer = entry.getKey(); + long expiration = entry.getValue().getExpiration(); + // LONG transactions will either have a negative expiration or expiration set to the long timeout + // use a fudge factor on the expiration check, since expiraton is set based on system time, not the write pointer + if (entry.getValue().getType() == null && + (expiration < 0 || + (getTxExpirationFromWritePointer(writePointer, defaultLongTimeout) - expiration + < longTimeoutTolerance))) { + // handle null expiration + long newExpiration = getTxExpirationFromWritePointer(writePointer, defaultLongTimeout); + InProgressTx compatTx = + new InProgressTx(entry.getValue().getVisibilityUpperBound(), newExpiration, TransactionType.LONG, + entry.getValue().getCheckpointWritePointers()); + entry.setValue(compatTx); + } else if (entry.getValue().getType() == null) { + InProgressTx compatTx = + new InProgressTx(entry.getValue().getVisibilityUpperBound(), entry.getValue().getExpiration(), + TransactionType.SHORT, entry.getValue().getCheckpointWritePointers()); + entry.setValue(compatTx); + } + } + return inProgress; + } + + /** + * Resets the state of the transaction manager. + */ + public void resetState() { + this.logWriteLock.lock(); + try { + // Take a snapshot before resetting the state, for debugging purposes + doSnapshot(false); + // Clear the state + clear(); + // Take another snapshot: if no snapshot is taken after clearing the state + // and the manager is restarted, we will recover from the snapshot taken + // before resetting the state, which would be really bad + // This call will also init a new WAL + doSnapshot(false); + } catch (IOException e) { + LOG.error("Snapshot failed when resetting state!", e); + e.printStackTrace(); + } finally { + this.logWriteLock.unlock(); + } + } + + /** + * Replay all logged edits from the given transaction logs. + */ + private void replayLogs(Collection<TransactionLog> logs) { + for (TransactionLog log : logs) { + LOG.info("Replaying edits from transaction log " + log.getName()); + int editCnt = 0; + try { + TransactionLogReader reader = log.getReader(); + // reader may be null in the case of an empty file + if (reader == null) { + continue; + } + TransactionEdit edit = null; + while ((edit = reader.next()) != null) { + editCnt++; + switch (edit.getState()) { + case INPROGRESS: + long expiration = edit.getExpiration(); + TransactionType type = edit.getType(); + // Check if transaction needs to be migrated to have expiration and type. Previous version of + // long running transactions were represented with expiration time as -1. + // This can be removed when we stop supporting TransactionEditCodecV2. + if (expiration < 0) { + expiration = getTxExpirationFromWritePointer(edit.getWritePointer(), defaultLongTimeout); + type = TransactionType.LONG; + } else if (type == null) { + type = TransactionType.SHORT; + } + addInProgressAndAdvance(edit.getWritePointer(), edit.getVisibilityUpperBound(), + expiration, type); + break; + case COMMITTING: + addCommittingChangeSet(edit.getWritePointer(), edit.getChanges()); + break; + case COMMITTED: + // TODO: need to reconcile usage of transaction id v/s write pointer TEPHRA-140 + long transactionId = edit.getWritePointer(); + long[] checkpointPointers = edit.getCheckpointPointers(); + long writePointer = checkpointPointers == null || checkpointPointers.length == 0 ? + transactionId : checkpointPointers[checkpointPointers.length - 1]; + doCommit(transactionId, writePointer, edit.getChanges(), + edit.getCommitPointer(), edit.getCanCommit()); + break; + case INVALID: + doInvalidate(edit.getWritePointer()); + break; + case ABORTED: + type = edit.getType(); + // Check if transaction edit needs to be migrated to have type. Previous versions of + // ABORTED edits did not contain type. + // This can be removed when we stop supporting TransactionEditCodecV2. + if (type == null) { + InProgressTx inProgressTx = inProgress.get(edit.getWritePointer()); + if (inProgressTx != null) { + type = inProgressTx.getType(); + } else { + // If transaction is not in-progress, then it has either been already aborted or invalidated. + // We cannot determine the transaction's state based on current information, to be safe invalidate it. + LOG.warn("Invalidating transaction {} as it's type cannot be determined during replay", + edit.getWritePointer()); + doInvalidate(edit.getWritePointer()); + break; + } + } + doAbort(edit.getWritePointer(), edit.getCheckpointPointers(), type); + break; + case TRUNCATE_INVALID_TX: + if (edit.getTruncateInvalidTxTime() != 0) { + doTruncateInvalidTxBefore(edit.getTruncateInvalidTxTime()); + } else { + doTruncateInvalidTx(edit.getTruncateInvalidTx()); + } + break; + case CHECKPOINT: + doCheckpoint(edit.getWritePointer(), edit.getParentWritePointer()); + break; + default: + // unknown type! + throw new IllegalArgumentException("Invalid state for WAL entry: " + edit.getState()); + } + } + } catch (IOException ioe) { + throw Throwables.propagate(ioe); + } catch (InvalidTruncateTimeException e) { + throw Throwables.propagate(e); + } + LOG.info("Read " + editCnt + " edits from log " + log.getName()); + } + } + + @Override + public void doStop() { + Stopwatch timer = new Stopwatch().start(); + LOG.info("Shutting down gracefully..."); + // signal the cleanup thread to stop + if (cleanupThread != null) { + cleanupThread.shutdown(); + try { + cleanupThread.join(30000L); + } catch (InterruptedException ie) { + LOG.warn("Interrupted waiting for cleanup thread to stop"); + Thread.currentThread().interrupt(); + } + } + if (metricsThread != null) { + metricsThread.shutdown(); + try { + metricsThread.join(30000L); + } catch (InterruptedException ie) { + LOG.warn("Interrupted waiting for cleanup thread to stop"); + Thread.currentThread().interrupt(); + } + } + if (snapshotThread != null) { + // this will trigger a final snapshot on stop + snapshotThread.shutdown(); + try { + snapshotThread.join(30000L); + } catch (InterruptedException ie) { + LOG.warn("Interrupted waiting for snapshot thread to stop"); + Thread.currentThread().interrupt(); + } + } + + persistor.stopAndWait(); + txMetricsCollector.stop(); + timer.stop(); + LOG.info("Took " + timer + " to stop"); + notifyStopped(); + } + + /** + * Immediately shuts down the service, without going through the normal close process. + * @param message A message describing the source of the failure. + * @param error Any exception that caused the failure. + */ + private void abortService(String message, Throwable error) { + if (isRunning()) { + LOG.error("Aborting transaction manager due to: " + message, error); + notifyFailed(error); + } + } + + private void ensureAvailable() { + Preconditions.checkState(isRunning(), "Transaction Manager is not running."); + } + + /** + * Start a short transaction with the default timeout. + */ + public Transaction startShort() { + return startShort(defaultTimeout); + } + + /** + * Start a short transaction with a given timeout. + * @param timeoutInSeconds the time out period in seconds. + */ + public Transaction startShort(int timeoutInSeconds) { + Preconditions.checkArgument(timeoutInSeconds > 0, "timeout must be positive but is %s", timeoutInSeconds); + txMetricsCollector.rate("start.short"); + Stopwatch timer = new Stopwatch().start(); + long expiration = getTxExpiration(timeoutInSeconds); + Transaction tx = startTx(expiration, TransactionType.SHORT); + txMetricsCollector.histogram("start.short.latency", (int) timer.elapsedMillis()); + return tx; + } + + private static long getTxExpiration(long timeoutInSeconds) { + long currentTime = System.currentTimeMillis(); + return currentTime + TimeUnit.SECONDS.toMillis(timeoutInSeconds); + } + + public static long getTxExpirationFromWritePointer(long writePointer, long timeoutInSeconds) { + return writePointer / TxConstants.MAX_TX_PER_MS + TimeUnit.SECONDS.toMillis(timeoutInSeconds); + } + + private long getNextWritePointer() { + // We want to align tx ids with current time. We assume that tx ids are sequential, but not less than + // System.currentTimeMillis() * MAX_TX_PER_MS. + return Math.max(lastWritePointer + 1, System.currentTimeMillis() * TxConstants.MAX_TX_PER_MS); + } + + /** + * Start a long transaction. Long transactions and do not participate in conflict detection. Also, aborting a long + * transaction moves it to the invalid list because we assume that its writes cannot be rolled back. + */ + public Transaction startLong() { + txMetricsCollector.rate("start.long"); + Stopwatch timer = new Stopwatch().start(); + long expiration = getTxExpiration(defaultLongTimeout); + Transaction tx = startTx(expiration, TransactionType.LONG); + txMetricsCollector.histogram("start.long.latency", (int) timer.elapsedMillis()); + return tx; + } + + private Transaction startTx(long expiration, TransactionType type) { + Transaction tx = null; + long txid; + // guard against changes to the transaction log while processing + this.logReadLock.lock(); + try { + synchronized (this) { + ensureAvailable(); + txid = getNextWritePointer(); + tx = createTransaction(txid, type); + addInProgressAndAdvance(tx.getTransactionId(), tx.getVisibilityUpperBound(), expiration, type); + } + // appending to WAL out of global lock for concurrent performance + // we should still be able to arrive at the same state even if log entries are out of order + appendToLog(TransactionEdit.createStarted(tx.getTransactionId(), tx.getVisibilityUpperBound(), expiration, type)); + } finally { + this.logReadLock.unlock(); + } + return tx; + } + + private void addInProgressAndAdvance(long writePointer, long visibilityUpperBound, + long expiration, TransactionType type) { + inProgress.put(writePointer, new InProgressTx(visibilityUpperBound, expiration, type)); + advanceWritePointer(writePointer); + } + + private void advanceWritePointer(long writePointer) { + // don't move the write pointer back if we have out of order transaction log entries + if (writePointer > lastWritePointer) { + lastWritePointer = writePointer; + } + } + + public boolean canCommit(Transaction tx, Collection<byte[]> changeIds) throws TransactionNotInProgressException { + txMetricsCollector.rate("canCommit"); + Stopwatch timer = new Stopwatch().start(); + if (inProgress.get(tx.getTransactionId()) == null) { + // invalid transaction, either this has timed out and moved to invalid, or something else is wrong. + if (invalid.contains(tx.getTransactionId())) { + throw new TransactionNotInProgressException( + String.format("canCommit() is called for transaction %d that is not in progress (it is known to be invalid)", + tx.getTransactionId())); + } else { + throw new TransactionNotInProgressException( + String.format("canCommit() is called for transaction %d that is not in progress", tx.getTransactionId())); + } + } + + Set<ChangeId> set = Sets.newHashSetWithExpectedSize(changeIds.size()); + for (byte[] change : changeIds) { + set.add(new ChangeId(change)); + } + + if (hasConflicts(tx, set)) { + return false; + } + // guard against changes to the transaction log while processing + this.logReadLock.lock(); + try { + synchronized (this) { + ensureAvailable(); + addCommittingChangeSet(tx.getTransactionId(), set); + } + appendToLog(TransactionEdit.createCommitting(tx.getTransactionId(), set)); + } finally { + this.logReadLock.unlock(); + } + txMetricsCollector.histogram("canCommit.latency", (int) timer.elapsedMillis()); + return true; + } + + private void addCommittingChangeSet(long writePointer, Set<ChangeId> changes) { + committingChangeSets.put(writePointer, changes); + } + + public boolean commit(Transaction tx) throws TransactionNotInProgressException { + txMetricsCollector.rate("commit"); + Stopwatch timer = new Stopwatch().start(); + Set<ChangeId> changeSet = null; + boolean addToCommitted = true; + long commitPointer; + // guard against changes to the transaction log while processing + this.logReadLock.lock(); + try { + synchronized (this) { + ensureAvailable(); + // we record commits at the first not-yet assigned transaction id to simplify clearing out change sets that + // are no longer visible by any in-progress transactions + commitPointer = lastWritePointer + 1; + if (inProgress.get(tx.getTransactionId()) == null) { + // invalid transaction, either this has timed out and moved to invalid, or something else is wrong. + if (invalid.contains(tx.getTransactionId())) { + throw new TransactionNotInProgressException( + String.format("canCommit() is called for transaction %d that is not in progress " + + "(it is known to be invalid)", tx.getTransactionId())); + } else { + throw new TransactionNotInProgressException( + String.format("canCommit() is called for transaction %d that is not in progress", tx.getTransactionId())); + } + } + + // these should be atomic + // NOTE: whether we succeed or not we don't need to keep changes in committing state: same tx cannot + // be attempted to commit twice + changeSet = committingChangeSets.remove(tx.getTransactionId()); + + if (changeSet != null) { + // double-checking if there are conflicts: someone may have committed since canCommit check + if (hasConflicts(tx, changeSet)) { + return false; + } + } else { + // no changes + addToCommitted = false; + } + doCommit(tx.getTransactionId(), tx.getWritePointer(), changeSet, commitPointer, addToCommitted); + } + appendToLog(TransactionEdit.createCommitted(tx.getTransactionId(), changeSet, commitPointer, addToCommitted)); + } finally { + this.logReadLock.unlock(); + } + txMetricsCollector.histogram("commit.latency", (int) timer.elapsedMillis()); + return true; + } + + private void doCommit(long transactionId, long writePointer, Set<ChangeId> changes, long commitPointer, + boolean addToCommitted) { + // In case this method is called when loading a previous WAL, we need to remove the tx from these sets + committingChangeSets.remove(transactionId); + if (addToCommitted && !changes.isEmpty()) { + // No need to add empty changes to the committed change sets, they will never trigger any conflict + + // Record the committed change set with the next writePointer as the commit time. + // NOTE: we use current next writePointer as key for the map, hence we may have multiple txs changesets to be + // stored under one key + Set<ChangeId> changeIds = committedChangeSets.get(commitPointer); + if (changeIds != null) { + // NOTE: we modify the new set to prevent concurrent modification exception, as other threads (e.g. in + // canCommit) use it unguarded + changes.addAll(changeIds); + } + committedChangeSets.put(commitPointer, changes); + } + // remove from in-progress set, so that it does not get excluded in the future + InProgressTx previous = inProgress.remove(transactionId); + if (previous == null) { + // tx was not in progress! perhaps it timed out and is invalid? try to remove it there. + if (invalid.rem(transactionId)) { + invalidArray = invalid.toLongArray(); + LOG.info("Tx invalid list: removed committed tx {}", transactionId); + } + } + // moving read pointer + moveReadPointerIfNeeded(writePointer); + + // All committed change sets that are smaller than the earliest started transaction can be removed. + // here we ignore transactions that have no timeout, they are long-running and don't participate in + // conflict detection. + // TODO: for efficiency, can we do this once per-log in replayLogs instead of once per edit? + committedChangeSets.headMap(TxUtils.getFirstShortInProgress(inProgress)).clear(); + } + + public void abort(Transaction tx) { + // guard against changes to the transaction log while processing + txMetricsCollector.rate("abort"); + Stopwatch timer = new Stopwatch().start(); + this.logReadLock.lock(); + try { + synchronized (this) { + ensureAvailable(); + doAbort(tx.getTransactionId(), tx.getCheckpointWritePointers(), tx.getType()); + } + appendToLog(TransactionEdit.createAborted(tx.getTransactionId(), tx.getType(), tx.getCheckpointWritePointers())); + txMetricsCollector.histogram("abort.latency", (int) timer.elapsedMillis()); + } finally { + this.logReadLock.unlock(); + } + } + + private void doAbort(long writePointer, long[] checkpointWritePointers, TransactionType type) { + committingChangeSets.remove(writePointer); + + if (type == TransactionType.LONG) { + // Long running transactions cannot be aborted as their change sets are not saved, + // and hence the changes cannot be rolled back. Invalidate the long running transaction instead. + doInvalidate(writePointer); + return; + } + + // makes tx visible (assumes that all operations were rolled back) + // remove from in-progress set, so that it does not get excluded in the future + InProgressTx removed = inProgress.remove(writePointer); + if (removed == null) { + // tx was not in progress! perhaps it timed out and is invalid? try to remove it there. + if (invalid.rem(writePointer)) { + // remove any invalidated checkpoint pointers + // this will only be present if the parent write pointer was also invalidated + if (checkpointWritePointers != null) { + for (int i = 0; i < checkpointWritePointers.length; i++) { + invalid.rem(checkpointWritePointers[i]); + } + } + invalidArray = invalid.toLongArray(); + LOG.info("Tx invalid list: removed aborted tx {}", writePointer); + // removed a tx from excludes: must move read pointer + moveReadPointerIfNeeded(writePointer); + } + } else { + // removed a tx from excludes: must move read pointer + moveReadPointerIfNeeded(writePointer); + } + } + + public boolean invalidate(long tx) { + // guard against changes to the transaction log while processing + txMetricsCollector.rate("invalidate"); + Stopwatch timer = new Stopwatch().start(); + this.logReadLock.lock(); + try { + boolean success; + synchronized (this) { + ensureAvailable(); + success = doInvalidate(tx); + } + appendToLog(TransactionEdit.createInvalid(tx)); + txMetricsCollector.histogram("invalidate.latency", (int) timer.elapsedMillis()); + return success; + } finally { + this.logReadLock.unlock(); + } + } + + private boolean doInvalidate(long writePointer) { + Set<ChangeId> previousChangeSet = committingChangeSets.remove(writePointer); + // remove from in-progress set, so that it does not get excluded in the future + InProgressTx previous = inProgress.remove(writePointer); + // This check is to prevent from invalidating committed transactions + if (previous != null || previousChangeSet != null) { + // add tx to invalids + invalid.add(writePointer); + if (previous == null) { + LOG.debug("Invalidating tx {} in committing change sets but not in-progress", writePointer); + } else { + // invalidate any checkpoint write pointers + LongArrayList childWritePointers = previous.getCheckpointWritePointers(); + if (childWritePointers != null) { + for (int i = 0; i < childWritePointers.size(); i++) { + invalid.add(childWritePointers.get(i)); + } + } + } + LOG.info("Tx invalid list: added tx {} because of invalidate", writePointer); + // todo: find a more efficient way to keep this sorted. Could it just be an array? + Collections.sort(invalid); + invalidArray = invalid.toLongArray(); + if (previous != null && !previous.isLongRunning()) { + // tx was short-running: must move read pointer + moveReadPointerIfNeeded(writePointer); + } + return true; + } + return false; + } + + /** + * Removes the given transaction ids from the invalid list. + * @param invalidTxIds transaction ids + * @return true if invalid list got changed, false otherwise + */ + public boolean truncateInvalidTx(Set<Long> invalidTxIds) { + // guard against changes to the transaction log while processing + txMetricsCollector.rate("truncateInvalidTx"); + Stopwatch timer = new Stopwatch().start(); + this.logReadLock.lock(); + try { + boolean success; + synchronized (this) { + ensureAvailable(); + success = doTruncateInvalidTx(invalidTxIds); + } + appendToLog(TransactionEdit.createTruncateInvalidTx(invalidTxIds)); + txMetricsCollector.histogram("truncateInvalidTx.latency", (int) timer.elapsedMillis()); + return success; + } finally { + this.logReadLock.unlock(); + } + } + + private boolean doTruncateInvalidTx(Set<Long> invalidTxIds) { + LOG.info("Removing tx ids {} from invalid list", invalidTxIds); + boolean success = invalid.removeAll(invalidTxIds); + if (success) { + invalidArray = invalid.toLongArray(); + } + return success; + } + + /** + * Removes all transaction ids started before the given time from invalid list. + * @param time time in milliseconds + * @return true if invalid list got changed, false otherwise + * @throws InvalidTruncateTimeException if there are any in-progress transactions started before given time + */ + public boolean truncateInvalidTxBefore(long time) throws InvalidTruncateTimeException { + // guard against changes to the transaction log while processing + txMetricsCollector.rate("truncateInvalidTxBefore"); + Stopwatch timer = new Stopwatch().start(); + this.logReadLock.lock(); + try { + boolean success; + synchronized (this) { + ensureAvailable(); + success = doTruncateInvalidTxBefore(time); + } + appendToLog(TransactionEdit.createTruncateInvalidTxBefore(time)); + txMetricsCollector.histogram("truncateInvalidTxBefore.latency", (int) timer.elapsedMillis()); + return success; + } finally { + this.logReadLock.unlock(); + } + } + + private boolean doTruncateInvalidTxBefore(long time) throws InvalidTruncateTimeException { + LOG.info("Removing tx ids before {} from invalid list", time); + long truncateWp = time * TxConstants.MAX_TX_PER_MS; + // Check if there any in-progress transactions started earlier than truncate time + if (inProgress.lowerKey(truncateWp) != null) { + throw new InvalidTruncateTimeException("Transactions started earlier than " + time + " are in-progress"); + } + + // Find all invalid transactions earlier than truncateWp + Set<Long> toTruncate = Sets.newHashSet(); + for (long wp : invalid) { + // invalid list is sorted, hence can stop as soon as we reach a wp >= truncateWp + if (wp >= truncateWp) { + break; + } + toTruncate.add(wp); + } + return doTruncateInvalidTx(toTruncate); + } + + public Transaction checkpoint(Transaction originalTx) throws TransactionNotInProgressException { + txMetricsCollector.rate("checkpoint"); + Stopwatch timer = new Stopwatch().start(); + + Transaction checkpointedTx = null; + long txId = originalTx.getTransactionId(); + long newWritePointer = 0; + // guard against changes to the transaction log while processing + this.logReadLock.lock(); + try { + synchronized (this) { + ensureAvailable(); + // check that the parent tx is in progress + InProgressTx parentTx = inProgress.get(txId); + if (parentTx == null) { + if (invalid.contains(txId)) { + throw new TransactionNotInProgressException( + String.format("Transaction %d is not in progress because it was invalidated", txId)); + } else { + throw new TransactionNotInProgressException( + String.format("Transaction %d is not in progress", txId)); + } + } + newWritePointer = getNextWritePointer(); + doCheckpoint(newWritePointer, txId); + // create a new transaction with the same read snapshot, plus the additional checkpoint write pointer + // the same read snapshot is maintained to + checkpointedTx = new Transaction(originalTx, newWritePointer, + parentTx.getCheckpointWritePointers().toLongArray()); + } + // appending to WAL out of global lock for concurrent performance + // we should still be able to arrive at the same state even if log entries are out of order + appendToLog(TransactionEdit.createCheckpoint(newWritePointer, txId)); + } finally { + this.logReadLock.unlock(); + } + txMetricsCollector.histogram("checkpoint.latency", (int) timer.elapsedMillis()); + + return checkpointedTx; + } + + private void doCheckpoint(long newWritePointer, long parentWritePointer) { + InProgressTx existingTx = inProgress.get(parentWritePointer); + existingTx.addCheckpointWritePointer(newWritePointer); + advanceWritePointer(newWritePointer); + } + + // hack for exposing important metric + public int getExcludedListSize() { + return invalid.size() + inProgress.size(); + } + + /** + * @return the size of invalid list + */ + public int getInvalidSize() { + return this.invalid.size(); + } + + int getCommittedSize() { + return this.committedChangeSets.size(); + } + + private boolean hasConflicts(Transaction tx, Set<ChangeId> changeIds) { + if (changeIds.isEmpty()) { + return false; + } + + for (Map.Entry<Long, Set<ChangeId>> changeSet : committedChangeSets.entrySet()) { + // If commit time is greater than tx read-pointer, + // basically not visible but committed means "tx committed after given tx was started" + if (changeSet.getKey() > tx.getTransactionId()) { + if (overlap(changeSet.getValue(), changeIds)) { + return true; + } + } + } + return false; + } + + private boolean overlap(Set<ChangeId> a, Set<ChangeId> b) { + // iterate over the smaller set, and check for every element in the other set + if (a.size() > b.size()) { + for (ChangeId change : b) { + if (a.contains(change)) { + return true; + } + } + } else { + for (ChangeId change : a) { + if (b.contains(change)) { + return true; + } + } + } + return false; + } + + private void moveReadPointerIfNeeded(long committedWritePointer) { + if (committedWritePointer > readPointer) { + readPointer = committedWritePointer; + } + } + + /** + * Creates a new Transaction. This method only get called from start transaction, which is already + * synchronized. + */ + private Transaction createTransaction(long writePointer, TransactionType type) { + // For holding the first in progress short transaction Id (with timeout >= 0). + long firstShortTx = Transaction.NO_TX_IN_PROGRESS; + LongArrayList inProgressIds = new LongArrayList(inProgress.size()); + for (Map.Entry<Long, InProgressTx> entry : inProgress.entrySet()) { + long txId = entry.getKey(); + inProgressIds.add(txId); + // add any checkpointed write pointers to the in-progress list + LongArrayList childIds = entry.getValue().getCheckpointWritePointers(); + if (childIds != null) { + for (int i = 0; i < childIds.size(); i++) { + inProgressIds.add(childIds.get(i)); + } + } + if (firstShortTx == Transaction.NO_TX_IN_PROGRESS && !entry.getValue().isLongRunning()) { + firstShortTx = txId; + } + } + + return new Transaction(readPointer, writePointer, invalidArray, inProgressIds.toLongArray(), firstShortTx, type); + } + + private void appendToLog(TransactionEdit edit) { + try { + Stopwatch timer = new Stopwatch().start(); + currentLog.append(edit); + txMetricsCollector.rate("wal.append.count"); + txMetricsCollector.histogram("wal.append.latency", (int) timer.elapsedMillis()); + } catch (IOException ioe) { + abortService("Error appending to transaction log", ioe); + } + } + + private void appendToLog(List<TransactionEdit> edits) { + try { + Stopwatch timer = new Stopwatch().start(); + currentLog.append(edits); + txMetricsCollector.rate("wal.append.count", edits.size()); + txMetricsCollector.histogram("wal.append.latency", (int) timer.elapsedMillis()); + } catch (IOException ioe) { + abortService("Error appending to transaction log", ioe); + } + } + + /** + * Called from the tx service every 10 seconds. + * This hack is needed because current metrics system is not flexible when it comes to adding new metrics. + */ + public void logStatistics() { + LOG.info("Transaction Statistics: write pointer = " + lastWritePointer + + ", invalid = " + invalid.size() + + ", in progress = " + inProgress.size() + + ", committing = " + committingChangeSets.size() + + ", committed = " + committedChangeSets.size()); + } + + private abstract static class DaemonThreadExecutor extends Thread { + private AtomicBoolean stopped = new AtomicBoolean(false); + + public DaemonThreadExecutor(String name) { + super(name); + setDaemon(true); + } + + public void run() { + try { + while (!isInterrupted() && !stopped.get()) { + doRun(); + synchronized (stopped) { + stopped.wait(getSleepMillis()); + } + } + } catch (InterruptedException ie) { + LOG.info("Interrupted thread " + getName()); + } + // perform any final cleanup + onShutdown(); + LOG.info("Exiting thread " + getName()); + } + + public abstract void doRun(); + + protected abstract long getSleepMillis(); + + protected void onShutdown() { + } + + public void shutdown() { + if (stopped.compareAndSet(false, true)) { + synchronized (stopped) { + stopped.notifyAll(); + } + } + } + } + + /** + * Represents some of the info on in-progress tx + */ + public static final class InProgressTx { + /** the oldest in progress tx at the time of this tx start */ + private final long visibilityUpperBound; + private final long expiration; + private final TransactionType type; + private LongArrayList checkpointWritePointers = new LongArrayList(); + + public InProgressTx(long visibilityUpperBound, long expiration, TransactionType type) { + this(visibilityUpperBound, expiration, type, new LongArrayList()); + } + + public InProgressTx(long visibilityUpperBound, long expiration, TransactionType type, + LongArrayList checkpointWritePointers) { + this.visibilityUpperBound = visibilityUpperBound; + this.expiration = expiration; + this.type = type; + this.checkpointWritePointers = checkpointWritePointers; + } + + // For backwards compatibility when long running txns were represented with -1 expiration + @Deprecated + public InProgressTx(long visibilityUpperBound, long expiration) { + this(visibilityUpperBound, expiration, null); + } + + public long getVisibilityUpperBound() { + return visibilityUpperBound; + } + + public long getExpiration() { + return expiration; + } + + @Nullable + public TransactionType getType() { + return type; + } + + public boolean isLongRunning() { + if (type == null) { + // for backwards compatibility when long running txns were represented with -1 expiration + return expiration == -1; + } + return type == TransactionType.LONG; + } + + public void addCheckpointWritePointer(long checkpointWritePointer) { + checkpointWritePointers.add(checkpointWritePointer); + } + + public LongArrayList getCheckpointWritePointers() { + return checkpointWritePointers; + } + + @Override + public boolean equals(Object o) { + if (o == null || !(o instanceof InProgressTx)) { + return false; + } + + if (this == o) { + return true; + } + + InProgressTx other = (InProgressTx) o; + return Objects.equal(visibilityUpperBound, other.getVisibilityUpperBound()) && + Objects.equal(expiration, other.getExpiration()) && + Objects.equal(type, other.type) && + Objects.equal(checkpointWritePointers, other.checkpointWritePointers); + } + + @Override + public int hashCode() { + return Objects.hashCode(visibilityUpperBound, expiration, type, checkpointWritePointers); + } + + @Override + public String toString() { + return Objects.toStringHelper(this) + .add("visibilityUpperBound", visibilityUpperBound) + .add("expiration", expiration) + .add("type", type) + .add("checkpointWritePointers", checkpointWritePointers) + .toString(); + } + } + +}
http://git-wip-us.apache.org/repos/asf/incubator-tephra/blob/9c693743/tephra-core/src/main/java/org/apache/tephra/TransactionServiceMain.java ---------------------------------------------------------------------- diff --git a/tephra-core/src/main/java/org/apache/tephra/TransactionServiceMain.java b/tephra-core/src/main/java/org/apache/tephra/TransactionServiceMain.java new file mode 100644 index 0000000..0a9dd1d --- /dev/null +++ b/tephra-core/src/main/java/org/apache/tephra/TransactionServiceMain.java @@ -0,0 +1,149 @@ +/* + * 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.tephra; + +import com.google.inject.Guice; +import com.google.inject.Injector; +import org.apache.hadoop.conf.Configuration; +import org.apache.tephra.distributed.TransactionService; +import org.apache.tephra.runtime.ConfigModule; +import org.apache.tephra.runtime.DiscoveryModules; +import org.apache.tephra.runtime.TransactionClientModule; +import org.apache.tephra.runtime.TransactionModules; +import org.apache.tephra.runtime.ZKModule; +import org.apache.tephra.util.ConfigurationFactory; +import org.apache.twill.zookeeper.ZKClientService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.CountDownLatch; + +/** + * Driver class to start and stop tx in distributed mode. + */ +public class TransactionServiceMain { + + private static final Logger LOG = LoggerFactory.getLogger(TransactionServiceMain.class); + + private Configuration conf; + private TransactionService txService; + + public static void main(String args[]) throws Exception { + TransactionServiceMain instance = new TransactionServiceMain(); + instance.doMain(args); + } + + public TransactionServiceMain() { + this(null); + } + + public TransactionServiceMain(Configuration conf) { + this.conf = conf; + } + + /** + * The main method. It simply call methods in the same sequence + * as if the program is started by jsvc. + */ + public void doMain(final String[] args) throws Exception { + final CountDownLatch shutdownLatch = new CountDownLatch(1); + Runtime.getRuntime().addShutdownHook(new Thread() { + @Override + public void run() { + try { + try { + TransactionServiceMain.this.stop(); + } finally { + try { + TransactionServiceMain.this.destroy(); + } finally { + shutdownLatch.countDown(); + } + } + } catch (Throwable t) { + LOG.error("Exception when shutting down: " + t.getMessage(), t); + } + } + }); + init(args); + start(); + + shutdownLatch.await(); + } + + /** + * Invoked by jsvc to initialize the program. + */ + public void init(String[] args) { + if (conf == null) { + conf = new ConfigurationFactory().get(); + } + } + + /** + * Invoked by jsvc to start the program. + */ + public void start() throws Exception { + Injector injector = Guice.createInjector( + new ConfigModule(conf), + new ZKModule(), + new DiscoveryModules().getDistributedModules(), + new TransactionModules().getDistributedModules(), + new TransactionClientModule() + ); + + ZKClientService zkClientService = injector.getInstance(ZKClientService.class); + zkClientService.startAndWait(); + + // start a tx server + txService = injector.getInstance(TransactionService.class); + try { + LOG.info("Starting {}", getClass().getSimpleName()); + txService.startAndWait(); + } catch (Exception e) { + System.err.println("Failed to start service: " + e.getMessage()); + } + } + + /** + * Invoked by jsvc to stop the program. + */ + public void stop() { + LOG.info("Stopping {}", getClass().getSimpleName()); + if (txService == null) { + return; + } + try { + if (txService.isRunning()) { + txService.stopAndWait(); + } + } catch (Throwable e) { + LOG.error("Failed to shutdown transaction service.", e); + // because shutdown hooks execute concurrently, the logger may be closed already: thus also print it. + System.err.println("Failed to shutdown transaction service: " + e.getMessage()); + e.printStackTrace(System.err); + } + } + + /** + * Invoked by jsvc for resource cleanup. + */ + public void destroy() { } + +} http://git-wip-us.apache.org/repos/asf/incubator-tephra/blob/9c693743/tephra-core/src/main/java/org/apache/tephra/TransactionSystemClient.java ---------------------------------------------------------------------- diff --git a/tephra-core/src/main/java/org/apache/tephra/TransactionSystemClient.java b/tephra-core/src/main/java/org/apache/tephra/TransactionSystemClient.java new file mode 100644 index 0000000..9009f84 --- /dev/null +++ b/tephra-core/src/main/java/org/apache/tephra/TransactionSystemClient.java @@ -0,0 +1,142 @@ +/* + * 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.tephra; + +import java.io.InputStream; +import java.util.Collection; +import java.util.Set; + +/** + * Client talking to transaction system. + * See also {@link TransactionAware}. + * todo: explain Omid. + */ +public interface TransactionSystemClient { + /** + * Starts new short transaction. + * @return instance of {@link Transaction} + */ + // TODO: "short" and "long" are very misleading names. Use transaction attributes like "detect conflicts or not", etc. + Transaction startShort(); + + /** + * Starts new short transaction. + * @param timeout the timeout for the transaction + * @return instance of {@link Transaction} + */ + Transaction startShort(int timeout); + + /** + * Starts new long transaction. + * @return instance of {@link Transaction} + */ + Transaction startLong(); + + // this pre-commit detects conflicts with other transactions committed so far + // NOTE: the changes set should not change after this operation, this may help us do some extra optimizations + // NOTE: there should be time constraint on how long does it take to commit changes by the client after this operation + // is submitted so that we can cleanup related resources + // NOTE: as of now you can call this method multiple times, each time the changeSet of tx will be updated. Not sure + // if we can call it a feature or a side-affect of implementation. It makes more sense to append changeset, but + // before we really need it we don't do it because it will slow down tx manager throughput. + + /** + * Checks if transaction with the set of changes can be committed. E.g. it can check conflicts with other changes and + * refuse commit if there are conflicts. It is assumed that no other changes will be done in between this method call + * and {@link #commit(Transaction)} which may check conflicts again to avoid races. + * <p/> + * Since we do conflict detection at commit time as well, this may seem redundant. The idea is to check for conflicts + * before we persist changes to avoid rollback in case of conflicts as much as possible. + * NOTE: in some situations we may want to skip this step to save on RPC with a risk of many rollback ops. So by + * default we take safe path. + * + * @param tx transaction to verify + * @param changeIds ids of changes made by transaction + * @return true if transaction can be committed otherwise false + */ + boolean canCommit(Transaction tx, Collection<byte[]> changeIds) throws TransactionNotInProgressException; + + /** + * Makes transaction visible. It will again check conflicts of changes submitted previously with + * {@link #canCommit(Transaction, java.util.Collection)} + * @param tx transaction to make visible. + * @return true if transaction can be committed otherwise false + */ + boolean commit(Transaction tx) throws TransactionNotInProgressException; + + /** + * Makes transaction visible. You should call it only when all changes of this tx are undone. + * NOTE: it will not throw {@link TransactionNotInProgressException} if transaction has timed out. + * @param tx transaction to make visible. + */ + void abort(Transaction tx); + + /** + * Makes transaction invalid. You should call it if not all changes of this tx could be undone. + * NOTE: it will not throw {@link TransactionNotInProgressException} if transaction has timed out. + * @param tx transaction id to invalidate. + * @return true if transaction has been successfully invalidated + */ + boolean invalidate(long tx); + + /** + * Performs a checkpoint operation on the current transaction, returning a new Transaction instance with the + * updated state. A checkpoint operation assigns a new write pointer for the current transaction. + * @param tx the current transaction to checkpoint + * @return an updated transaction instance with the new write pointer + */ + Transaction checkpoint(Transaction tx) throws TransactionNotInProgressException; + + /** + * Retrieves the state of the transaction manager and send it as a stream. The snapshot will not be persisted. + * @return an input stream containing an encoded snapshot of the transaction manager + */ + InputStream getSnapshotInputStream() throws TransactionCouldNotTakeSnapshotException; + + /** + * Return the status of the transaction Manager + * @return a String which denotes the status of txManager + */ + String status(); + + /** + * Resets the state of the transaction manager. + */ + void resetState(); + + /** + * Removes the given transaction ids from the invalid list. + * @param invalidTxIds transaction ids + * @return true if invalid list got changed, false otherwise + */ + boolean truncateInvalidTx(Set<Long> invalidTxIds); + + /** + * Removes all transaction ids started before the given time from invalid list. + * @param time time in milliseconds + * @return true if invalid list got changed, false otherwise + * @throws InvalidTruncateTimeException if there are any in-progress transactions started before given time + */ + boolean truncateInvalidTxBefore(long time) throws InvalidTruncateTimeException; + + /** + * @return the size of invalid list + */ + int getInvalidSize(); +} http://git-wip-us.apache.org/repos/asf/incubator-tephra/blob/9c693743/tephra-core/src/main/java/org/apache/tephra/TxConstants.java ---------------------------------------------------------------------- diff --git a/tephra-core/src/main/java/org/apache/tephra/TxConstants.java b/tephra-core/src/main/java/org/apache/tephra/TxConstants.java new file mode 100644 index 0000000..7f7fc8c --- /dev/null +++ b/tephra-core/src/main/java/org/apache/tephra/TxConstants.java @@ -0,0 +1,343 @@ +/* + * 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.tephra; + +import org.apache.tephra.snapshot.DefaultSnapshotCodec; +import org.apache.tephra.snapshot.SnapshotCodecV2; +import org.apache.tephra.snapshot.SnapshotCodecV3; +import org.apache.tephra.snapshot.SnapshotCodecV4; + +import java.util.concurrent.TimeUnit; + +/** + * Transaction system constants + */ +public class TxConstants { + /** + * Defines what level of conflict detection should be used for transactions. {@code ROW} means that only the + * table name and row key for each change will be used to determine if the transaction change sets conflict. + * {@code COLUMN} means that the table name, row key, column family, and column qualifier will all be used to + * identify write conflicts. {@code NONE} means that no conflict detection will be performed, but transaction + * clients will still track the current transaction's change set to rollback any persisted changes in the event of + * a failure. This should only be used where writes to the same coordinate should never conflict, such as + * append-only data. The default value used by {@code TransactionAwareHTable} implementations is {@code COLUMN}. + * + * <p> + * <strong>Note: for a given table, all clients must use the same conflict detection setting!</strong> + * Otherwise conflicts will not be flagged correctly. + * </p> + */ + public enum ConflictDetection { + ROW, + COLUMN, + NONE + } + + /** + * Property set for {@code org.apache.hadoop.hbase.HColumnDescriptor} to configure time-to-live on data within + * the column family. The value given is in milliseconds. Once a cell's data has surpassed the given value in age, + * the cell's data will no longer be visible and may be garbage collected. + */ + public static final String PROPERTY_TTL = "dataset.table.ttl"; + + /** + * This is how many tx we allow per millisecond, if you care about the system for 100 years: + * Long.MAX_VALUE / (System.currentTimeMillis() + TimeUnit.DAYS.toMillis(365 * 100)) = + * (as of Feb 20, 2014) 2,028,653. It is safe and convenient to use 1,000,000 as multiplier: + * <ul> + * <li> + * we hardly can do more than 1 billion txs per second + * </li> + * <li> + * long value will not overflow for 200 years + * </li> + * <li> + * makes reading & debugging easier if multiplier is 10^n + * </li> + * </ul> + */ + public static final long MAX_TX_PER_MS = 1000000; + + /** + * Since HBase {@code Delete} operations cannot be undone at the same timestamp, "deleted" data is instead + * overwritten with an empty {@code byte[]} to flag it as removed. Cells with empty values will be filtered out + * of the results for read operations. If cells with empty values should be included in results (meaning data + * cannot be transactionally deleted), then set this configuration property to true. + */ + public static final String ALLOW_EMPTY_VALUES_KEY = "data.tx.allow.empty.values"; + public static final boolean ALLOW_EMPTY_VALUES_DEFAULT = false; + + /** + * Key used to set the serialized transaction as an attribute on Get and Scan operations. + */ + public static final String TX_OPERATION_ATTRIBUTE_KEY = "cask.tx"; + /** + * Key used to flag a delete operation as part of a transaction rollback. This is used so that the + * {@code TransactionProcessor} coprocessor loaded on a table can differentiate between deletes issued + * as part of a normal client operation versus those performed when rolling back a transaction. + */ + public static final String TX_ROLLBACK_ATTRIBUTE_KEY = "cask.tx.rollback"; + + /** + * Column qualifier used for a special delete marker tombstone, which identifies an entire column family as deleted. + */ + public static final byte[] FAMILY_DELETE_QUALIFIER = new byte[0]; + + // Constants for monitoring status + public static final String STATUS_OK = "OK"; + public static final String STATUS_NOTOK = "NOTOK"; + + /** + * Indicates whether data written before Tephra was enabled on a table should be readable. Reading non-transactional + * data can lead to slight performance penalty. Hence it is disabled by default. + * @see <a href="https://issues.cask.co/browse/TEPHRA-89";>TEPHRA-89</a> + */ + public static final String READ_NON_TX_DATA = "data.tx.read.pre.existing"; + public static final boolean DEFAULT_READ_NON_TX_DATA = false; + + /** + * TransactionManager configuration. + */ + public static final class Manager { + // TransactionManager configuration + public static final String CFG_DO_PERSIST = "tx.persist"; + /** Directory in HDFS used for transaction snapshot and log storage. */ + public static final String CFG_TX_SNAPSHOT_DIR = "data.tx.snapshot.dir"; + /** Directory on the local filesystem used for transaction snapshot and log storage. */ + public static final String CFG_TX_SNAPSHOT_LOCAL_DIR = "data.tx.snapshot.local.dir"; + /** How often to clean up timed out transactions, in seconds, or 0 for no cleanup. */ + public static final String CFG_TX_CLEANUP_INTERVAL = "data.tx.cleanup.interval"; + /** The user id to access HDFS if not running in secure HDFS. */ + public static final String CFG_TX_HDFS_USER = "data.tx.hdfs.user"; + /** Default value for how often to check in-progress transactions for expiration, in seconds. */ + public static final int DEFAULT_TX_CLEANUP_INTERVAL = 10; + /** + * The timeout for a transaction, in seconds. If the transaction is not finished in that time, + * it is marked invalid. + */ + public static final String CFG_TX_TIMEOUT = "data.tx.timeout"; + /** Default value for transaction timeout, in seconds. */ + public static final int DEFAULT_TX_TIMEOUT = 30; + /** + * The timeout for a long running transaction, in seconds. If the transaction is not finished in that time, + * it is marked invalid. + */ + public static final String CFG_TX_LONG_TIMEOUT = "data.tx.long.timeout"; + /** Default value for long running transaction timeout, in seconds. */ + public static final int DEFAULT_TX_LONG_TIMEOUT = (int) TimeUnit.DAYS.toSeconds(1); + /** The frequency (in seconds) to perform periodic snapshots, or 0 for no periodic snapshots. */ + public static final String CFG_TX_SNAPSHOT_INTERVAL = "data.tx.snapshot.interval"; + /** Default value for frequency of periodic snapshots of transaction state. */ + public static final long DEFAULT_TX_SNAPSHOT_INTERVAL = 300; + /** Number of most recent transaction snapshots to retain. */ + public static final String CFG_TX_SNAPSHOT_RETAIN = "data.tx.snapshot.retain"; + /** Default value for number of most recent snapshots to retain. */ + public static final int DEFAULT_TX_SNAPSHOT_RETAIN = 10; + } + + /** + * TransactionService configuration. + */ + public static final class Service { + + /** for the zookeeper quorum string for leader election for tx server. */ + public static final String CFG_DATA_TX_ZOOKEEPER_QUORUM + = "data.tx.zookeeper.quorum"; + + /** for the name used to announce service availability to discovery service */ + public static final String CFG_DATA_TX_DISCOVERY_SERVICE_NAME + = "data.tx.discovery.service.name"; + + /** for the port of the tx server. */ + public static final String CFG_DATA_TX_BIND_PORT + = "data.tx.bind.port"; + + /** for the address (hostname) of the tx server. */ + public static final String CFG_DATA_TX_BIND_ADDRESS + = "data.tx.bind.address"; + + /** the number of IO threads in the tx service. */ + public static final String CFG_DATA_TX_SERVER_IO_THREADS + = "data.tx.server.io.threads"; + + /** the number of handler threads in the tx service. */ + public static final String CFG_DATA_TX_SERVER_THREADS + = "data.tx.server.threads"; + + public static final String CFG_DATA_TX_THRIFT_MAX_READ_BUFFER + = "data.tx.thrift.max.read.buffer"; + + public static final String DEFAULT_DATA_TX_DISCOVERY_SERVICE_NAME + = "transaction"; + + /** default tx service port. */ + public static final int DEFAULT_DATA_TX_BIND_PORT + = 15165; + + /** default tx service address. */ + public static final String DEFAULT_DATA_TX_BIND_ADDRESS + = "0.0.0.0"; + + /** default number of handler IO threads in tx service. */ + public static final int DEFAULT_DATA_TX_SERVER_IO_THREADS + = 2; + + /** default number of handler threads in tx service. */ + public static final int DEFAULT_DATA_TX_SERVER_THREADS + = 20; + + /** default thrift max read buffer size */ + public static final int DEFAULT_DATA_TX_THRIFT_MAX_READ_BUFFER + = 16 * 1024 * 1024; + + // Configuration key names and defaults used by tx client. + + /** to specify the tx client socket timeout in ms. */ + public static final String CFG_DATA_TX_CLIENT_TIMEOUT + = "data.tx.client.timeout"; + + /** to specify the tx client provider strategy. */ + public static final String CFG_DATA_TX_CLIENT_PROVIDER + = "data.tx.client.provider"; + + /** to specify the number of threads for client provider "pool". */ + public static final String CFG_DATA_TX_CLIENT_COUNT + = "data.tx.client.count"; + + /** timeout (in milliseconds) for obtaining client from client provider "pool". */ + public static final String CFG_DATA_TX_CLIENT_OBTAIN_TIMEOUT_MS + = "data.tx.client.obtain.timeout"; + + /** to specify the retry strategy for a failed thrift call. */ + public static final String CFG_DATA_TX_CLIENT_RETRY_STRATEGY + = "data.tx.client.retry.strategy"; + + /** to specify the number of times to retry a failed thrift call. */ + public static final String CFG_DATA_TX_CLIENT_ATTEMPTS + = "data.tx.client.retry.attempts"; + + /** to specify the initial sleep time for retry strategy backoff. */ + public static final String CFG_DATA_TX_CLIENT_BACKOFF_INITIAL + = "data.tx.client.retry.backoff.initial"; + + /** to specify the backoff factor for retry strategy backoff. */ + public static final String CFG_DATA_TX_CLIENT_BACKOFF_FACTOR + = "data.tx.client.retry.backoff.factor"; + + /** to specify the sleep time limit for retry strategy backoff. */ + public static final String CFG_DATA_TX_CLIENT_BACKOFF_LIMIT + = "data.tx.client.retry.backoff.limit"; + + /** the default tx client socket timeout in milli seconds. */ + public static final int DEFAULT_DATA_TX_CLIENT_TIMEOUT_MS + = 30 * 1000; + + /** default number of tx clients for client provider "pool". */ + public static final int DEFAULT_DATA_TX_CLIENT_COUNT + = 50; + + /** default timeout (in milliseconds) for obtaining client from client provider "pool". */ + public static final long DEFAULT_DATA_TX_CLIENT_OBTAIN_TIMEOUT_MS + = TimeUnit.SECONDS.toMillis(3); + + /** default tx client provider strategy. */ + public static final String DEFAULT_DATA_TX_CLIENT_PROVIDER + = "pool"; + + /** retry strategy for thrift clients, e.g. backoff, or n-times. */ + public static final String DEFAULT_DATA_TX_CLIENT_RETRY_STRATEGY + = "backoff"; + + /** default number of attempts for strategy n-times. */ + public static final int DEFAULT_DATA_TX_CLIENT_ATTEMPTS + = 2; + + /** default initial sleep is 100ms. */ + public static final int DEFAULT_DATA_TX_CLIENT_BACKOFF_INITIAL + = 100; + + /** default backoff factor is 4. */ + public static final int DEFAULT_DATA_TX_CLIENT_BACKOFF_FACTOR + = 4; + + /** default sleep limit is 30 sec. */ + public static final int DEFAULT_DATA_TX_CLIENT_BACKOFF_LIMIT + = 30 * 1000; + } + + /** + * Configuration properties for metrics reporting + */ + public static final class Metrics { + /** + * Frequency at which metrics should be reported, in seconds. + */ + public static final String REPORT_PERIOD_KEY = "data.tx.metrics.period"; + /** + * Default report period for metrics, in seconds. + */ + public static final int REPORT_PERIOD_DEFAULT = 60; + } + + /** + * Configuration properties used by HBase + */ + public static final class HBase { + public static final String ZOOKEEPER_QUORUM = "hbase.zookeeper.quorum"; + public static final String ZK_SESSION_TIMEOUT = "zookeeper.session.timeout"; + public static final int DEFAULT_ZK_SESSION_TIMEOUT = 180 * 1000; + } + + /** + * Configuration for the TransactionDataJanitor coprocessor. + */ + public static final class DataJanitor { + /** + * Whether or not the TransactionDataJanitor coprocessor should be enabled on tables. + * Disable for testing. + */ + public static final String CFG_TX_JANITOR_ENABLE = "data.tx.janitor.enable"; + public static final boolean DEFAULT_TX_JANITOR_ENABLE = true; + } + + /** + * Configuration for the transaction snapshot persistence. + */ + public static final class Persist { + /** + * The class names of all known transaction snapshot codecs. + */ + public static final String CFG_TX_SNAPHOT_CODEC_CLASSES = "data.tx.snapshot.codecs"; + public static final Class[] DEFAULT_TX_SNAPHOT_CODEC_CLASSES = + { DefaultSnapshotCodec.class, SnapshotCodecV2.class, SnapshotCodecV3.class, SnapshotCodecV4.class }; + } + + /** + * Configuration for transaction log edit entries + */ + public static final class TransactionLog { + /** + * Key used to denote the number of entries appended. + */ + public static final String NUM_ENTRIES_APPENDED = "count"; + public static final String VERSION_KEY = "version"; + public static final byte CURRENT_VERSION = 2; + } + +}
