You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@accumulo.apache.org by ec...@apache.org on 2014/01/10 22:37:48 UTC
[4/6] git commit: ACCUMULO-2172 wait for minor compaction flags to be
flushed to the WAL
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/master
Commit: 3b41d37ed27b5dc8724b427ed6970b083464bd94
Parents: 4822b13
Author: Eric Newton <er...@gmail.com>
Authored: Fri Jan 10 16:27:01 2014 -0500
Committer: Eric Newton <er...@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;
}
});