From 295b64e6eaa5c5037748a34f1f3043c8bfa84617 Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Tue, 10 Jan 2012 16:24:10 +0530 Subject: [PATCH 01/22] Add framework for running a python function in a child process using the standard call-return function semantics --- src/calibre/utils/ipc/simple_worker.py | 148 ++++++++++++++++++++++ src/calibre/utils/ipc/worker.py | 10 +- src/calibre/utils/pyconsole/controller.py | 3 +- 3 files changed, 157 insertions(+), 4 deletions(-) create mode 100644 src/calibre/utils/ipc/simple_worker.py diff --git a/src/calibre/utils/ipc/simple_worker.py b/src/calibre/utils/ipc/simple_worker.py new file mode 100644 index 0000000000..171bd2dc7e --- /dev/null +++ b/src/calibre/utils/ipc/simple_worker.py @@ -0,0 +1,148 @@ +#!/usr/bin/env python +# vim:fileencoding=UTF-8:ts=4:sw=4:sta:et:sts=4:ai +from __future__ import (unicode_literals, division, absolute_import, + print_function) + +__license__ = 'GPL v3' +__copyright__ = '2012, Kovid Goyal ' +__docformat__ = 'restructuredtext en' + +import os, cPickle, traceback, time, importlib +from binascii import hexlify, unhexlify +from multiprocessing.connection import Listener, arbitrary_address, Client +from threading import Thread +from contextlib import closing + +from calibre.constants import iswindows +from calibre.utils.ipc.launch import Worker + +class WorkerError(Exception): + def __init__(self, msg, orig_tb=''): + Exception.__init__(self, msg) + self.org_tb = orig_tb + +class ConnectedWorker(Thread): + + def __init__(self, listener, args): + Thread.__init__(self) + self.daemon = True + + self.listener = listener + self.args = args + self.accepted = False + self.tb = None + self.res = None + + def run(self): + conn = tb = None + for i in range(2): + # On OS X an EINTR can interrupt the accept() call + try: + conn = self.listener.accept() + break + except: + tb = traceback.format_exc() + pass + if conn is None: + self.tb = tb + return + self.accepted = True + with closing(conn): + try: + try: + conn.send(self.args) + except: + # Maybe an EINTR + conn.send(self.args) + try: + self.res = conn.recv() + except: + # Maybe an EINTR + self.res = conn.recv() + except: + self.tb = traceback.format_exc() + +def communicate(ans, worker, listener, args, timeout=300): + cw = ConnectedWorker(listener, args) + cw.start() + st = time.time() + while worker.is_alive and cw.is_alive(): + cw.join(0.01) + delta = time.time() - st + if not cw.accepted and delta > min(10, timeout): + break + if delta > timeout: + raise WorkerError('Worker appears to have hung') + if not cw.accepted: + if not cw.tb: + raise WorkerError('Failed to connect to worker process') + raise WorkerError('Failed to connect to worker process', cw.tb) + + if cw.tb: + raise WorkerError('Failed to communicate with worker process') + if cw.res.get('tb', None): + raise WorkerError('Worker failed', cw.res['tb']) + ans['result'] = cw.res['result'] + +def fork_job(mod_name, func_name, args=(), kwargs={}, timeout=300, # seconds + cwd=None, priority='normal', env={}, no_output=False): + + ans = {'result':None, 'stdout_stderr':None} + + 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: + communicate(ans, w, listener, (mod_name, func_name, args, kwargs), + timeout=timeout) + finally: + w.kill() + if no_output: + try: + os.remove(w.log_path) + except: + pass + if not no_output: + ans['stdout_stderr'] = w.log_path + return ans + +def main(): + # The entry point for the simple worker process + address = cPickle.loads(unhexlify(os.environ['CALIBRE_WORKER_ADDRESS'])) + key = unhexlify(os.environ['CALIBRE_WORKER_KEY']) + with closing(Client(address, authkey=key)) as conn: + try: + args = conn.recv() + except: + # Maybe EINTR + args = conn.recv() + try: + mod, func, args, kwargs = args + mod = importlib.import_module(mod) + func = getattr(mod, func) + res = {'result':func(*args, **kwargs)} + except: + res = {'tb': traceback.format_exc()} + + try: + conn.send(res) + except: + # Maybe EINTR + conn.send(res) + + + diff --git a/src/calibre/utils/ipc/worker.py b/src/calibre/utils/ipc/worker.py index 1381abbc2d..6d3019bbdd 100644 --- a/src/calibre/utils/ipc/worker.py +++ b/src/calibre/utils/ipc/worker.py @@ -167,9 +167,13 @@ def main(): # so launch the gui as usual from calibre.gui2.main import main as gui_main return gui_main(['calibre']) - if 'CALIBRE_LAUNCH_INTERPRETER' in os.environ: - from calibre.utils.pyconsole.interpreter import main - return main() + csw = os.environ.get('CALIBRE_SIMPLE_WORKER', None) + if csw: + mod, _, func = csw.partition(':') + mod = importlib.import_module(mod) + func = getattr(mod, func) + func() + return address = cPickle.loads(unhexlify(os.environ['CALIBRE_WORKER_ADDRESS'])) key = unhexlify(os.environ['CALIBRE_WORKER_KEY']) resultf = unhexlify(os.environ['CALIBRE_WORKER_RESULT']) diff --git a/src/calibre/utils/pyconsole/controller.py b/src/calibre/utils/pyconsole/controller.py index d372cb4ebc..59f13a7022 100644 --- a/src/calibre/utils/pyconsole/controller.py +++ b/src/calibre/utils/pyconsole/controller.py @@ -39,7 +39,8 @@ class Controller(QThread): authkey=self.auth_key, backlog=4) self.env = { - 'CALIBRE_LAUNCH_INTERPRETER': '1', + 'CALIBRE_SIMPLE_WORKER': + 'calibre.utils.pyconsole.interpreter:main', 'CALIBRE_WORKER_ADDRESS': hexlify(cPickle.dumps(self.listener.address, -1)), 'CALIBRE_WORKER_KEY': hexlify(self.auth_key) From bb9112fbf028b1a53ac0925a8f7494256cbd4536 Mon Sep 17 00:00:00 2001 From: Kovid Goyal Date: Tue, 10 Jan 2012 22:37:18 +0530 Subject: [PATCH 02/22] EPUB CFI: Video bookmarking works. Also fix use of coffeescript compiler in threads by deelgating compiling to a worker process, much slower, but doesn't segfault --- src/TestHTTPServer.py | 132 +++++++++++++++++ .../ebooks/oeb/display/test-cfi/birds.mp4 | Bin 1048576 -> 0 bytes .../ebooks/oeb/display/test-cfi/birds.webm | Bin 0 -> 1324945 bytes .../oeb/display/test-cfi/cfi-test.coffee | 2 + .../ebooks/oeb/display/test-cfi/index.html | 7 +- src/calibre/utils/serve_coffee.py | 138 ++++++++++-------- 6 files changed, 214 insertions(+), 65 deletions(-) create mode 100644 src/TestHTTPServer.py delete mode 100644 src/calibre/ebooks/oeb/display/test-cfi/birds.mp4 create mode 100644 src/calibre/ebooks/oeb/display/test-cfi/birds.webm diff --git a/src/TestHTTPServer.py b/src/TestHTTPServer.py new file mode 100644 index 0000000000..35077fe41c --- /dev/null +++ b/src/TestHTTPServer.py @@ -0,0 +1,132 @@ +#!/usr/bin/env python +# vim:fileencoding=UTF-8:ts=4:sw=4:sta:et:sts=4:ai +from __future__ import (unicode_literals, division, absolute_import, + print_function) + +__license__ = 'GPL v3' +__copyright__ = '2011, Kovid Goyal ' +__docformat__ = 'restructuredtext en' + +import os, io +from SimpleHTTPServer import SimpleHTTPRequestHandler + +class HTTPRequestHandler(SimpleHTTPRequestHandler): + ''' + Handle Range headers, as browsers insist on using range for