Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Already on GitHub? Sign in to your account

Chain of too many subtasks throws recursion error #1078

Closed
dnathe4th opened this Issue Nov 20, 2012 · 4 comments

Comments

Projects
None yet
4 participants

The following is my test case:

def text_users(userinfo, message, number):
    user_text_subtasks = []
    for user in userinfo:
        user_text_subtasks.append(text_individual_user.s(user, message, number))

    results = {
        "success": [],
        "error": [],
        "invalid_number": []
    }

    if len(user_text_subtasks):
        user_text_subtasks[0] = text_individual_user.s(results, userinfo[0], message, number)

    user_text_subtasks.append(text_callback.s(message))

    chain(*user_text_subtasks).apply_async()

@celery.task
def text_individual_user(result, user, message, number):
    # text them
    # return updated result

If the userinfo object I pass in is 75 items long (resulting in 76 subtasks in the chain because of the collection function at the end) the process executes fine. However if userinfo is 100 items long (101 in the subtask chain) I get a recursion error.

The following is the stack trace:

Traceback (most recent call last):

File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/celery-3.0.12-py2.7.egg/celery/task/trace.py", line 224, in trace_task
  R = retval = fun(*args, **kwargs)
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/celery-3.0.12-py2.7.egg/celery/task/trace.py", line 406, in __protected_call__
  return self.run(*args, **kwargs)
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/example-0.1-py2.7-linux-x86_64.egg/example/views/text.py", line 220, in text_users
  chain(*driver_text_subtasks).apply_async()
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/celery-3.0.12-py2.7.egg/celery/canvas.py", line 150, in apply_async
  return self.type.apply_async(args, kwargs, **options)
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/celery-3.0.12-py2.7.egg/celery/app/builtins.py", line 245, in apply_async
  tasks[0].apply_async()
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/celery-3.0.12-py2.7.egg/celery/canvas.py", line 150, in apply_async
  return self.type.apply_async(args, kwargs, **options)
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/celery-3.0.12-py2.7.egg/celery/app/task.py", line 478, in apply_async
  **options)
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/celery-3.0.12-py2.7.egg/celery/app/amqp.py", line 229, in publish_task
  **kwargs)
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/kombu-2.4.8-py2.7.egg/kombu/messaging.py", line 150, in publish
  compression, headers)
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/kombu-2.4.8-py2.7.egg/kombu/messaging.py", line 201, in _prepare
  body) = encode(body, serializer=serializer)
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/kombu-2.4.8-py2.7.egg/kombu/serialization.py", line 162, in encode
  payload = encoder(data)
File "/home/example/api/master/23/env/local/lib/python2.7/site-packages/kombu-2.4.8-py2.7.egg/kombu/serialization.py", line 341, in dumps
  return dumper(obj, protocol=pickle_protocol)
RuntimeError: maximum recursion depth exceeded while calling a Python object
Owner

ask commented Nov 21, 2012

I don't think this is caused by the number of tasks in the chain, it rather seems like you have a data structure that cannot be pickled because it's recursing (probably objects pointing to each other in a loop)

Here is a revised test case anyone can run

def chain_test(x):
    tasks = []

    for i in xrange(x):
        tasks.append(single_task.s(i))

    result = {
        "success": [],
        "error": []
    }
    tasks[0] = single_task.s(result, 0)

    tasks.append(final.s())

    chain(*tasks).apply_async()


@celery.task
def single_task(result, i):
    print result
    result['success'].append(i)
    return result


@celery.task
def final(result):
    print "FINAL: ", result

Calling chain_test with anything less than 96 works fine for me and prints out the [non self-referencing result] every task. Calling it with 96 or more immediately bails out with the following stacktrace:

Traceback (most recent call last):
  File "/home/example/api/env/local/lib/python2.7/site-packages/celery-3.0.12-py2.7.egg/celery/task/trace.py", line 256, in trace_task
    for callback in task_request.callbacks or []]
  File "/home/example/api/env/local/lib/python2.7/site-packages/celery-3.0.12-py2.7.egg/celery/canvas.py", line 398, in subtask
    return varies.clone()
  File "/home/example/api/env/local/lib/python2.7/site-packages/celery-3.0.12-py2.7.egg/celery/canvas.py", line 124, in clone
    'kwargs': kwargs, 'options': deepcopy(opts),
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 190, in deepcopy
    y = _reconstruct(x, rv, 1, memo)
  File "/usr/lib/python2.7/copy.py", line 328, in _reconstruct
    args = deepcopy(args, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 237, in _deepcopy_tuple
    y.append(deepcopy(a, memo))
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 257, in _deepcopy_dict
    y[deepcopy(key, memo)] = deepcopy(value, memo)
  File "/usr/lib/python2.7/copy.py", line 163, in deepcopy
    y = copier(x, memo)
  File "/usr/lib/python2.7/copy.py", line 230, in _deepcopy_list
    y.append(deepcopy(a, memo))
  File "...", line -1, in [rest of traceback truncated]
RuntimeError: maximum recursion depth exceeded while calling a Python object
Owner

ask commented Nov 22, 2012

Thanks, the test case does indeed reproduce the issue.

This is actually a limitation with pickle and I don't think this can be easily fixed without reinventing the data structure (A chain is a tree of Signature objects linked by their Options). The patch above fixes the deepcopy case, but it's still a problem for pickle. It does work with json, but the multiprocessing worker uses pickle anyway to send it to child processes.

You can make it work by increasing the recursion limit:

sys.setrecursionlimit(10000)

Do you really need to chain 97 tasks together? Remember that you will have to send the complete chain in the first message, so as tasks are executed the chain list becomes smaller.

If it does make sense to have chains of 100 and more objects, I guess the only sensible fix
would be to 'flatten' the tree so that instead of having:

{'task_id' 1001,
 'link': {
     'args': [1],
     'link': {
         'args': [2],
         'link': {
             ...
         },
    },
}

we would use a list:

{'task_id': 1002,
 'link_chain': [
     {'args': [1]},
     {'args': [2]},
     ...,
 ]} 

But this would not be backwards compatible, so it would have to be part of 3.2 (maybe 3.1 if someone has the time to prepare a patch asap)

st3fan commented Apr 28, 2014

I'm having the same issue with celery==3.1.11. The sys.setrecursionlimit trick works, but feels bad.

@ask ask closed this in d3e1282 Oct 31, 2015

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment