core/src/main/java/org/apache/accumulo/fate/Fate.java - accumulo - Git at Google

 /*
  * 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.accumulo.fate;

 import java.util.EnumSet;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;

 import org.apache.accumulo.fate.ReadOnlyTStore.TStatus;
 import org.apache.accumulo.fate.util.LoggingRunnable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 /**
  * Fault tolerant executor
  */
 public class Fate<T> {

   private static final String DEBUG_PROP = "debug";
   private static final String AUTO_CLEAN_PROP = "autoClean";
   private static final String EXCEPTION_PROP = "exception";
   private static final String RETURN_PROP = "return";

   private static final Logger log = LoggerFactory.getLogger(Fate.class);
   private final Logger runnerLog = LoggerFactory.getLogger(TransactionRunner.class);

   private TStore<T> store;
   private T environment;
   private ExecutorService executor;

   private static final EnumSet<TStatus> FINISHED_STATES = EnumSet.of(TStatus.FAILED,
       TStatus.SUCCESSFUL, TStatus.UNKNOWN);

   private AtomicBoolean keepRunning = new AtomicBoolean(true);

   private class TransactionRunner implements Runnable {

     @Override
     public void run() {
       while (keepRunning.get()) {
         long deferTime = 0;
         Long tid = null;
         try {
           tid = store.reserve();
           TStatus status = store.getStatus(tid);
           Repo<T> op = store.top(tid);
           if (status == TStatus.FAILED_IN_PROGRESS) {
             processFailed(tid, op);
           } else {
             Repo<T> prevOp = null;
             try {
               deferTime = op.isReady(tid, environment);

               if (deferTime == 0) {
                 prevOp = op;
                 op = op.call(tid, environment);
               } else
                 continue;

             } catch (Exception e) {
               transitionToFailed(tid, op, e);
               continue;
             }

             if (op == null) {
               // transaction is finished
               String ret = prevOp.getReturn();
               if (ret != null)
                 store.setProperty(tid, RETURN_PROP, ret);
               store.setStatus(tid, TStatus.SUCCESSFUL);
               doCleanUp(tid);
             } else {
               try {
                 store.push(tid, op);
               } catch (StackOverflowException e) {
                 // the op that failed to push onto the stack was never executed, so no need to undo
                 // it
                 // just transition to failed and undo the ops that executed
                 transitionToFailed(tid, op, e);
                 continue;
               }
             }
           }
         } catch (Exception e) {
           runnerLog.error("Uncaught exception in FATE runner thread.", e);
         } finally {
           if (null != tid) {
             store.unreserve(tid, deferTime);
           }
         }

       }
     }

     private void transitionToFailed(long tid, Repo<T> op, Exception e) {
       String tidStr = String.format("%016x", tid);
       final String msg = "Failed to execute Repo, tid=" + tidStr;
       // Certain FATE ops that throw exceptions don't need to be propagated up to the Monitor
       // as a warning. They're a normal, handled failure condition.
       if (e instanceof AcceptableException) {
         log.debug(msg, e.getCause());
       } else {
         log.warn(msg, e);
       }
       store.setProperty(tid, EXCEPTION_PROP, e);
       store.setStatus(tid, TStatus.FAILED_IN_PROGRESS);
       log.info("Updated status for Repo with tid={} to FAILED_IN_PROGRESS", tidStr);
     }

     private void processFailed(long tid, Repo<T> op) {
       while (op != null) {
         undo(tid, op);

         store.pop(tid);
         op = store.top(tid);
       }

       store.setStatus(tid, TStatus.FAILED);
       doCleanUp(tid);
     }

     private void doCleanUp(long tid) {
       Boolean autoClean = (Boolean) store.getProperty(tid, AUTO_CLEAN_PROP);
       if (autoClean != null && autoClean) {
         store.delete(tid);
       } else {
         // no longer need persisted operations, so delete them to save space in case
         // TX is never cleaned up...
         while (store.top(tid) != null)
           store.pop(tid);
       }
     }

     private void undo(long tid, Repo<T> op) {
       try {
         op.undo(tid, environment);
       } catch (Exception e) {
         log.warn("Failed to undo Repo, tid=" + String.format("%016x", tid), e);
       }
     }

   }

   /**
    * Creates a Fault-tolerant executor.
    * <p>
    * Note: Users of this class should call {@link #startTransactionRunners(int)} to launch the
    * worker threads after creating a Fate object.
    */
   public Fate(T environment, TStore<T> store) {
     this.store = store;
     this.environment = environment;
   }

   /**
    * Launches the specified number of worker threads.
    */
   public void startTransactionRunners(int numThreads) {
     final AtomicInteger runnerCount = new AtomicInteger(0);
     executor = Executors.newFixedThreadPool(numThreads, r -> {
       Thread t = new Thread(new LoggingRunnable(log, r),
           "Repo runner " + runnerCount.getAndIncrement());
       t.setDaemon(true);
       return t;
     });
     for (int i = 0; i < numThreads; i++) {
       executor.execute(new TransactionRunner());
     }
   }

   // get a transaction id back to the requester before doing any work
   public long startTransaction() {
     return store.create();
   }

   // start work in the transaction.. it is safe to call this
   // multiple times for a transaction... but it will only seed once
   public void seedTransaction(long tid, Repo<T> repo, boolean autoCleanUp) {
     store.reserve(tid);
     try {
       if (store.getStatus(tid) == TStatus.NEW) {
         if (store.top(tid) == null) {
           try {
             store.push(tid, repo);
           } catch (StackOverflowException e) {
             // this should not happen
             throw new RuntimeException(e);
           }
         }

         if (autoCleanUp)
           store.setProperty(tid, AUTO_CLEAN_PROP, autoCleanUp);

         store.setProperty(tid, DEBUG_PROP, repo.getDescription());

         store.setStatus(tid, TStatus.IN_PROGRESS);
       }
     } finally {
       store.unreserve(tid, 0);
     }

   }

   // check on the transaction
   public TStatus waitForCompletion(long tid) {
     return store.waitForStatusChange(tid, FINISHED_STATES);
   }

   // resource cleanup
   public void delete(long tid) {
     store.reserve(tid);
     try {
       switch (store.getStatus(tid)) {
         case NEW:
         case FAILED:
         case SUCCESSFUL:
           store.delete(tid);
           break;
         case FAILED_IN_PROGRESS:
         case IN_PROGRESS:
           throw new IllegalStateException(
               "Can not delete in progress transaction " + String.format("%016x", tid));
         case UNKNOWN:
           // nothing to do, it does not exist
           break;
       }
     } finally {
       store.unreserve(tid, 0);
     }
   }

   public String getReturn(long tid) {
     store.reserve(tid);
     try {
       if (store.getStatus(tid) != TStatus.SUCCESSFUL)
         throw new IllegalStateException("Tried to get exception when transaction "
             + String.format("%016x", tid) + " not in successful state");
       return (String) store.getProperty(tid, RETURN_PROP);
     } finally {
       store.unreserve(tid, 0);
     }
   }

   // get reportable failures
   public Exception getException(long tid) {
     store.reserve(tid);
     try {
       if (store.getStatus(tid) != TStatus.FAILED)
         throw new IllegalStateException("Tried to get exception when transaction "
             + String.format("%016x", tid) + " not in failed state");
       return (Exception) store.getProperty(tid, EXCEPTION_PROP);
     } finally {
       store.unreserve(tid, 0);
     }
   }

   /**
    * Flags that FATE threadpool to clear out and end. Does not actively stop running FATE processes.
    */
   public void shutdown() {
     keepRunning.set(false);
     executor.shutdown();
   }

 }
	/*
	* 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.accumulo.fate;

	import java.util.EnumSet;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.Executors;
	import java.util.concurrent.atomic.AtomicBoolean;
	import java.util.concurrent.atomic.AtomicInteger;

	import org.apache.accumulo.fate.ReadOnlyTStore.TStatus;
	import org.apache.accumulo.fate.util.LoggingRunnable;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	/**
	* Fault tolerant executor
	*/
	public class Fate<T> {

	private static final String DEBUG_PROP = "debug";
	private static final String AUTO_CLEAN_PROP = "autoClean";
	private static final String EXCEPTION_PROP = "exception";
	private static final String RETURN_PROP = "return";

	private static final Logger log = LoggerFactory.getLogger(Fate.class);
	private final Logger runnerLog = LoggerFactory.getLogger(TransactionRunner.class);

	private TStore<T> store;
	private T environment;
	private ExecutorService executor;

	private static final EnumSet<TStatus> FINISHED_STATES = EnumSet.of(TStatus.FAILED,
	TStatus.SUCCESSFUL, TStatus.UNKNOWN);

	private AtomicBoolean keepRunning = new AtomicBoolean(true);

	private class TransactionRunner implements Runnable {

	@Override
	public void run() {
	while (keepRunning.get()) {
	long deferTime = 0;
	Long tid = null;
	try {
	tid = store.reserve();
	TStatus status = store.getStatus(tid);
	Repo<T> op = store.top(tid);
	if (status == TStatus.FAILED_IN_PROGRESS) {
	processFailed(tid, op);
	} else {
	Repo<T> prevOp = null;
	try {
	deferTime = op.isReady(tid, environment);

	if (deferTime == 0) {
	prevOp = op;
	op = op.call(tid, environment);
	} else
	continue;

	} catch (Exception e) {
	transitionToFailed(tid, op, e);
	continue;
	}

	if (op == null) {
	// transaction is finished
	String ret = prevOp.getReturn();
	if (ret != null)
	store.setProperty(tid, RETURN_PROP, ret);
	store.setStatus(tid, TStatus.SUCCESSFUL);
	doCleanUp(tid);
	} else {
	try {
	store.push(tid, op);
	} catch (StackOverflowException e) {
	// the op that failed to push onto the stack was never executed, so no need to undo
	// it
	// just transition to failed and undo the ops that executed
	transitionToFailed(tid, op, e);
	continue;
	}
	}
	}
	} catch (Exception e) {
	runnerLog.error("Uncaught exception in FATE runner thread.", e);
	} finally {
	if (null != tid) {
	store.unreserve(tid, deferTime);
	}
	}

	}
	}

	private void transitionToFailed(long tid, Repo<T> op, Exception e) {
	String tidStr = String.format("%016x", tid);
	final String msg = "Failed to execute Repo, tid=" + tidStr;
	// Certain FATE ops that throw exceptions don't need to be propagated up to the Monitor
	// as a warning. They're a normal, handled failure condition.
	if (e instanceof AcceptableException) {
	log.debug(msg, e.getCause());
	} else {
	log.warn(msg, e);
	}
	store.setProperty(tid, EXCEPTION_PROP, e);
	store.setStatus(tid, TStatus.FAILED_IN_PROGRESS);
	log.info("Updated status for Repo with tid={} to FAILED_IN_PROGRESS", tidStr);
	}

	private void processFailed(long tid, Repo<T> op) {
	while (op != null) {
	undo(tid, op);

	store.pop(tid);
	op = store.top(tid);
	}

	store.setStatus(tid, TStatus.FAILED);
	doCleanUp(tid);
	}

	private void doCleanUp(long tid) {
	Boolean autoClean = (Boolean) store.getProperty(tid, AUTO_CLEAN_PROP);
	if (autoClean != null && autoClean) {
	store.delete(tid);
	} else {
	// no longer need persisted operations, so delete them to save space in case
	// TX is never cleaned up...
	while (store.top(tid) != null)
	store.pop(tid);
	}
	}

	private void undo(long tid, Repo<T> op) {
	try {
	op.undo(tid, environment);
	} catch (Exception e) {
	log.warn("Failed to undo Repo, tid=" + String.format("%016x", tid), e);
	}
	}

	}

	/**
	* Creates a Fault-tolerant executor.
	* <p>
	* Note: Users of this class should call {@link #startTransactionRunners(int)} to launch the
	* worker threads after creating a Fate object.
	*/
	public Fate(T environment, TStore<T> store) {
	this.store = store;
	this.environment = environment;
	}

	/**
	* Launches the specified number of worker threads.
	*/
	public void startTransactionRunners(int numThreads) {
	final AtomicInteger runnerCount = new AtomicInteger(0);
	executor = Executors.newFixedThreadPool(numThreads, r -> {
	Thread t = new Thread(new LoggingRunnable(log, r),
	"Repo runner " + runnerCount.getAndIncrement());
	t.setDaemon(true);
	return t;
	});
	for (int i = 0; i < numThreads; i++) {
	executor.execute(new TransactionRunner());
	}
	}

	// get a transaction id back to the requester before doing any work
	public long startTransaction() {
	return store.create();
	}

	// start work in the transaction.. it is safe to call this
	// multiple times for a transaction... but it will only seed once
	public void seedTransaction(long tid, Repo<T> repo, boolean autoCleanUp) {
	store.reserve(tid);
	try {
	if (store.getStatus(tid) == TStatus.NEW) {
	if (store.top(tid) == null) {
	try {
	store.push(tid, repo);
	} catch (StackOverflowException e) {
	// this should not happen
	throw new RuntimeException(e);
	}
	}

	if (autoCleanUp)
	store.setProperty(tid, AUTO_CLEAN_PROP, autoCleanUp);

	store.setProperty(tid, DEBUG_PROP, repo.getDescription());

	store.setStatus(tid, TStatus.IN_PROGRESS);
	}
	} finally {
	store.unreserve(tid, 0);
	}

	}

	// check on the transaction
	public TStatus waitForCompletion(long tid) {
	return store.waitForStatusChange(tid, FINISHED_STATES);
	}

	// resource cleanup
	public void delete(long tid) {
	store.reserve(tid);
	try {
	switch (store.getStatus(tid)) {
	case NEW:
	case FAILED:
	case SUCCESSFUL:
	store.delete(tid);
	break;
	case FAILED_IN_PROGRESS:
	case IN_PROGRESS:
	throw new IllegalStateException(
	"Can not delete in progress transaction " + String.format("%016x", tid));
	case UNKNOWN:
	// nothing to do, it does not exist
	break;
	}
	} finally {
	store.unreserve(tid, 0);
	}
	}

	public String getReturn(long tid) {
	store.reserve(tid);
	try {
	if (store.getStatus(tid) != TStatus.SUCCESSFUL)
	throw new IllegalStateException("Tried to get exception when transaction "
	+ String.format("%016x", tid) + " not in successful state");
	return (String) store.getProperty(tid, RETURN_PROP);
	} finally {
	store.unreserve(tid, 0);
	}
	}

	// get reportable failures
	public Exception getException(long tid) {
	store.reserve(tid);
	try {
	if (store.getStatus(tid) != TStatus.FAILED)
	throw new IllegalStateException("Tried to get exception when transaction "
	+ String.format("%016x", tid) + " not in failed state");
	return (Exception) store.getProperty(tid, EXCEPTION_PROP);
	} finally {
	store.unreserve(tid, 0);
	}
	}

	/**
	* Flags that FATE threadpool to clear out and end. Does not actively stop running FATE processes.
	*/
	public void shutdown() {
	keepRunning.set(false);
	executor.shutdown();
	}

	}