Background processing in Google App Engine (GAE) is possible thanks to the Task Queue API. This API allows developers to run asynchronous tasks through web posts. Until recently, the Task Queue API in AppScale (the open source implementation of GAE) was implemented using the GAE SDK implementation which consisted of a local thread dispatching requests and doing exponential backoff upon failures. For many reason this implementation was not correct in a distributed setting such as AppScale where there are multiple application servers running the same application. These reasons include not tracking state of tasks across nodes, keeping track of tasks names to prevent tasks fork bombs, and the proper load balancing of tasks across application servers and nodes. 

The AppScale implementation uses RabbitMQ as its engine for message passing in a distributed setting. We've found that RabbitMQ is well documented, has a relatively low memory footprint, and is very stable. The setup was simple for a distributed cluster with features such as acknowledgement of messages, high availability, and message durability. 

The figure above shows how tasks are distributed between nodes and application servers. Each machine may have multiple applications servers running either the same app or different apps. Within an application server (right box of the figure) we see there is a separate thread which listens for incoming tasks. When a message is received, this thread will then post to the local load balancer (orange arrow), where it will be balanced across one of the many application servers handing said application. If all application servers are unavailable, then the task will be re-enqueued and its retry number, which is stored in the header, will be incremented. Each time a failure occurs, the amount of time to backoff is a random number of seconds between 0 and 2^n, where n is the number of failures thus far.

Tasks names and state are stored in the datastore. There are three states a task can be in: running, completed, and error. Currently, only a single queue is used and features like rate limiting and task deferment are not implemented, but on the roadmap. Task deferment is not trivial as RabbitMQ does not support it delaying messages. This is the main feature currently lacking but is in the works.

The load balancer in AppScale is based on a combo of Nginx and HAProxy. HAProxy gives us the capability to queue up requests and serve them to the next available application server, while also doing health checks. Nginx gives the capability to do SSL and static file serving (spares the application server to handle only dynamic requests). 

The RabbitMQ Server runs on each node configured in a cluster. Any client that listens in on a particular queue will receive a message in a round robin fashion. If a client has a task and it fails, RabbitMQ will automatically distribute it to another client, providing fault tolerance. Our testing shows this works as promised. Thus far we are very happy with how RabbitMQ has been performing and encourage you to try it out for your message passing needs. 

This implementation will be in the upcoming AppScale 1.6 release. 

 

Here is a quick and dirty tutorial on how to do range queries in your favorite BigTable clone datastore (although Cassandra is a BigTable/Dynamo hybrid). Depending on how you set your keys you can do some fun stuff (like your own secondary indexes). Lets say you have the following keys in the same keyspace:

Google App Engine's Blobstore API is the primary method of storing large objects. This blog post talks about the API and how it is implemented in AppScale.

Let's step through the above workflow of how blobs are uploaded within AppScale. 

Hello Everyone,

Fantasm is a Google App Engine library which abstracts away TaskQueues by configuring work flows as finite state machines. Other comparable projects include the Pipeline API and the MapReduce API. Fantasm is great for processing large amounts of data which cannot be done normally due to timeout constraints.

Configuration
Hook fantasm up into your app.yaml file.

- url: /fantasm/.*
  script: fantasm/main.py
  login: admin

State machines are specified in a fsm.yaml file. In the file you give your state machine a name and individual states and transitions.

State machines have a single starting state and can have multiple final states.

Each state's execution ends with that state emitting a string to signify what the next state should be.

Make sure you use the full path of the action class. Example:

  - action: serverside.computations.InitialClass

Otherwise you'll get a ModuleNotFound error.

Communication Between States
"context" is passed from from one state to another and done so by arguments in the url. By default you should just pass strings and not send more context than can fit in a single POST request.

Communication Internal to a State
"obj" is passed from doing a continuation to the actual execution of a state. The "obj" is not serialized between states.

Advance Settings
It is possible to fork off a new process by calling context.fork(data=dictionary_of_new_context).

Be Careful
Make sure you have non-idempotent statements (statements with side effects, like updating an entity in the datastore) are done last. There probably still are some race conditions even if you do this, but they should be rare. Use locks via memcache to ensure there are none.

