Merge pull request #8 from dgomezferro/master

Early stop & aborts snapshot

src/main/java/com/yahoo/omid/client/TSOClient.java

@@ -32,8 +32,6 @@
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executors;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.jboss.netty.bootstrap.ClientBootstrap;
 import org.jboss.netty.channel.Channel;
@@ -48,6 +46,8 @@
 import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
 import org.jboss.netty.handler.execution.ExecutionHandler;
 import org.jboss.netty.handler.execution.OrderedMemoryAwareThreadPoolExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.yahoo.omid.tso.Committed;
 import com.yahoo.omid.tso.RowKey;
@@ -67,7 +67,7 @@
 import com.yahoo.omid.tso.serialization.TSOEncoder;
 
 public class TSOClient extends SimpleChannelHandler {
-   private static final Log LOG = LogFactory.getLog(TSOClient.class);
+   private static final Logger LOG = LoggerFactory.getLogger(TSOClient.class);
 
    public static long askedTSO = 0;
 
@@ -333,8 +333,8 @@ public TSOClient(Configuration conf) throws IOException {
 
       String host = conf.get("tso.host");
       int port = conf.getInt("tso.port", 1234);
-      max_retries = conf.getInt("tso.max_retries", 10);
-      retry_delay_ms = conf.getInt("tso.retry_delay_ms", 3000); 
+      max_retries = conf.getInt("tso.max_retries", 100);
+      retry_delay_ms = conf.getInt("tso.retry_delay_ms", 1000);
 
       if (host == null) {
          throw new IOException("tso.host missing from configuration");
@@ -359,7 +359,12 @@ private State connectIfNeeded() throws IOException {
             throw e;
          }
          retries++;
-         bootstrap.connect(addr);
+         bootstrap.connect(addr).addListener(new ChannelFutureListener() {
+            @Override
+            public void operationComplete(ChannelFuture future) throws Exception {
+               LOG.debug("Connection completed. Success: " + future.isSuccess());
+            }
+         });
          state = State.CONNECTING;
          return state;
       }
@@ -422,6 +427,7 @@ public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
          retries = 0;
       }
       clearState();
+      LOG.debug("Channel connected");
       Op o = queuedOps.poll();;
       while (o != null && state == State.CONNECTED) {
          o.execute(channel);
@@ -442,8 +448,24 @@ private void clearState() {
    public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e)
          throws Exception {
       synchronized(state) {
+         LOG.debug("Channel disconnected");
          channel = null;
          state = State.DISCONNECTED;
+         for (CreateCallback cb : createCallbacks) {
+            cb.error(new IOException("Channel Disconnected"));
+         }
+         for (CommitCallback cb : commitCallbacks.values()) {
+            cb.error(new IOException("Channel Disconnected"));
+         }
+         for (List<CommitQueryCallback> lcqb : isCommittedCallbacks.values()) {
+            for (CommitQueryCallback cqb : lcqb) {
+               cqb.error(new IOException("Channel Disconnected"));
+            }
+         }
+         createCallbacks.clear();
+         commitCallbacks.clear();
+         isCommittedCallbacks.clear();
+         connectIfNeeded();
       }
    }
 
@@ -545,15 +567,14 @@ public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
    public void exceptionCaught(ChannelHandlerContext ctx,
                                ExceptionEvent e)
          throws Exception {
-      System.out.println("Unexpected exception " + e.getCause());
-      e.getCause().printStackTrace();
+      LOG.error("Unexpected exception", e.getCause());
 
       synchronized(state) {
 
          if (state == State.CONNECTING) {
             state = State.RETRY_CONNECT_WAIT;
-            if (LOG.isTraceEnabled()) {
-               LOG.trace("Retrying connect in " + retry_delay_ms + "ms " + retries);
+            if (LOG.isDebugEnabled()) {
+               LOG.debug("Retrying connect in " + retry_delay_ms + "ms " + retries);
             }
             try {
                retryTimer.schedule(new TimerTask() {

src/main/java/com/yahoo/omid/tso/AbortedTransaction.java

@@ -0,0 +1,51 @@
+package com.yahoo.omid.tso;
+
+public class AbortedTransaction {
+   private long startTimestamp;
+   private long snapshot;
+
+   public AbortedTransaction(long startTimestamp, long snapshot) {
+      super();
+      this.startTimestamp = startTimestamp;
+      this.snapshot = snapshot;
+   }
+
+   public long getStartTimestamp() {
+      return startTimestamp;
+   }
+
+   public void setStartTimestamp(long startTimestamp) {
+      this.startTimestamp = startTimestamp;
+   }
+
+   public long getSnapshot() {
+      return snapshot;
+   }
+
+   public void setSnapshot(long snapshot) {
+      this.snapshot = snapshot;
+   }
+
+   @Override
+   public int hashCode() {
+      final int prime = 31;
+      int result = 1;
+      result = prime * result + (int) (startTimestamp ^ (startTimestamp >>> 32));
+      return result;
+   }
+
+   @Override
+   public boolean equals(Object obj) {
+      if (this == obj)
+         return true;
+      if (obj == null)
+         return false;
+      if (getClass() != obj.getClass())
+         return false;
+      AbortedTransaction other = (AbortedTransaction) obj;
+      if (startTimestamp != other.startTimestamp)
+         return false;
+      return true;
+   }
+
+}

src/main/java/com/yahoo/omid/tso/Bucket.java

@@ -59,7 +59,7 @@ public synchronized Set<Long> abortUncommited(long id) {
          return aborted;
       }
 
-      LOG.debug("Performing scanning...");
+      LOG.trace("Performing scanning...");
 
       for (int i = transactions.nextClearBit(firstUncommited); i >= 0
             && i <= lastCommited; i = transactions.nextClearBit(i + 1)) {

src/main/java/com/yahoo/omid/tso/ClientHandler.java

@@ -19,7 +19,6 @@
 import java.io.IOException;
 import java.util.Collections;
 import java.util.Date;
-import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
@@ -37,8 +36,6 @@
 import org.jboss.netty.channel.ChannelFutureListener;
 import org.jboss.netty.channel.ChannelHandlerContext;
 import org.jboss.netty.channel.ChannelStateEvent;
-import org.jboss.netty.channel.Channels;
-import org.jboss.netty.channel.ExceptionEvent;
 
 import com.yahoo.omid.client.SyncAbortCompleteCallback;
 import com.yahoo.omid.client.SyncCommitCallback;
@@ -161,13 +158,9 @@ public ClientHandler(Configuration conf, int nbMessage, int inflight, boolean pa
    @Override
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
       super.channelConnected(ctx, e);
-      try {
-         Thread.sleep(15000);
-      } catch (InterruptedException e1) {
-         //ignore
-      }
       startDate = new Date();
       channel = e.getChannel();
+      outstandingTransactions = 0;
       startTransaction();
    }
 
@@ -269,18 +262,6 @@ private long getSizeAborted() {
       return aborted.size() * 8 * 8;
    }
 
-   @Override
-   public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
-      if (e.getCause() instanceof IOException) {
-         LOG.warn("IOException from downstream.", e.getCause());
-      } else {
-         LOG.warn("Unexpected exception from downstream.", e.getCause());
-      }
-      // Offer default object
-      answer.offer(false);
-      Channels.close(e.getChannel());
-   }
-
    private java.util.Random rnd;
 
    private boolean pauseClient;

src/main/java/com/yahoo/omid/tso/CommitHashMap.java

@@ -16,6 +16,12 @@
 
 package com.yahoo.omid.tso;
 
+import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.jboss.netty.util.internal.ConcurrentHashMap;
+
 /**
  * A hash map that uses byte[] for the key rather than longs.
  * 
@@ -138,21 +144,27 @@ public CommitHashMap(int initialCapacity, float loadFactor) {
 
    // set of half aborted transactions
    // TODO: set the initial capacity in a smarter way
-   java.util.HashSet<Long> halfAborted = new java.util.HashSet<Long>(10000);
+   Set<AbortedTransaction> halfAborted = Collections.newSetFromMap(new ConcurrentHashMap<AbortedTransaction, Boolean>(10000));
+
+   private AtomicLong abortedSnapshot = new AtomicLong();
+
+   long getAndIncrementAbortedSnapshot() {
+      return abortedSnapshot.getAndIncrement();
+   }
 
    // add a new half aborted transaction
    void setHalfAborted(long startTimestamp) {
-      halfAborted.add(startTimestamp);
+      halfAborted.add(new AbortedTransaction(startTimestamp, abortedSnapshot.get()));
    }
 
    // call when a half aborted transaction is fully aborted
    void setFullAborted(long startTimestamp) {
-      halfAborted.remove(startTimestamp);
+      halfAborted.remove(new AbortedTransaction(startTimestamp, 0));
    }
 
    // query to see if a transaction is half aborted
    boolean isHalfAborted(long startTimestamp) {
-      return halfAborted.contains(startTimestamp);
+      return halfAborted.contains(new AbortedTransaction(startTimestamp, 0));
    }
 }