This may be alleviated by applying Asynchrony: the technique of passing the information for a long, complex computation to a second execution context, to either be alerted on completion (using a callback) or by repeatedly checking for updates (using polling) for a finite number of times.
Asynchrony within a platform is best expressed in terms of callbacks or promises. Asynchrony between platforms is easiest expressed in terms of polling. A complex web application with heavy front-end and back-end codebases requires both. In this blog post we give an overview of the various ways asynchrony was used in a project we worked on with Stroud Water Research Center: Model My Watershed.
The TaskModel is generic enough that it can is reused for a number of different interactions, including analyzing areas of interest and running geoprocessing models.
The blue arrow represents sending initial input, the orange arrow represents getting back a job id, the grey arrows represent polling for said job id, and the green arrow represents getting final results back.
However, if we let the web server do all the processing in addition to handling requests, it will overload and not be able to serve many clients. Thus, we add asynchrony on the back-end as well.
Celery provides nice APIs for waiting for tasks, retries, message routing, and error handling. Tasks are triggered by messages sent by the broker. Retries are handled by a simple decorator. Routing can be specified using exchange and routing_key parameters which ensure that subsequent tasks execute on the same worker in case they depend on external state not passed explicitly between tasks. Error handling is accomplished using the link_error parameter, which when added to a task chain will capture any exception raised by a constituent task.
Start a histogram geoprocessing job that for the given area of interest will return a count of the number of cells of each land cover type in the underlying raster.
Translate the results into a format expected by the front-end.
Start a MapShed data gathering geoprocessing job, that given an area of interest and a set of rasters, will return a grouped count of each combination of raster values.
Translate results into a format usable by the next step.
However, that's not quite right. Often times, the geoprocessing step in a Celery chain is by far the longest, and it would be inefficient to let it block an entire worker. You may have noticed that for each geoprocessing task in the above workflows, we had two Celery tasks: a start and a get or finish. That's because our geoprocessing service is asynchronous as well.
GeoTrellis is a high-performance geographic data processing engine that does all of the raster processing for Model My Watershed. It is made available to the main app as a web service, built on Spark JobServer (SJS). SJS offers a similar workflow as the one described above in Celery, comprised of an initial request with a payload which results in a job id, and then polling that job id until results are available.
where RasterLinesJoin, RasterGroupedCount, and RasterGroupedAverage are kinds of operations that could be called on the geoprocessing service. Since this is an internal service where we have perfect control over the environment, we assume that incoming configuration is aways valid.
Celery does not natively support conditional or dynamic task chains. We got around this by creating a list of tasks depending on input, and converting it to a Celery chain at the very end.
While the combine and collect_data tasks could be merged into one, that would have made a chord the penultimate task in the chain, whose reference would be returned instead of the final task save_job_results. We worked around this by splitting the chord's callback in two.
Even though the MapShed workflow shown above is complex with multiple specialized tasks, the error handling is very general. Regardless of the task that raises an exception or encounters an error, we simply want to log it to the database and return to the client. We assumed that given its general applicability, we could specify the error handler once on the parent chain and have it be catch errors raised anywhere in the constituent tasks. Unfortunately this is not true, and we must attach the error handler separately to every task in every subchain. We do this by defining the error handling task once, and passing it as a parameter to the function that creates all geoprocessing task chains.
Celery implements chords using a special chord_unlock task under the hood which repeatedly checks to see if all the tasks in its group have completed or not. If not, it continues retrying infinitely until they do, signified by the hard-coded, unconfigurable max_retries=None default value. For the most part this works fine as the tasks either complete successfully or fail.
However, for our MapShed workflow we use a group of chains, not a group of tasks. Since each chain returns the reference of the final task in the chain, chord_unlock tests to see only if the final task in the chain has completed. In case there is an error in any of the non-final tasks of the chain, the final task will never be executed at all, which results in chord_unlock being called an infinite amount of times.
Our strategy to address this was twofold: firstly we capture all exceptions in all non-final tasks in the chain and instead of raising them, pass them on to the next task as output. Each task checks if it receives an error, and if so passes it along immediately. If the final task receives an error, it raises an exception.
Secondly, even though the above should take care of all foreseeable errors, in order to make our application more robust we wrote a custom version of chord_unlock to patch into Celery on application start. The code is identical except for the new configuration parameter CELERY_CHORD_UNLOCK_MAX_RETRIES which ensures that our running application does not get stuck in an infinite loop.
Celery implements retries as an exception, to indicate to the worker that the current task should exit immediately and be rescheduled. Unfortunately, when we started capturing all exceptions for handling exceptions in chains above, we also captured the retry exceptions and broke the retry functionality. This was fixed by capturing retry separately and simply re-raising it.
Our deployment stack consists of a load balancer, a number of App VMs that handle web requests, a number of Worker VMs each running two Celery workers and an instance of SJS, and a number of cloud services on AWS. All Worker VMs share the same redis instance for brokering messages between various Celery workers. We use the blue-green deployment strategy for pushing builds to production. Assuming the current active stack is blue, we will spin up a new green stack of all the VMs, test it while it is dark, switch over to it once testing has passed, and then spin down the blue stack once it is no longer being used.
We use the Celery routing_key parameter to ensure that geoprocessing calls go to the same instance of SJS, since we cannot start a job in one instance and query for it in another. Other tasks can be routed to any available worker, and thus don't necessarily need to be routed explicitly. And in most cases this works correctly.
However, during deployment there are two stacks of Worker VMs, and thus Celery workers, that all share the same redis broker. This could lead to tasks from the newly spun dark stack to leak to production workers, clogging them up and reducing production capacity, but also leaking production tasks to workers in the dark stack, which may expect data in a different format and cause errors in production! To prevent this we explicitly route all tasks with a choose_worker method that gets a list of active workers in the current stack.
Making a large system asynchronous can be complex and have a number of side-effects that need to be addressed. It often times requires touching the entire stack from top to bottom, and thus needs solutions at every stage. Scaling out is much easier if a system is designed to be so from the beginning. When done right, the user experience is also significantly better.
The code for Model My Watershed is available on GitHub where all development happens in the open. Feel free to browse to find the actual implementation, which covers all edge cases and runs in production.
Have you worked with Celery, or Spark JobServer, or asynchronous web applications?
Share your experiences with us on Twitter @azavea.
Terence Tuhinanshu is a Software Engineer on the Civic Applications Team. Learn more about our software development services.