accumulo-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From e..@apache.org
Subject git commit: ACCUMULO-2172 wait for minor compaction flags to be flushed to the WAL
Date Fri, 10 Jan 2014 21:37:25 GMT
Updated Branches:
  refs/heads/1.5.1-SNAPSHOT 4822b1318 -> 3b41d37ed


ACCUMULO-2172 wait for minor compaction flags to be flushed to the WAL


Project: http://git-wip-us.apache.org/repos/asf/accumulo/repo
Commit: http://git-wip-us.apache.org/repos/asf/accumulo/commit/3b41d37e
Tree: http://git-wip-us.apache.org/repos/asf/accumulo/tree/3b41d37e
Diff: http://git-wip-us.apache.org/repos/asf/accumulo/diff/3b41d37e

Branch: refs/heads/1.5.1-SNAPSHOT
Commit: 3b41d37ed27b5dc8724b427ed6970b083464bd94
Parents: 4822b13
Author: Eric Newton <eric.newton@gmail.com>
Authored: Fri Jan 10 16:27:01 2014 -0500
Committer: Eric Newton <eric.newton@gmail.com>
Committed: Fri Jan 10 16:27:01 2014 -0500

----------------------------------------------------------------------
 .../server/tabletserver/log/DfsLogger.java      | 58 +++++++++-----------
 .../tabletserver/log/TabletServerLogger.java    |  4 +-
 2 files changed, 29 insertions(+), 33 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/accumulo/blob/3b41d37e/server/src/main/java/org/apache/accumulo/server/tabletserver/log/DfsLogger.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/accumulo/server/tabletserver/log/DfsLogger.java
b/server/src/main/java/org/apache/accumulo/server/tabletserver/log/DfsLogger.java
index 213c885..ded2962 100644
--- a/server/src/main/java/org/apache/accumulo/server/tabletserver/log/DfsLogger.java
+++ b/server/src/main/java/org/apache/accumulo/server/tabletserver/log/DfsLogger.java
@@ -43,6 +43,7 @@ import org.apache.accumulo.core.conf.Property;
 import org.apache.accumulo.core.data.KeyExtent;
 import org.apache.accumulo.core.data.Mutation;
 import org.apache.accumulo.core.util.Daemon;
+import org.apache.accumulo.core.util.Pair;
 import org.apache.accumulo.core.util.StringUtil;
 import org.apache.accumulo.server.logger.LogFileKey;
 import org.apache.accumulo.server.logger.LogFileValue;
