@@ -0,0 +1,3320 @@
+/*%0A * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0%0A * (the %E2%80%9CLicense%E2%80%9D). You may not use this work except in compliance with the License, which is%0A * available at www.apache.org/licenses/LICENSE-2.0%0A *%0A * This software is distributed on an %22AS IS%22 basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,%0A * either express or implied, as more fully set forth in the License.%0A *%0A * See the NOTICE file distributed with this work for information regarding copyright ownership.%0A */%0A%0Apackage alluxio.master.journal;%0A%0Aimport alluxio.Constants;%0Aimport alluxio.proto.journal.Journal.JournalEntry;%0A%0Aimport com.google.common.base.Preconditions;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport java.io.IOException;%0Aimport java.util.concurrent.ConcurrentLinkedQueue;%0Aimport java.util.concurrent.atomic.AtomicLong;%0Aimport java.util.concurrent.locks.ReentrantLock;%0A%0A/**%0A *%0A */%0Apublic final class AsyncJournalWriter %7B%0A  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);%0A%0A  private final JournalWriter mJournalWriter;%0A  private final ConcurrentLinkedQueue%3CJournalEntry%3E mQueue;%0A  private final AtomicLong mCounter;%0A  private final AtomicLong mFlushCounter;%0A%0A  /**%0A   * Use a %7B@link ReentrantLock%7D to guard the journal writing. Using the fairness policy seems to%0A   * result in better throughput.%0A   */%0A  private final ReentrantLock mFlushLock = new ReentrantLock(true);%0A%0A  /**%0A   * Creates a %7B@link AsyncJournalWriter%7D.%0A   *%0A   * @param journalWriter the %7B@link JournalWriter%7D to use for writing%0A   */%0A  public AsyncJournalWriter(JournalWriter journalWriter) %7B%0A    mJournalWriter = Preconditions.checkNotNull(journalWriter);%0A    mQueue = new ConcurrentLinkedQueue%3C%3E();%0A    mCounter = new AtomicLong(0);%0A    mFlushCounter = new AtomicLong(0);%0A  %7D%0A%0A  /**%0A   * Appends a %7B@link JournalEntry%7D for writing to the journal.%0A   *%0A   * @param entry the %7B@link JournalEntry%7D to append%0A   * @return a counter for the entry, for flushing%0A   */%0A  public long appendEntry(JournalEntry entry) %7B%0A    mCounter.incrementAndGet();%0A    mQueue.offer(entry);%0A    return mCounter.get();%0A  %7D%0A%0A  /**%0A   * Flushes and waits until the specified counter is flushed to the journal. If the specified%0A   * counter is already flushed, this is essentially a no-op.%0A   *%0A   * @param counter the counter to flush%0A   * @throws IOException if an error occurs in flushing the journal%0A   */%0A  public void flush(final long counter) throws IOException %7B%0A    if (counter %3C= mFlushCounter.get()) %7B%0A      return;%0A    %7D%0A    // Using reentrant lock, since it seems to result in higher throughput than using 'synchronized'%0A    mFlushLock.lock();%0A    try %7B%0A      long flushCounter = mFlushCounter.get();%0A      if (counter %3C= flushCounter) %7B%0A        return;%0A      %7D%0A      while (counter %3E flushCounter) %7B%0A        for (;;) %7B%0A          JournalEntry entry = mQueue.poll();%0A          if (entry == null) %7B%0A            // No more entries in the queue.%0A            break;%0A          %7D%0A          mJournalWriter.getEntryOutputStream().writeEntry(entry);%0A          flushCounter++;%0A%0A          // TODO(gpang): limit the amount of time for writing.%0A        %7D%0A      %7D%0A      mJournalWriter.getEntryOutputStream().flush();%0A      mFlushCounter.set(flushCounter);%0A    %7D finally %7B%0A      mFlushLock.unlock();%0A    %7D%0A  %7D%0A%0A%7D%0A

Implement an async journal writer/flusher