Task fanout - how to bulk add Tasks to the Queue - more than 5

Question

I am using a task (queueing-task) to queue multiple others tasks — fanout. When I try to use Queue.add with task argument being a list of Task instances with more than 5 element's and in transaction… I get this error.

JointException: taskqueue.DatastoreError caused by: 
    <class 'google.appengine.api.datastore_errors.BadRequestError'> 
    Too many messages, maximum allowed 5

Is there another way to queue more than 5 tasks in a transaction?

Or...

Maybe I don't need a transaction, cause:

1. I don't care if any of those tasks get queued twice anyway, and
2. if queueing will fail for any of them, then the whole queueing-task will be re run.

So tell me how do I queue more than 5 tasks in a transaction or tell me to not use transaction cause I don't really need one.

Answer 1

One solution close to solving your problem is to add one transactional task that fans-out the remaining tasks. Just add the one fan-out task in your existing transaction.

Unless there is a business logic reason to do so, do not re-run a task that has already run. Preventing tasks from being re-inserted (i.e. duplicated) is straightforward and saves resources. Your fan-out task will basically look like:

class FanOutTask(webapp.RequestHandler):
  def get(self):
    name = self.request.get('name')
    params = deserialize(self.request.get('params'))

    try:
      task_params = params.get('stuff')
      taskqueue.add(url='/worker/1', name=name + '-1', params=task_params)
    except TaskAlreadyExistsError:
      pass

    try:
      task_params = params.get('more')
      taskqueue.add(url='/worker/2', name=name + '-2', params=task_params)
    except TaskAlreadyExistsError:
      pass

Adding the fan-out task transactionally ensures it is enqueued. Errors resulting from the task already being run get caught and ignored, other errors cause the fan-out task to re-run. With this pattern you can insert many sub-tasks pretty easily.