Recommandation for using RQ with fork unsafe libraries

[stripedpumpkin]

Hello. I was wondering what the recommandation would be to use RQ with fork unsafe libraries such as PyMongo (https://pymongo.readthedocs.io/en/stable/faq.html#is-pymongo-fork-safe).

The use case would be the following: accessing a mongo database within the execution of an RQ job.

In the current case:

• either the client (== pool) is initialize pre-fork => that leads to errors,
• it is initialized post-fork, but then we are initializing a fresh connection with each job, which is not ideal as this limits scalability and adds a lot of overhead to each job execution.

(If it matters this setup would happen in the context of Django app.)

Would you recommend using another parallelization mechanism ? If so what would be the go to reference and how to set this up ?

Thanks for reading

[ccrvlh]

@stripedpumpkin I don't have an easy solution, I'm not particular familiar with how pymongo handles connection/pooling etc. Maybe more details on why grabbing connections from the pool wont work could be helpful to try to implement something fork-safe?

You could customize the Worker class though to try to make it work with pymongo. The forking happens here:

rq/rq/worker.py

Lines 795 to 807 in c5a7a6b

	def fork_work_horse(self, job: 'Job', queue: 'Queue'):
	"""Spawns a work horse to perform the actual work and passes it a job.
	"""
	child_pid = os.fork()
	os.environ['RQ_WORKER_ID'] = self.name
	os.environ['RQ_JOB_ID'] = job.id
	if child_pid == 0:
	os.setsid()
	self.main_work_horse(job, queue)
	os._exit(0) # just in case
	else:
	self._horse_pid = child_pid
	self.procline('Forked {0} at {1}'.format(child_pid, time.time()))

I think you could get way with doing something similar, but instead of forking, creating a process manually. You might need to pass some things that are on the parent process manually though: I found this article interesting when reading about this specific topic.

You would probably also need to handle this method:

rq/rq/worker.py

Lines 919 to 935 in c5a7a6b

	def main_work_horse(self, job: 'Job', queue: 'Queue'):
	"""This is the entry point of the newly spawned work horse."""
	# After fork()'ing, always assure we are generating random sequences
	# that are different from the worker.
	random.seed()
	self.setup_work_horse_signals()
	self._is_horse = True
	self.log = logger
	try:
	self.perform_job(job, queue)
	except: # noqa
	os._exit(1)
	# os._exit() is the way to exit from childs after a fork(), in
	# contrast to the regular sys.exit()
	os._exit(0)

To be honest, I'm not sure if that would work at all, but might be worth the shot.
I'm moving this to the discussions, seems more appropriate as it's not actually an issue when considering the way RQ works. Feel free to follow up on this, would be great to hear how you handle this case, could be helpful to others as well.

[ccrvlh]

@stripedpumpkin I completely missed the SimpleWorker which does not fork() and runs jobs in the same process, perhaps you can give that a try as well.