Cutlery/calibre
1
0
Fork 0
mirror of https://github.com/kovidgoyal/calibre.git synced 2025-07-09 03:04:10 -04:00
Code Issues Packages Projects Releases Wiki Activity

Refactor simple worker creation

Browse Source
This commit is contained in:
Kovid Goyal 2013-11-05 15:40:21 +05:30
parent 7a7cacc920
commit a09214d507
1 changed files with 25 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
src/calibre/utils/ipc/simple_worker.py
View File

@ -18,6 +18,7 @@ from calibre.utils.ipc import eintr_retry_call
from calibre.utils.ipc.launch import Worker from calibre.utils.ipc.launch import Worker
class WorkerError(Exception): class WorkerError(Exception):
def __init__(self, msg, orig_tb=''): def __init__(self, msg, orig_tb=''):
Exception.__init__(self, msg) Exception.__init__(self, msg)
self.orig_tb = orig_tb self.orig_tb = orig_tb
@ -83,6 +84,26 @@ def communicate(ans, worker, listener, args, timeout=300, heartbeat=None,
raise WorkerError('Worker failed', cw.res['tb']) raise WorkerError('Worker failed', cw.res['tb'])
ans['result'] = cw.res['result'] ans['result'] = cw.res['result']
def create_worker(env, priority='normal', cwd=None, func='main'):
address = arbitrary_address('AF_PIPE' if iswindows else 'AF_UNIX')
if iswindows and address[1] == ':':
address = address[2:]
auth_key = os.urandom(32)
listener = Listener(address=address, authkey=auth_key)
env = dict(env)
env.update({
'CALIBRE_WORKER_ADDRESS' :
hexlify(cPickle.dumps(listener.address, -1)),
'CALIBRE_WORKER_KEY' : hexlify(auth_key),
'CALIBRE_SIMPLE_WORKER':
'calibre.utils.ipc.simple_worker:%s' % func,
})
w = Worker(env)
w(cwd=cwd, priority=priority)
return listener, w
def fork_job(mod_name, func_name, args=(), kwargs={}, timeout=300, # seconds def fork_job(mod_name, func_name, args=(), kwargs={}, timeout=300, # seconds
cwd=None, priority='normal', env={}, no_output=False, heartbeat=None, cwd=None, priority='normal', env={}, no_output=False, heartbeat=None,
abort=None, module_is_source_code=False): abort=None, module_is_source_code=False):
@ -136,24 +157,7 @@ def fork_job(mod_name, func_name, args=(), kwargs={}, timeout=300, # seconds
''' '''
ans = {'result':None, 'stdout_stderr':None} ans = {'result':None, 'stdout_stderr':None}
listener, w = create_worker(env, priority, cwd)
address = arbitrary_address('AF_PIPE' if iswindows else 'AF_UNIX')
if iswindows and address[1] == ':':
address = address[2:]
auth_key = os.urandom(32)
listener = Listener(address=address, authkey=auth_key)
env = dict(env)
env.update({
'CALIBRE_WORKER_ADDRESS' :
hexlify(cPickle.dumps(listener.address, -1)),
'CALIBRE_WORKER_KEY' : hexlify(auth_key),
'CALIBRE_SIMPLE_WORKER':
'calibre.utils.ipc.simple_worker:main',
})
w = Worker(env)
w(cwd=cwd, priority=priority)
try: try:
communicate(ans, w, listener, (mod_name, func_name, args, kwargs, communicate(ans, w, listener, (mod_name, func_name, args, kwargs,
module_is_source_code), timeout=timeout, heartbeat=heartbeat, module_is_source_code), timeout=timeout, heartbeat=heartbeat,