我有一个多处理的Web服务器,其流程永无止境,我想在实时环境中检查整个项目的代码覆盖率(不仅仅是测试).

问题是,由于进程永远不会结束,我没有一个设置cov.start() cov.stop() cov.save()钩子的好地方.

因此,我想到产生一个线程,在无限循环中将保存并组合覆盖数据然后休眠一段时间,但是这种方法不起作用,覆盖报告似乎是空的,除了睡眠线.

我很乐意收到有关如何获取我的代码的任何想法,或任何关于为什么我的想法不起作用的建议.这是我的代码片段:

import coverage
cov = coverage.Coverage()
import time
import threading
import os

class CoverageThread(threading.Thread):
    _kill_now = False
    _sleep_time = 2

@classmethod
def exit_gracefully(cls):
    cls._kill_now = True

def sleep_some_time(self):
    time.sleep(CoverageThread._sleep_time)

def run(self):
    while True:
        cov.start()
        self.sleep_some_time()
        cov.stop()
        if os.path.exists('.coverage'):
            cov.combine()
        cov.save()
        if self._kill_now:
            break
    cov.stop()
    if os.path.exists('.coverage'):
        cov.combine()
    cov.save()
    cov.html_report(directory="coverage_report_data.html")
    print "End of the program. I was killed gracefully :)"

我正在使用Python 3.6,并且正在尝试跟随下面网站上的第一个示例(下面的完整代码)并且我得到以下错误:https: //docs.python.org/3.6/library/multiprocessing. HTML

错误信息: AttributeError: module '__main__' has no attribute '__spec__'

完整示例代码:

from multiprocessing import Pool

def f(x):
    return x*x

if __name__ == '__main__':
    with Pool(5) as p:
        print(p.map(f, [1, 2, 3]))

我尝试使用Google搜索并搜索Stack Overflow,但我只发现了另一个此错误的情况并且没有答案.

我正在使用 torch、gunicorn 和 Flask 创建一个应该使用 CUDA 的推理服务。为了减少资源需求，我使用了gunicorn的预加载选项，因此模型在工作进程之间共享。然而，这会导致 CUDA 出现问题。下面的代码片段显示了一个最小的重现示例：

from flask import Flask, request
import torch

app = Flask('dummy')

model = torch.rand(500)
model = model.to('cuda:0')


@app.route('/', methods=['POST'])
def f():
    data = request.get_json()
    x = torch.rand((data['number'], 500))
    x = x.to('cuda:0')
    res = x * model
    return {
        "result": res.sum().item()
    }

使用 启动服务器CUDA_VISIBLE_DEVICES=1 gunicorn -w 3 -b $HOST_IP:8080 --preload run_server:app可以让服务成功启动。然而，一旦执行第一个请求（curl -X POST -d '{"number": 1}'），工作人员就会抛出以下错误：

