The Tor Browser: comparison mobile/android/base/sync/synchronizer/RecordsChannel.java

--1:000000000000
+:2e732d1037bd
+/* 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 / file comparison

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

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