在 ipython/jupyter 笔记本中运行单元的新线程

Question

有时运行单个单元格需要很长时间，当它运行时，我想在同一个笔记本中编写和运行其他单元格，访问同一上下文中的变量。

是否有任何 ipython 魔法可以使用，当它被添加到单元格时，运行单元格将自动创建一个新线程并使用笔记本中的共享全局数据运行？

Answer 1

它可能不是一个答案，而是它的方向。我没有看到类似的东西，但我对这个也很感兴趣。

我目前的发现表明需要定义它是自己的自定义细胞魔法。很好的参考资料是文档中的 custom cell magic 部分 以及我会考虑的两个示例：

• memit：IPython https://gist.github.com/vene/3022718 的神奇内存使用基准测试
• 说明 Python 多线程与多处理：http://nathangrigg.net/2015/04/python-threading-vs-processes/

这两个链接都将代码包装在一个线程中。这可能是一个起点。

更新： github 上的 ngcm-tutorial 有后台作业类的描述

##github.com/jupyter/ngcm-tutorial/blob/master/Day-1/IPython%20Kernel/Background%20Jobs.ipynb
from IPython.lib import backgroundjobs as bg
jobs = bg.BackgroundJobManager()

def printfunc(interval=1, reps=5):
    for n in range(reps):
        time.sleep(interval)
        print('In the background... %i' % n)
        sys.stdout.flush()
    print('All done!')
    sys.stdout.flush()

jobs.new('printfunc(1,3)')
jobs.status()

更新 2： 另一种选择：

from IPython.display import display
from ipywidgets import IntProgress

import threading

class App(object):
    def __init__(self, nloops=2000):
        self.nloops = nloops
        self.pb = IntProgress(description='Thread loops', min=0, max=self.nloops)

    def start(self):
        display(self.pb)
        while self.pb.value < self.nloops:
            self.pb.value += 1 
        self.pb.color = 'red'

app = App(nloops=20000)

t = threading.Thread(target=app.start)

t.start()
#t.join()

Answer 2

这是我想出的一个小sn-p

def jobs_manager():
    from IPython.lib.backgroundjobs import BackgroundJobManager
    from IPython.core.magic import register_line_magic
    from IPython import get_ipython

    jobs = BackgroundJobManager()

    @register_line_magic
    def job(line):
        ip = get_ipython()
        jobs.new(line, ip.user_global_ns)

    return jobs

它使用 IPython 内置模块 IPython.lib.backgroundjobs 。所以代码小而简单，没有引入新的依赖。

我是这样使用的：

jobs = jobs_manager()

%job [fetch_url(_) for _ in urls]  # saves html file to disk
Starting job # 0 in a separate thread.

然后你可以通过以下方式监控状态：

jobs.status()

Running jobs:
1 : [fetch_url(_) for _ in urls]

Dead jobs:
0 : [fetch_url(_) for _ in urls]

如果作业失败，您可以检查堆栈跟踪

jobs.traceback(0)

没有办法杀死工作。所以我小心地使用了这个肮脏的黑客：

def kill_thread(thread):
    import ctypes

    id = thread.ident
    code = ctypes.pythonapi.PyThreadState_SetAsyncExc(
        ctypes.c_long(id),
        ctypes.py_object(SystemError)
    )
    if code == 0:
        raise ValueError('invalid thread id')
    elif code != 1:
        ctypes.pythonapi.PyThreadState_SetAsyncExc(
            ctypes.c_long(id),
            ctypes.c_long(0)
        )
        raise SystemError('PyThreadState_SetAsyncExc failed')

它在给定线程中引发SystemError。所以为了杀死我所做的工作

kill_thread(jobs.all[1])

为了杀死我所做的所有正在运行的作业

for thread in jobs.running:
    kill_thread(thread)

我喜欢将%job 与基于小部件的进度条https://github.com/alexanderkuk/log-progress 一起使用，如下所示：

%job [fetch_url(_) for _ in log_progress(urls, every=1)]

http://g.recordit.co/iZJsJm8BOL.gif

甚至可以使用%job 代替multiprocessing.TreadPool：

for chunk in get_chunks(urls, 3):
    %job [fetch_url(_) for _ in log_progress(chunk, every=1)]

http://g.recordit.co/oTVCwugZYk.gif

这段代码有一些明显的问题：

1. 您不能在%job 中使用任意代码。例如，不能有作业，也不能打印。所以我将它与将结果存储在硬盘上的例程一起使用

2. kill_thread 中的脏黑客有时不起作用。我认为这就是为什么IPython.lib.backgroundjobs 在设计上没有此功能的原因。如果线程正在执行诸如sleep 或read 之类的系统调用，则会忽略异常。

3. 它使用线程。 Python 有 GIL，所以 %job 不能用于一些需要 python 字节码的繁重计算