The Tor Browser: mobile/android/base/sync/synchronizer/RecordsChannel.java@fc2d59ddac77 (annotated)

mobile/android/base/sync/synchronizer/RecordsChannel.java@fc2d59ddac77 (annotated)

mobile/android/base/sync/synchronizer/RecordsChannel.java

Wed, 31 Dec 2014 07:22:50 +0100

author: Michael Schloh von Bennewitz <michael@schloh.com>
date: Wed, 31 Dec 2014 07:22:50 +0100
branch: TOR_BUG_3246
changeset 4: fc2d59ddac77
permissions: -rw-r--r--

Correct previous dual key logic pending first delivery installment.

 /* This Source Code Form is subject to the terms of the Mozilla Public
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 package org.mozilla.gecko.sync.synchronizer;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.mozilla.gecko.background.common.log.Logger;
 import org.mozilla.gecko.sync.ThreadPool;
 import org.mozilla.gecko.sync.repositories.InvalidSessionTransitionException;
 import org.mozilla.gecko.sync.repositories.NoStoreDelegateException;
 import org.mozilla.gecko.sync.repositories.RepositorySession;
 import org.mozilla.gecko.sync.repositories.delegates.DeferredRepositorySessionBeginDelegate;
 import org.mozilla.gecko.sync.repositories.delegates.DeferredRepositorySessionStoreDelegate;
 import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionBeginDelegate;
 import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionFetchRecordsDelegate;
 import org.mozilla.gecko.sync.repositories.delegates.RepositorySessionStoreDelegate;
 import org.mozilla.gecko.sync.repositories.domain.Record;
 /**
  * Pulls records from `source`, applying them to `sink`.
  * Notifies its delegate of errors and completion.
  *
  * All stores (initiated by a fetch) must have been completed before storeDone
  * is invoked on the sink. This is to avoid the existing stored items being
  * considered as the total set, with onStoreCompleted being called when they're
  * done:
  *
  *   store(A) store(B)
  *   store(C) storeDone()
  *   store(A) finishes. Store job begins.
  *   store(C) finishes. Store job begins.
  *   storeDone() finishes.
  *   Storing of A complete.
  *   Storing of C complete.
  *   We're done! Call onStoreCompleted.
  *   store(B) finishes... uh oh.
  *
  * In other words, storeDone must be gated on the synchronous invocation of every store.
  *
  * Similarly, we require that every store callback have returned before onStoreCompleted is invoked.
  *
  * This whole set of guarantees should be achievable thusly:
  *
  * * The fetch process must run in a single thread, and invoke store()
  *   synchronously. After processing every incoming record, storeDone is called,
  *   setting a flag.
  *   If the fetch cannot be implicitly queued, it must be explicitly queued.
  *   In this implementation, we assume that fetch callbacks are strictly ordered in this way.
  *
  * * The store process must be (implicitly or explicitly) queued. When the
  *   queue empties, the consumer checks the storeDone flag. If it's set, and the
  *   queue is exhausted, invoke onStoreCompleted.
  *
  * RecordsChannel exists to enforce this ordering of operations.
  *
  * @author rnewman
  *
  */
 public class RecordsChannel implements
   RepositorySessionFetchRecordsDelegate,
   RepositorySessionStoreDelegate,
   RecordsConsumerDelegate,
   RepositorySessionBeginDelegate {
   private static final String LOG_TAG = "RecordsChannel";
   public RepositorySession source;
   public RepositorySession sink;
   private RecordsChannelDelegate delegate;
   private long fetchEnd = -1;
   protected final AtomicInteger numFetched = new AtomicInteger();
   protected final AtomicInteger numFetchFailed = new AtomicInteger();
   protected final AtomicInteger numStored = new AtomicInteger();
   protected final AtomicInteger numStoreFailed = new AtomicInteger();
   public RecordsChannel(RepositorySession source, RepositorySession sink, RecordsChannelDelegate delegate) {
     this.source    = source;
     this.sink      = sink;
     this.delegate  = delegate;
   }
   /*
    * We push fetched records into a queue.
    * A separate thread is waiting for us to notify it of work to do.
    * When we tell it to stop, it'll stop. We do that when the fetch
    * is completed.
    * When it stops, we tell the sink that there are no more records,
    * and wait for the sink to tell us that storing is done.
    * Then we notify our delegate of completion.
    */
   private RecordConsumer consumer;
   private boolean waitingForQueueDone = false;
   private ConcurrentLinkedQueue<Record> toProcess = new ConcurrentLinkedQueue<Record>();
   @Override
   public ConcurrentLinkedQueue<Record> getQueue() {
     return toProcess;
   }
   protected boolean isReady() {
     return source.isActive() && sink.isActive();
   }
   /**
    * Get the number of records fetched so far.
    *
    * @return number of fetches.
    */
   public int getFetchCount() {
     return numFetched.get();
   }
   /**
    * Get the number of fetch failures recorded so far.
    *
    * @return number of fetch failures.
    */
   public int getFetchFailureCount() {
     return numFetchFailed.get();
   }
   /**
    * Get the number of store attempts (successful or not) so far.
    *
    * @return number of stores attempted.
    */
   public int getStoreCount() {
     return numStored.get();
   }
   /**
    * Get the number of store failures recorded so far.
    *
    * @return number of store failures.
    */
   public int getStoreFailureCount() {
     return numStoreFailed.get();
   }
   /**
    * Start records flowing through the channel.
    */
   public void flow() {
     if (!isReady()) {
       RepositorySession failed = source;
       if (source.isActive()) {
         failed = sink;
       }
       this.delegate.onFlowBeginFailed(this, new SessionNotBegunException(failed));
       return;
     }
     if (!source.dataAvailable()) {
       Logger.info(LOG_TAG, "No data available: short-circuiting flow from source " + source);
       long now = System.currentTimeMillis();
       this.delegate.onFlowCompleted(this, now, now);
       return;
     }
     sink.setStoreDelegate(this);
     numFetched.set(0);
     numFetchFailed.set(0);
     numStored.set(0);
     numStoreFailed.set(0);
     // Start a consumer thread.
     this.consumer = new ConcurrentRecordConsumer(this);
     ThreadPool.run(this.consumer);
     waitingForQueueDone = true;
     source.fetchSince(source.getLastSyncTimestamp(), this);
   }
   /**
    * Begin both sessions, invoking flow() when done.
    * @throws InvalidSessionTransitionException
    */
   public void beginAndFlow() throws InvalidSessionTransitionException {
     Logger.trace(LOG_TAG, "Beginning source.");
     source.begin(this);
   }
   @Override
   public void store(Record record) {
     numStored.incrementAndGet();
     try {
       sink.store(record);
     } catch (NoStoreDelegateException e) {
       Logger.error(LOG_TAG, "Got NoStoreDelegateException in RecordsChannel.store(). This should not occur. Aborting.", e);
       delegate.onFlowStoreFailed(this, e, record.guid);
     }
   }
   @Override
   public void onFetchFailed(Exception ex, Record record) {
     Logger.warn(LOG_TAG, "onFetchFailed. Calling for immediate stop.", ex);
     numFetchFailed.incrementAndGet();
     this.consumer.halt();
     delegate.onFlowFetchFailed(this, ex);
   }
   @Override
   public void onFetchedRecord(Record record) {
     numFetched.incrementAndGet();
     this.toProcess.add(record);
     this.consumer.doNotify();
   }
   @Override
   public void onFetchCompleted(final long fetchEnd) {
     Logger.trace(LOG_TAG, "onFetchCompleted. Stopping consumer once stores are done.");
     Logger.trace(LOG_TAG, "Fetch timestamp is " + fetchEnd);
     this.fetchEnd = fetchEnd;
     this.consumer.queueFilled();
   }
   @Override
   public void onRecordStoreFailed(Exception ex, String recordGuid) {
     Logger.trace(LOG_TAG, "Failed to store record with guid " + recordGuid);
     numStoreFailed.incrementAndGet();
     this.consumer.stored();
     delegate.onFlowStoreFailed(this, ex, recordGuid);
     // TODO: abort?
   }
   @Override
   public void onRecordStoreSucceeded(String guid) {
     Logger.trace(LOG_TAG, "Stored record with guid " + guid);
     this.consumer.stored();
   }
   @Override
   public void consumerIsDone(boolean allRecordsQueued) {
     Logger.trace(LOG_TAG, "Consumer is done. Are we waiting for it? " + waitingForQueueDone);
     if (waitingForQueueDone) {
       waitingForQueueDone = false;
       this.sink.storeDone();                 // Now we'll be waiting for onStoreCompleted.
     }
   }
   @Override
   public void onStoreCompleted(long storeEnd) {
     Logger.trace(LOG_TAG, "onStoreCompleted. Notifying delegate of onFlowCompleted. " +
                           "Fetch end is " + fetchEnd + ", store end is " + storeEnd);
     // TODO: synchronize on consumer callback?
     delegate.onFlowCompleted(this, fetchEnd, storeEnd);
   }
   @Override
   public void onBeginFailed(Exception ex) {
     delegate.onFlowBeginFailed(this, ex);
   }
   @Override
   public void onBeginSucceeded(RepositorySession session) {
     if (session == source) {
       Logger.trace(LOG_TAG, "Source session began. Beginning sink session.");
       try {
         sink.begin(this);
       } catch (InvalidSessionTransitionException e) {
         onBeginFailed(e);
         return;
       }
     }
     if (session == sink) {
       Logger.trace(LOG_TAG, "Sink session began. Beginning flow.");
       this.flow();
       return;
     }
     // TODO: error!
   }
   @Override
   public RepositorySessionStoreDelegate deferredStoreDelegate(final ExecutorService executor) {
     return new DeferredRepositorySessionStoreDelegate(this, executor);
   }
   @Override
   public RepositorySessionBeginDelegate deferredBeginDelegate(final ExecutorService executor) {
     return new DeferredRepositorySessionBeginDelegate(this, executor);
   }
   @Override
   public RepositorySessionFetchRecordsDelegate deferredFetchDelegate(ExecutorService executor) {
     // Lie outright. We know that all of our fetch methods are safe.
     return this;
   }
 }

The Tor Browser / annotate

mobile/android/base/sync/synchronizer/RecordsChannel.java@fc2d59ddac77 (annotated)

mobile/android/base/sync/synchronizer/RecordsChannel.java