Example worker restarting itself on 3.9 about faktory_worker_python HOT 6 CLOSED

Does it work if you pass use_threads=True to the Worker class?

I don't have access to an M1 Mac so I can't debug this. Python support in 3.9.1 is experimental only. Are you able to test on one of the Python 3.10 alphas?

from faktory_worker_python.

Does it work if you pass use_threads=True to the Worker class?

Yes it did.

I don't have access to an M1 Mac so I can't debug this. Python support in 3.9.1 is experimental only. Are you able to test on one of the Python 3.10 alphas?

I'm installing everything via brew. I will check if it's available and let you know.

from faktory_worker_python.

I haven't had the time to install 3.10 alphas but brew has 3.8 that runs on the M1. One of my tests I managed to get this error:

RuntimeError: An attempt has been made to start a new process before the current process has finished its bootstrapping phase. This probably means that you are not using fork to start your child processes and you have forgotten to use the proper idiom in the main module: if name == 'main': freeze_support() ... The "freeze_support()" line can be omitted if the program is not going to be frozen to produce an executable.

So I modified it the sample code to:

from faktory import Worker

def your_function(x, y):
    print("your_function")
    return x + y


if __name__ == '__main__':
  w = Worker(queues=['default'], concurrency=1, use_threads=False)
  w.register('test', your_function)
  print("after_register")
  w.run()  # runs until control-c or worker shutdown from Faktory web UI

And it works now (the sample code, but sadly not mine yet). Does this give you any ideas on where the issue could lie or where I can investigate further?

Thank you.

from faktory_worker_python.

Forgot to mention that adding that if makes the sample code also work under 3.9.

from faktory_worker_python.

If I understand correctly, adding if __name__ == '__main__': fixes the sample code, but not your production code? If so, I'll update the readme.

Could you try adding:

multiprocessing.set_start_method("fork", force=True)

and see if that helps at all?

from faktory_worker_python.

multiprocessing.set_start_method("fork", force=True)

I changed that and some jobs run, others fail with a BrokenPipe [ErrNo 32] error (which is shown in Faktory's UI). This is the code that I'm currently running (as a test case):

import sys
from faktory import Worker
import time
import logging
import os
import multiprocessing

multiprocessing.set_start_method("fork", force=True)

print("after import")

def myfunc(phone, amount, params = {}):
  time.sleep(1)
  print("myfunc")

if __name__ == '__main__':
  print("entering main")
  w = Worker(queues=['foo'], concurrency=3)
  w.register('FooJob', myfunc)
  w.run()

I tried changing the concurrency to 1, but the effect was the same. Sometimes it runs, other times I get the BrokenPipe error.

from faktory_worker_python.