Getting Started with the Asynchronous API

The best way to get started with the asynchronous API is to not use the API at all. Simply porting your application to the driver may provide enough of a performance boost without having to use the asynchronous API directly. Once you have ported your application you will also have a better understanding of what parts of your application will benefit the most.

Assuming you have already ported the application and still need more performance: Choose a single processing flow that is not meeting its performance objectives, or is causing the application to stall waiting for responses from MongoDB. When looking at various flows choose the most common or worst performing flow. Re-write that flow using the suggestions below and then reevaluate the applications performance. If its performance is still not satisfactory, update another flow, reevaluate, and repeat until the application is performing as needed.

Asynchronous Options

The asynchronous API is broken into two flavors. One is based on Futures and the other is based on Callbacks. We will cover futures first and then the callback versions but before we get into the details it is important to realize that mixing and matching the asynchronous APIs is perfectly acceptable and even encouraged. It is even possible to mix the asynchronous APIs with the traditional methods in high performance applications, if done carefully. More on that later.

Asynchronous, lock free processing is hard. There is no doubt that in order to achieve optimal performance the developer needs to think about how to best achieve the desired result. The good news is that when done right the improvements in performance can be extreme.

There is one feature of the driver that is often over looked but can help immensely with asynchronous programming. The MongoClient interface has a method: asSerializedClient(). The method returns what looks like another normal instance of a MongoClient. It has one very special property. All of the requests to the server that are derived from that MongoClient will be sent via a single logical connection to the cluster and so will be serialized and processed in the same order on the server as they are issued on the client. Note that ReadPreferences and sharding can change the server used but all requests to a single server will be processed in the same order as the client issues them. We have intentionally made the asSerializedClient() and MongoDatabase and MongoCollection objects as lightweight as possible to encourage their use. There are many problems that are extremely difficult to solve without the asSerializedClient() guarantees.

Examples Using the Future API.

String line;
MongoCollection collection = null;

BufferedReader reader = new BufferedReader(new StringReader("f"));
while( (line = reader.readLine()) != null )  {
    Document query = queryForLine(line);
    Document update = updateForLine(line);
    
    long result = collection.update(query, update);
    processResult(result);
}

String line;
MongoCollection collection = null;
BlockingQueue<Future<Long>> pendingResults = new ArrayBlockingQueue<Future<Long>>(1000);

BufferedReader reader = new BufferedReader(new StringReader("f"));
while ((line = reader.readLine()) != null) {
    Document query = queryForLine(line);
    Document update = updateForLine(line);

    Future<Long> result = collection.updateAsync(query, update);

    while (!pendingResults.offer(result)) {
        Future<Long> delayed = pendingResults.take();

        processResult(delayed);
    }
}

Future<Long> delayed;
while ((delayed = pendingResults.poll()) != null) {
    processResult(delayed);
}

Error Handling with the Future API.

Any error encountered while processing the request will be thrown when the get() method of the future is called. The exception is wrapped in an ExecutionException. The original exception can be retrieved from the ExecutionException via the ExecutionException.getCause() method.

Example Using the Callback API.

Lets return to our File processing example. Using the Callback interface is a fairly minor change to the original processing. Using the anonymous inner class obfuscates the following example but the intent is still fairly clear.

String line;
MongoCollection collection = null;

BufferedReader reader = new BufferedReader(new StringReader("f"));
while ((line = reader.readLine()) != null) {
    Document query = queryForLine(line);
    Document update = updateForLine(line);

    collection.updateAsync(new Callback<Long>() {
        public void callback(Long result) {
            processResult(result);
        }
        public void exception(Throwable thrown) {
            // Handle error.
        }
    }, query, update);
}

For a LambdaCallback using a Java 8 lambda expression this becomes even more succinct.

String line;
MongoCollection collection = null;

