Threads, locks and fork()

[pelme]

I ran into issues with the approach used in celerytest:

Launching a celery worker in a new thread, which then forks of new processes causes problems. This happens because there might be locks held in other threads when celery fork()s. This happens commonly if the logging module is used since it uses thread locks. Mixing threads, thread locks and fork()s turns out to be a bad idea and unreliable: http://bugs.python.org/issue6721

I now instead start a celery worker by just launching a celery worker in a subprocess, which seems to work reliable.

[thauk-copperleaf]

I now instead start a celery worker by just launching a celery worker in a subprocess, which seems to work reliable.

Can you describe this in more detail? Do you have a PR or some code you can share?

[pelme]

The code currently lives in my project and depends on Django, pytest and pytest-services. I have not made any effort factoring it out into something reusable. I am not sure that is really useful since the principle is quite simple: just run celery in a subprocess. :) You need to consider how to configure celery properly in the new process, you can use a specific test config module and/or pass settings via env variables.

FWIW I have been running thousands of testruns with this code/approach and it seems to work fine. :)

Here is the relevant pytest fixture from my project that launches celery, I hope it may help even though a lot of context and other surrounding code are missing:

@pytest.fixture
def celery_worker(request, redis_server, watcher_getter, tmp_media_root):
    env = os.environ.copy()

    # Port cannot be given as an empty string, it must be 0 if no port is specified
    db_attrs = settings.DATABASES['default'].copy()
    db_attrs['PORT'] = db_attrs['PORT'] or 0
    env['DATABASE_URL'] = 'postgres://{USER}:{PASSWORD}@{HOST}:{PORT}/{NAME}'.format(**db_attrs)
    env['REDIS_URL'] = settings.REDIS_URL
    env['MEDIA_ROOT'] = tmp_media_root

    watcher_getter('celery', ['worker', '-A', 'myproject.conf.celery_worker:app', '--concurrency=1'],
                         kwargs={'env': env}, checker=lambda: True)

[thauk-copperleaf]

Thanks, @pelme. I had a look at this but I wasn't able to get it to work. I first noticed that watcher_getter doesn't exist, but I found it in pytest-services as a fixture, so I installed that. But the plugin isn't loaded and various fixtures (like watcher_getter) aren't found when pytest runs. (I also think you mean to have return watcher_getter('celery'... instead of just watcher_getter('celery'..., or am I wrong?).

[pelme]

Yes, the code is incomplete and does not work as is. You would probably be better of by writing the code from scratch for your own project.

My point with the code was to show the concept and idea. You need to adapt that in your own project. The important parts to consider are:

• Start celery in a subprocess via a call to subprocess.Popen() (or similar). Before your test start. In my case that is handled via pytest fixtures and watcher_getter.
• Pass any configuration variables as env variables via the env kwarg to Popen().
• Terminate the celery subprocess at the end of the test or test session. I restart celery in every test case. This may or may not work for you.

[thauk-copperleaf]

Thanks @pelme. I've got something working now like that. My spawned processes aren't being killed properly at the end of my test, so I've got a bug there, but otherwise I've been able to successfully do my performance testing of our celery tasks.

I agree with you that forking from a thread is not the correct approach. Really, celery should be providing the testing infrastructure.