All states with continuation should also have final as a potential state. This is needed for the execute method for the case of no results in the query.

When is your job done?

Right now there is no way to get a callback or a trigger that a job is done.

Useful Iteration

The documentation on the Google Article Site does not talk about this method which shows up in the testing code. This method does not require you to use cursors as when using the continuation function. Here's how to count up all the accounts for your application if your application is really popular (otherwise it might be best to just use count() for on the query):

from fantasm.action import FSMAction, DatastoreContinuationFSMAction

class AllAccountsClass(DatastoreContinuationFSMAction):
  def getQuery(self, context, obj):
    return Accounts.all()

  def execute(self, context, obj):
    if not obj['result']:
      return None
    return "peraccount"

# Fan in here every X seconds

class CountAccountsClass(FSMAction):
  def execute(self, contexts, obj):
    """Transactionally update our batch counter"""
    batch_key = "num_accounts"

    def tx():
      batch = Batch.get_by_key_name(batch_key)
      if not batch:
        # For whatever reason it was not already created in previous state
        batch = BadgeBatch(key_name=batch_key)
        batch.put()
      batch.counter += len(contexts)
      batch.put()
    db.run_in_transaction(tx)

What Does Your State Machine Look Like?

See your state machine by going to the url: fantasm/graph/<state_machine_name>

It uses the google chart API.

Fanning In

You can have a state where you specify in the fsm.yaml file to accumulate context every X seconds (fan_in: X). In your execute function you'll have a contexts or list_of_contexts variable where you can get just the length (or more from each context if need be). Then inside a transaction increment some counter.

Code examples: http://code.google.com/p/userinfuser/wiki/Analytics
Fantasm Site: http://code.google.com/p/fantasm/w/list

Fantasm is developed by: http://www.vendasta.com/

There are times when you want to do remote logging or a remote API call. You may be okay with losing some updates for the tradeoff of adding little or no overhead for each call. For this case the asynchronous URL Fetch is your solution for Google App Engine. In the case I show below, the call is made and the returned result is never checked. See the GAE documentation on doing async calls which are started early in a request and then checked later in the request.

Some things to note when playing around with it is that it will not be truly asynchronous in the SDK version. In fact, if you use the code below, nothing will happen because in the SDK the call is actually made when you wait on the result. AppScale uses a modified SDK of the GAE and will suport asynchronous fetches in version 1.5.

Make sure to catch exceptions when put into production. The code below is pseudo code on GAE versus environments that allow threads.

GAE Method

from google.appengine.api import urlfetch
def url_async_post(url, argsdic):
    if isProductionGAE:
        # This will not work on the dev server for GAE, dev server must only use
        # synchronous calls
        rpc = urlfetch.create_rpc(deadline=10)
        urlfetch.make_fetch_call(rpc, url, payload=urllib.urlencode(argsdic), method=urlfetch.POST)
    else:
        raise

def call_remote(api_key, account, urlpath):
    argsdict = {"apikey":api_key,
               "accountid":account}
    url_async_post(urlpath, argsdict)
    return True
 
Threaded Method
This is how to do it for an environment that allows threads:

import threading
def my_threaded(callback=lambda *args, **kwargs: None, daemonic=True):
  """Decorate a function to run in its own thread and report the result
  by calling callback with it. Code yanked from stackoverflow.com"""
  def innerDecorator(func):
    def inner(*args, **kwargs):
      target = lambda: callback(func(*args, **kwargs))
      t = threading.Thread(target=target)
      t.setDaemon(daemonic)
      t.start()
    return inner
  return innerDecorator

@my_threaded()
def threaded_url_post(url, argsdic):
  self.url_post(url, argsdic)

def url_post(url, argsdic):
  import socket
  socket.setdefaulttimeout(5) #timeout value
  url_values = ""
  if argsdic:
    url_values = urllib.urlencode(argsdic)

  req = urllib2.Request(url, url_values)
  output = ""
  response = urllib2.urlopen(req)
  output = response.read()

  return output

One of Google's newest App Engine features is the Channel API which allows for the pushing of messages to a client's javascript code. This blog entry explains AppScale's scalable implementation which is built using ejabberd and strophejs.