BufferedReader reader = new BufferedReader(new StringReader("f"));
while ((line = reader.readLine()) != null) {
    Document query = queryForLine(line);
    Document update = updateForLine(line);

    collection.updateAsync((e, result) -> {
            if( e ) {
                // Handle error.
            } else {
                processResult(result)
            }
        }, query, update);
}

The anonymous inner class can be moved to a nested or top level class but then the link between the result of one request leading to the next stage of processing is lost. To recover the codes flow we have successfully used the following strategy:

• For each processing flow define the series of stages that must be completed. Each stage ends with an asynchronous request to MongoDB.
• Create a series of methods in the class responsible for the processing named something like: <flow_name><stage_name>(...). The arguments to each stage are the results of the previous stage's asynchronous request to MongoDB and any other state that needs to be carried forward. If your project uses automated code reformatting and sorting name the stages with a sortable prefix, e.g., flowName1Prepare, flowName2Compose, etc.
• Create simple nested and/or top level classes to act as the callback and bridge from one stage to the next. An instance of the callback will be used with the asynchronous request at the end of one stage and its callback(...) method will simply invoke the method for the next stage.

The combination of the method naming and ordering clearly communicates the processing flow to future developers. By keeping all logic out of the Callbacks we ensure they remain clear and concise bridge components.

Error Handling with the Callback API.

Any errors encountered while processing the request will be explicitly given to the Callback via the exception(Throwable) method. One of the virtues of the Callback method or handling asynchronous requests is that it forces the developer to explicitly deal with failures. A common mechanism is to retry the request.

Retries with Callbacks

The following base class can be used to intelligently retry requests to the MongoDB server. The logic for actually retrying the request is left to the derived class via a retry(Throwable).

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.allanbank.mongodb.Callback;

/**
 * AbstractRetryCallbackHandler provides a base class for {@link Callback}
 * instances that will retry the operation.
 * 
 * @param <V>
 *            The type of the callback's result.
 */
public abstract class AbstractRetryCallbackHandler<V> implements Callback<V> {

    /** The maximum amount of time to pause between retries. */
    public static final long MAX_PAUSE_MS = TimeUnit.MINUTES.toMillis(1);

    /** The executor for retries. */
    private final ScheduledExecutorService myExecutor;

    /** The maximum number of times to retry. */
    private final int myMaximumRetries;

    /** The number of milliseconds to wait before attempting another retry. */
    private long myNextPauseMS;

    /** The number of times we have attemtped to retry. */
    private int myRetries;

    /**
     * Creates a new AbstractRetryCallbackHandler.
     * 
     * @param retries
     *            The number of times we have attemtped to retry.
     * @param executor
     *            he executor for retries.
     */
    public AbstractRetryCallbackHandler(final int retries,
            final ScheduledExecutorService executor) {
        myMaximumRetries = retries;
        myExecutor = executor;
        myRetries = 0;
        myNextPauseMS = 1;
    }

    /**
     * Called when all of the retry attempts have been exhausted.
     * 
     * @param thrown
     *            The error causing the retry.
     */
    public abstract void dead(final Throwable thrown);

    /**
     * {@inheritDoc}
     * <p>
     * Overridden to retry the request after a suitable delay.
     * </p>
     */
    @Override
    public void exception(final Throwable thrown) {
        myRetries += 1;
        if (myRetries < myMaximumRetries) {

            myExecutor.schedule(new Runnable() {

                @Override
                public void run() {
                    retry(thrown);  
                    
                }
            }, myNextPauseMS, TimeUnit.MILLISECONDS);

            myNextPauseMS = Math.min(myNextPauseMS * 2, MAX_PAUSE_MS);
        }
        else {
            dead(thrown);
        }
    }

    /**
     * Called to attempt a retry.
     * 
     * @param thrown
     *            The error causing the retry.
     */
    public abstract void retry(final Throwable thrown);
}

Note: It is not always possible to simply repeat requests to the MongoDB server. Under some circumstances it is possible to use a findAndModify command (or two) to look for the case that the initial request was successful (and potentially, also the case it was not successful) and return the appropriate results.