@@ -89,7 +90,7 @@ public class DfsLogger {
 
   private final Object closeLock = new Object();
 
-  private static final DfsLogger.LogWork CLOSED_MARKER = new DfsLogger.LogWork(null, null);
+  private static final DfsLogger.LogWork CLOSED_MARKER = new DfsLogger.LogWork(null);
 
   private static final LogFileValue EMPTY = new LogFileValue();
 
@@ -145,12 +146,10 @@ public class DfsLogger {
   }
 
   static class LogWork {
-    List<TabletMutations> mutations;
     CountDownLatch latch;
     volatile Exception exception;
 
-    public LogWork(List<TabletMutations> mutations, CountDownLatch latch) {
-      this.mutations = mutations;
+    public LogWork(CountDownLatch latch) {
       this.latch = latch;
     }
   }
@@ -439,27 +438,20 @@ public class DfsLogger {
   public LoggerOperation log(int seq, int tid, Mutation mutation) throws IOException {
     return logManyTablets(Collections.singletonList(new TabletMutations(tid, seq, Collections.singletonList(mutation))));
   }
-
-  public LoggerOperation logManyTablets(List<TabletMutations> mutations) throws IOException
{
-    DfsLogger.LogWork work = new DfsLogger.LogWork(mutations, new CountDownLatch(1));
-
+  
+  private LoggerOperation logFileData(List<Pair<LogFileKey, LogFileValue>> keys)
throws IOException {
+    DfsLogger.LogWork work = new DfsLogger.LogWork(new CountDownLatch(1));
     synchronized (DfsLogger.this) {
       try {
-        for (TabletMutations tabletMutations : mutations) {
-          LogFileKey key = new LogFileKey();
-          key.event = MANY_MUTATIONS;
-          key.seq = tabletMutations.getSeq();
-          key.tid = tabletMutations.getTid();
-          LogFileValue value = new LogFileValue();
-          value.mutations = tabletMutations.getMutations();
-          write(key, value);
+        for (Pair<LogFileKey,LogFileValue> pair : keys) {
+          write(pair.getFirst(), pair.getSecond());
         }
       } catch (Exception e) {
         log.error(e, e);
         work.exception = e;
       }
     }
-
+    
     synchronized (closeLock) {
       // use a different lock for close check so that adding to work queue does not need
       // to wait on walog I/O operations
@@ -472,31 +464,35 @@ public class DfsLogger {
     return new LoggerOperation(work);
   }
 
-  public synchronized void minorCompactionFinished(int seq, int tid, String fqfn) throws
IOException {
+  public LoggerOperation logManyTablets(List<TabletMutations> mutations) throws IOException
{
+    List<Pair<LogFileKey, LogFileValue>> data = new ArrayList<Pair<LogFileKey,
LogFileValue>>();
+    for (TabletMutations tabletMutations : mutations) {
+      LogFileKey key = new LogFileKey();
+      key.event = MANY_MUTATIONS;
+      key.seq = tabletMutations.getSeq();
+      key.tid = tabletMutations.getTid();
+      LogFileValue value = new LogFileValue();
+      value.mutations = tabletMutations.getMutations();
+      data.add(new Pair<LogFileKey, LogFileValue>(key, value));
+    }
+    return logFileData(data);
+  }
+
+  public LoggerOperation minorCompactionFinished(int seq, int tid, String fqfn) throws IOException
{
     LogFileKey key = new LogFileKey();
     key.event = COMPACTION_FINISH;
     key.seq = seq;
     key.tid = tid;
-    try {
-      write(key, EMPTY);
-    } catch (IOException ex) {
-      log.error(ex);
-      throw ex;
-    }
+    return logFileData(Collections.singletonList(new Pair<LogFileKey, LogFileValue>(key,
EMPTY)));
   }
 
-  public synchronized void minorCompactionStarted(int seq, int tid, String fqfn) throws IOException
{
+  public LoggerOperation minorCompactionStarted(int seq, int tid, String fqfn) throws IOException
{
     LogFileKey key = new LogFileKey();
     key.event = COMPACTION_START;
     key.seq = seq;
     key.tid = tid;
     key.filename = fqfn;
-    try {
-      write(key, EMPTY);
-    } catch (IOException ex) {
-      log.error(ex);
-      throw ex;
-    }
+    return logFileData(Collections.singletonList(new Pair<LogFileKey, LogFileValue>(key,
EMPTY)));
   }
 
 }

http://git-wip-us.apache.org/repos/asf/accumulo/blob/3b41d37e/server/src/main/java/org/apache/accumulo/server/tabletserver/log/TabletServerLogger.java
----------------------------------------------------------------------
diff --git a/server/src/main/java/org/apache/accumulo/server/tabletserver/log/TabletServerLogger.java
b/server/src/main/java/org/apache/accumulo/server/tabletserver/log/TabletServerLogger.java
index dd9d3fa..6ca1ad0 100644
--- a/server/src/main/java/org/apache/accumulo/server/tabletserver/log/TabletServerLogger.java
+++ b/server/src/main/java/org/apache/accumulo/server/tabletserver/log/TabletServerLogger.java
@@ -390,7 +390,7 @@ public class TabletServerLogger {
     int seq = write(commitSession, true, new Writer() {
       @Override
       public LoggerOperation write(DfsLogger logger, int ignored) throws Exception {
-        logger.minorCompactionFinished(walogSeq, commitSession.getLogId(), fullyQualifiedFileName);
+        logger.minorCompactionFinished(walogSeq, commitSession.getLogId(), fullyQualifiedFileName).await();
         return null;
       }
     });
@@ -406,7 +406,7 @@ public class TabletServerLogger {
     write(commitSession, false, new Writer() {
       @Override
       public LoggerOperation write(DfsLogger logger, int ignored) throws Exception {
-        logger.minorCompactionStarted(seq, commitSession.getLogId(), fullyQualifiedFileName);
+        logger.minorCompactionStarted(seq, commitSession.getLogId(), fullyQualifiedFileName).await();
         return null;
       }
     });


Mime
View raw message