[2022-06-28 09:42:00,378] ERROR in app: Exception on / [POST]
Traceback (most recent call …

我正在尝试运行一些独立的计算(尽管从相同的数据中读取).我的代码在Ubuntu上运行时有效,但在Windows(Windows Server 2012 R2)上没有,我得到错误:

'module' object has no attribute ...

当我尝试使用时multiprocessing.Pool(它出现在内核控制台中,而不是笔记本本身的输出)

(而且我已经犯了在创建池后定义函数的错误,我也纠正了它,这不是问题).

甚至在最简单的例子中也会发生这种情况:

from multiprocessing import Pool
def f(x):
    return x**2
pool = Pool(4)
for res in pool.map(f,range(20)):
    print res

我知道它需要能够导入模块(我不知道在笔记本中工作时它是如何工作的),我听说过IPython.Parallel,但我一直无法找到任何文档或示例.

任何解决方案/替代品都是最受欢迎的.

我有很多泡菜文件。目前我循环阅读它们，但这需要很多时间。我想加快速度，但不知道如何做到这一点。

多处理不起作用，因为为了将数据从子进程传输到主进程数据需要序列化（腌制）和反序列化。

由于 GIL，使用线程也无济于事。

我认为解决方案是一些用 C 编写的库，它需要读取文件列表，然后运行多个线程（没有 GIL）。周围有这样的东西吗？

更新 回答您的问题：

• 文件是用于机器学习的数据处理的部分产品
• 有pandas.Series对象，但预先不知道 dtype
• 我想要很多文件，因为我们想轻松选择任何子集
• 我想要许多较小的文件而不是一个大文件，因为一个大文件的反序列​​化需要更多的内存（在某个时间点我们已经序列化了字符串和反序列化对象）
• 文件的大小可能会有很大差异
• 我使用 python 3.7 所以我相信它实际上是 cPickle
• 使用pickle非常灵活，因为我不必担心底层类型——我可以保存任何东西

我试图在python中编写一个简单的工作证明nonce-finder.

def proof_of_work(b, nBytes):
    nonce = 0
    # while the first nBytes of hash(b + nonce) are not 0
    while sha256(b + uint2bytes(nonce))[:nBytes] != bytes(nBytes):
        nonce = nonce + 1
    return nonce

现在我尝试进行多处理,因此它可以使用所有CPU内核并更快地找到nonce.我的想法是multiprocessing.Pool多次使用和执行函数proof_of_work,传递两个参数num_of_cpus_running,this_cpu_id如下所示:

def proof_of_work(b, nBytes, num_of_cpus_running, this_cpu_id):
    nonce = this_cpu_id
    while sha256(b + uint2bytes(nonce))[:nBytes] != bytes(nBytes):
        nonce = nonce + num_of_cpus_running
    return nonce

所以,如果有4个核心,每个核心将计算这样的随机数:

core 0: 0, 4, 8, 16, 32 ...
core 1: 1, 5, 9, 17, 33 ...
core 2: 2, …

我试图在具有64个CPU的CentOS 7机器上同时运行几个TensorFlow会话.我的同事报告说他可以使用以下两个代码块在他的机器上使用4个内核生成并行加速:

mnist.py

import numpy as np
import input_data
from PIL import Image
import tensorflow as tf
import time


def main(randint):
    print 'Set new seed:', randint
    np.random.seed(randint)
    tf.set_random_seed(randint)
    mnist = input_data.read_data_sets("MNIST_data/", one_hot=True)

    # Setting up the softmax architecture
    x = tf.placeholder("float", [None, 784])
    W = tf.Variable(tf.zeros([784, 10]))
    b = tf.Variable(tf.zeros([10]))
    y = tf.nn.softmax(tf.matmul(x, W) + b)

    # Setting up the cost function
    y_ = tf.placeholder("float", [None, 10])
    cross_entropy = -tf.reduce_sum(y_*tf.log(y))
    train_step = tf.train.GradientDescentOptimizer(0.01).minimize(cross_entropy)

    # Initialization 
    init = tf.initialize_all_variables()
    sess = …

我有一个使用该multiprocessing包的短代码,并在我的本地机器上正常工作.

当我上传到AWS Lambda那里并运行时,我收到以下错误(stacktrace trimmed):

[Errno 38] Function not implemented: OSError
Traceback (most recent call last):
  File "/var/task/recorder.py", line 41, in record
    pool = multiprocessing.Pool(10)
  File "/usr/lib64/python2.7/multiprocessing/__init__.py", line 232, in Pool
    return Pool(processes, initializer, initargs, maxtasksperchild)
  File "/usr/lib64/python2.7/multiprocessing/pool.py", line 138, in __init__
    self._setup_queues()
  File "/usr/lib64/python2.7/multiprocessing/pool.py", line 234, in _setup_queues
    self._inqueue = SimpleQueue()
  File "/usr/lib64/python2.7/multiprocessing/queues.py", line 354, in __init__
    self._rlock = Lock()
  File "/usr/lib64/python2.7/multiprocessing/synchronize.py", line 147, in __init__
    SemLock.__init__(self, SEMAPHORE, 1, 1)
  File "/usr/lib64/python2.7/multiprocessing/synchronize.py", line 75, in __init__ …

我有一个使用多处理(特别是joblib)的功能来加速使用多个核心的慢速例程.它很棒; 没有问题.

我有一个使用多处理的测试套件(目前只是multiprocessing.Pool()系统,但可以将其更改为joblib)以独立运行每个模块的测试功能.它很棒; 没有问题.

问题是我现在已经将多处理功能集成到模块的测试套件中,以便池进程运行多处理功能.我想这样做,以便内部函数知道它已经被多处理并且不会自行调整更多的分支.目前内部进程有时会挂起,但即使它没有,显然在已经并行的例程中没有多处理的好处.

我可以想到几种方法(使用锁定文件,设置某种全局变量等)来确定我们所处的状态,但我想知道是否有一些标准方法可以解决这个问题(在PY多处理中)或在joblib).如果它只适用于PY3,那就没问题,但显然在2.7或更低的情况下工作的解决方案会更好.谢谢!

我有一个foo消耗大量内存的功能,我希望并行运行几个实例.

假设我有一个带有4个物理内核的CPU,每个内核都有两个逻辑内核.

我的系统有足够的内存来容纳4个foo并行实例,但不是8个.而且,由于这8个内核中有4个是逻辑的,所以我也不希望使用所有8个内核可以提供超出使用4个物理内核的大量增益只要.

所以,我想运行foo在4个物理内核只.换句话说,我想确保执行multiprocessing.Pool(4)(4是由于内存限制而在此计算机上可以容纳的函数的最大并发运行次数)将作业调度到四个物理内核(例如,不是两个物理核心及其两个逻辑后代的组合.

如何在python中做到这一点？

编辑:

我之前使用过代码示例,multiprocessing但我与库无关,所以为了避免混淆,我删除了它.