mpire 2.6.0

进度条

假设我们想知道当前任务的状态：完成了多少任务，还有多久工作准备好？就像将progress_bar参数设置为True一样简单：

with WorkerPool(n_jobs=5) as pool:
    results = pool.map(time_consuming_function, range(10), progress_bar=True)

它会输出一个格式很好的tqdm进度条。如果您在笔记本中运行代码，它将自动切换到小部件。

MPIRE 还提供了一个仪表板，您需要为其安装额外的依赖项。有关更多信息，请参阅仪表板。

共享对象

注意：写时复制共享对象仅适用于 start 方法fork。对于线程，对象按原样共享。对于其他启动方法，共享对象为每个工作人员复制一次。

如果您想要在所有工作人员之间共享一个或多个对象，您可以使用 MPIRE 的写时复制 shared_objects选项。MPIRE 将只为每个工作人员传递这些对象一次，而无需复制/序列化。仅当您更改工作者函数中的对象时，它才会开始为该工作者复制它。

def time_consuming_function(some_object, x):
    time.sleep(1)  # Simulate that this function takes long to complete
    return ...

def main():
    some_object = ...
    with WorkerPool(n_jobs=5, shared_objects=some_object) as pool:
        results = pool.map(time_consuming_function, range(10), progress_bar=True)

有关详细信息，请参阅shared_objects。

工作器初始化

可以使用worker_init功能初始化工作程序。与worker_state一起，您可以加载模型，或建立数据库连接等：

def init(worker_state):
    # Load a big dataset or model and store it in a worker specific worker_state
    worker_state['dataset'] = ...
    worker_state['model'] = ...

def task(worker_state, idx):
    # Let the model predict a specific instance of the dataset
    return worker_state['model'].predict(worker_state['dataset'][idx])

with WorkerPool(n_jobs=5, use_worker_state=True) as pool:
    results = pool.map(task, range(10), worker_init=init)

同样，您可以使用worker_exit功能让 MPIRE 在 worker 终止时调用函数。你甚至可以让这个退出函数返回结果，这些结果可以在以后获得。有关更多信息，请参阅worker_init 和 worker_exit 部分。

员工洞察

当您的多处理设置没有按照您的意愿执行并且您不知道是什么原因导致它时，可以使用工作人员洞察功能。这将使您深入了解您的设置，但它不会分析您正在运行的功能（还有其他库）。相反，它会描述工作人员的启动时间、等待时间和工作时间。当提供worker init 和exit 函数时，它也会为这些函数计时。

也许您正在通过任务队列发送大量数据，这会增加等待时间。无论如何，您可以分别使用enable_insights标志和mpire.WorkerPool.get_insights函数启用和获取洞察：

with WorkerPool(n_jobs=5) as pool:
    results = pool.map(time_consuming_function, range(10), enable_insights=True)
    insights = pool.get_insights()

有关更详细的示例和预期输出，请参阅工作人员见解。

超时

可以为 target、 worker_init和worker_exit函数单独设置超时。当设置并达到超时时，它将抛出TimeoutError：

# Will raise TimeoutError, provided that the target function takes longer
# than half a second to complete
with WorkerPool(n_jobs=5) as pool:
    pool.map(time_consuming_function, range(10), task_timeout=0.5)

# Will raise TimeoutError, provided that the worker_init function takes longer
# than 3 seconds to complete or the worker_exit function takes longer than
# 150.5 seconds to complete
with WorkerPool(n_jobs=5) as pool:
    pool.map(time_consuming_function, range(10), worker_init=init, worker_exit=exit_,
             worker_init_timeout=3.0, worker_exit_timeout=150.5)

当使用线程作为启动方法时，MPIRE 将无法中断某些功能，例如time.sleep。

有关更多详细信息，请参阅超时。