Skip to content
New issue

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

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Worker pool handles spawned process #655

Draft
wants to merge 7 commits into
base: master
Choose a base branch
from

Conversation

chromium7
Copy link
Contributor

Fix #651
Ref #650

I can't reproduce the issue in #650, but I did get the same errors in #651
Seems like its an issue with multiprocessing start method in certain OS, as was answered in this stack overflow post: https://stackoverflow.com/questions/46908035/apps-arent-loaded-yet-exception-occurs-when-using-multi-processing-in-django

This solution builds on top of @jackkinsella answer, and can be updated further to handle his specific issue.

)


def run_django_worker(*args: Any, **kwargs: Any) -> None:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

FWIW The version of this function we're using in production was.

def run_django_worker(*args: Any, **kwargs: Any) -> None:
    alias = "default"

    try:
        connections[alias].close()
    except Exception:
        logger.error(f"Worker could not get a connection to the database with alias {alias}")

    run_worker(*args, **kwargs)

The exception catching and logging was transient and didn't stop us from using a 16 worker queue to process a few 100k events per day

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The django.setup() solution is likely better (but I don't understand it so can't offer feedback)

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I just checked production (Heroku - unix) and get_start_method() was "fork". I know that simply executing run_worker() will not work in that environment due to the psycog issues with the DB connection from pre-fork.

(Maybe this explains the failing CI test?)

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, I just realized that db connections are also closed pre-fork in the rqworker management command.

# Close any opened DB connection before any fork
reset_db_connections()

I called django.setup() because the spawned process doesn't load the installed apps at all. This works fine when I manually tested it, but fails in unit test. Probably its because django.setup() is not setting up the project in testing environment which leads to the process setting up its own DB instance.

I'm currently using this django code for parallel test runner as a reference. I'll try to update the PR in a day or two.
https://github.com/django/django/blob/53719d6b5b745dd99b1ab9315afb242f706ebbf1/django/test/runner.py#L424-L432

if get_start_method() == 'spawn':
django.setup()

run_worker(*args, **kwargs)
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similar to @jackkinsella's comment, we need to use our own run_worker command that calls reset_db_connections before running the original run_worker command:

def reset_db_connections():

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

RQ WorkerPool is not loading models at all
3 participants