标签: python-multiprocessing

以下代码不打印"here".问题是什么？我在我的两台机器上测试了它(Windows 7,Ubuntu 12.10)和 http://www.compileonline.com/execute_python_online.php 它并不是"here"在所有情况下打印.

from multiprocessing import Queue, Process


def runLang(que):
    print "start"
    myDict=dict()
    for i in xrange(10000):
        myDict[i]=i
    que.put(myDict)
    print "finish"


def run(fileToAnalyze):
    que=Queue()
    processList=[]
    dicList=[]
    langs= ["chi","eng"]
    for lang in langs:
        p=Process(target=runLang,args=(que,))
        processList.append(p)
        p.start()

    for p1 in processList:
        p1.join()

    print "here"

    for _ in xrange(len(langs)):
        item=que.get()
        print item
        dicList.append(item)

if __name__=="__main__":
    processList = []
    for fileToAnalyse in ["abc.txt","def.txt"]:
        p=Process(target=run,args=(fileToAnalyse,))
        processList.append(p)
        p.start()
    for p1 in processList:
        p1.join()

我在Windows 7和Windows 10笔记本电脑上都运行了一些代码:

def project(filename):
    **Do Something Here**


if __name__ == '__main__':
    pool = Pool(processes=4)
    results = [pool.apply_async(project, args=(filename,)) for filename in filenamelist]
    output = [p.get() for p in results]
    print output

两台计算机都是双核/ 4线程,因此它们都应该可以很好地运行4个进程.我遇到的问题是,当我在Windows 10上运行代码时,我会运行4个python进程,但是它们使用0%的cpu并且它们不会输出任何内容,这与Windows 7个数据库不同,它将全部运行4个线程和完美的工作.

如果我不使用多处理,代码在Windows 10笔记本电脑上工作正常,所以问题必须与此相关.使用Python的多处理在Windows 10中不起作用吗？我顺便在两台机器上运行Python 2.7.

[编辑]:Windows 7 pc处理器是i5-650,Windows 10笔记本处理器是i3-2370M

[更新]:我将笔记本电脑还原到Windows 8.1,完全相同的代码按预期运行,这绝对是Windows 10的问题.

[编辑]:我用来生成的filenamelist方法如下,但这在Windows 7上运行正常.

def get_unfinished_files(indir, outdir):
    filenamelist = []
    for filename in os.listdir(indir):
        if filename not in os.listdir(outdir) and filename.endswith('.png'):
            filenamelist.append(filename)
    return filenamelist

我有一些代码使用multiprocessing.Pool来分叉工作者并并行执行任务.我正在尝试找到运行此代码的单元测试的正确方法.

注意我并不是要尝试并行测试串行代码测试用例,我知道像nose支持的软件包.

如果我编写一个测试所述并行代码的测试函数,并尝试使用nose运行测试:nosetests tests/test_function.py非并行测试正确执行,但多处理尝试fork时并行测试失败,因为main不可导入:

      File "C:\python-2.7.10.amd64\lib\multiprocessing\forking.py", line 488, in prepare
assert main_name not in sys.modules, main_name
AssertionError: __main__
    assert main_name not in sys.modules, main_name
AssertionError:     _assert main_name not in sys.modules, main_name
_main__AssertionError
: __main__

在我终止任务之前,这只会重复.如果我修改tests/test_function.py为包括以下内容,我可以成功运行测试:

if __name__ == '__main__':
    import nose
    nose.main()

然后执行 python tests\test_function.py

那么,与单元测试包集成的"正确"方法是什么(不一定是鼻子)？

Environ:Windows 7 64位上的Python 2.7.10 amd64

Win 7、x64、Python 2.7.12

在下面的代码中，我将设置一些池进程以通过该multiprocessing.Pool.map()方法进行简单的乘法运算。输出数据收集在List_1.

注意：这是对我的实际代码的精简简化。实际应用中涉及到多个列表，个个都很大。

import multiprocessing
import numpy as np

def createLists(branches):

    firstList = branches[:] * node

    return firstList


def init_process(lNodes):

    global node
    node = lNodes
    print 'Starting', multiprocessing.current_process().name


if __name__ == '__main__':

    mgr = multiprocessing.Manager()
    nodes = mgr.list()
    pool_size = multiprocessing.cpu_count()

    branches = [i for i in range(1, 21)]
    lNodes = 10
    splitBranches = np.array_split(branches, int(len(branches)/pool_size))

    pool = multiprocessing.Pool(processes=pool_size, initializer=init_process, initargs=[lNodes])
    myList_1 = pool.map(createLists, splitBranches)

    pool.close() 
    pool.join()  

我现在添加一个额外的计算createLists()并尝试传回两个列表。

import multiprocessing
import numpy …

我需要帮助才能理解multiprocessing.Queue.我遇到的问题是,queue.get(...)与调用queue.put(...)和队列缓冲区(deque)相比,获得结果是非常有趣的.

这种泄漏抽象导致我调查队列的内部.它直截了当的源代码只是指向deque实现,这似乎也很简单,我不能用它来解释我所看到的行为.另外我读到Queue使用管道,但我似乎无法在源代码中找到它.

我把它归结为一个重现问题的最小例子,我在下面指定了一个可能的输出.

import threading
import multiprocessing
import queue

q = None
def enqueue(item):
    global q
    if q is None:
        q = multiprocessing.Queue()
        process = threading.Thread(target=worker, args=(q,))  # or multiprocessing.Process Doesn't matter
        process.start()
    q.put(item)
    print(f'len putted item: {len(item)}. qsize: {q.qsize()}. buffer len: {len(q._buffer)}')


def worker(local_queue):
    while True:
        try:
            while True:  # get all items
                item = local_queue.get(block=False)
                print(f'len got item: {len(item)}. qsize: {q.qsize()}. buffer len: {len(q._buffer)}')
        except queue.Empty:
            print('empty')


if …

我正在尝试使用 concurrent.futures 模块将一个需要很长时间的进程拆分为多个进程。附上下面的代码

主功能：

with concurrent.futures.ProcessPoolExecutor() as executor:
for idx, score in zip([idx for idx in range(dataframe.shape[0])],executor.map(get_max_fuzzy_score,[dataframe[idx:idx+1] for idx in range(dataframe.shape[0])])):
    print('processing '+str(idx+1)+' of '+str(dataframe.shape[0]+1))
    dataframe['max_row_score'].iloc[idx] = score

get_max_fuzzy_score 函数：

def get_max_fuzzy_score(picklepath_or_list, df):
import numpy as np
extracted_text_columns = list(df.filter(regex='extracted_text').columns)
data_list = [df[data].iloc[0] for data in extracted_text_columns if not df[data].isnull().values.any()]
try:
    size = len(picklepath_or_list)
    section_snippet_list = picklepath_or_list
except:
    section_snippet_list = pickle.load(open(picklepath_or_list,'rb'))
scores = []
for section_snippet in section_snippet_list:
    for data in data_list:
        scores.append(fuzz.partial_ratio(data,section_snippet))
score = max(scores)

return score

该函数采用几列的值，并从先前构建的列表中返回最大模糊分数。

这是我得到的错误： …

我的主要问题是在这里发布。由于还没有人给出解决方案，我决定找到一种解决方法。我正在寻找一种使用python 代码限制 python 脚本 CPU 使用率（不是优先级，而是 CPU 内核数）的方法。我知道我可以使用多处理库（池等）来做到这一点，但我不是使用多处理运行它的人。所以，我不知道该怎么做。而且我也可以通过终端来做到这一点，但是这个脚本是由另一个脚本导入的。不幸的是，我没有通过终端调用它的奢侈。

tl; dr：如何限制由另一个脚本导入的 python 脚本的CPU 使用率（核心数），我什至不知道为什么它并行运行，而不是通过终端运行它。请检查下面的代码片段。

导致问题的代码片段：

from sklearn.datasets import load_digits
from sklearn.decomposition import IncrementalPCA
import numpy as np

X, _ = load_digits(return_X_y=True)

#Copy-paste and increase the size of the dataset to see the behavior at htop.
for _ in range(8):
    X = np.vstack((X, X))

print(X.shape)

transformer = IncrementalPCA(n_components=7, batch_size=200)

#PARTIAL FIT RUNS IN PARALLEL! GOD WHY?
---------------------------------------
transformer.partial_fit(X[:100, :])
---------------------------------------
X_transformed = transformer.fit_transform(X)

print(X_transformed.shape)

版本：

• 蟒蛇 …

我得到了什么terminate和kill做。有什么作用close()？

这是python文档中的引用：

关闭 Process 对象，释放与其关联的所有资源。如果底层进程仍在运行，则会引发 ValueError。一旦 close() 成功返回，Process 对象的大多数其他方法和属性将引发 ValueError。

在我看来，这是有史以来最无用的调用：

• 如果进程仍在运行，它什么都不做，但失败并出现异常

• 如果进程已经终止......它要么忙于清理，要么完成了清理，为什么我要close()在它上面调用一个显式？（此调用的使用是否取决于操作系统？）

是不是这个 doc 条目非常尴尬，不知何故它确实意味着close()有能力终止一个线程，但它只在某些 python 定义的检查点这样做？

假设我有以下代码：

from scipy import *
import multiprocessing as mp
num_cores = mp.cpu_count()
from joblib import Parallel, delayed
import matplotlib.pyplot as plt

def func(x,y):
    return y/x
def main(y, xmin,xmax, dx):
    x = arange(xmin,xmax,dx)
    output = Parallel(n_jobs=num_cores)(delayed(func)(i, y) for i in x)
    return x, asarray(output)
def demo():
    x,z = main(2.,1.,30.,.1)
    plt.plot(x,z, label='All values')
    plt.plot(x[z>.1],z[z>.1], label='desired range') ## This is better to do in main()
    plt.show()

demo()

我只想计算输出直到输出 > 给定数字（可以假设输出元素随着 x 的增加单调减少）然后停止（不计算 x 的所有值然后排序，这对我的目的来说效率低下）。有没有办法使用并行、延迟或任何其他多处理来做到这一点？

因此，假设您有一个 Python 进程，它正在以每秒大约 500 行的速度实时收集来自排队系统的数据（这可以进一步并行化以减少到大约 50 ps）并将其附加到DataFrame：

rq = MyRedisQueue(..)
df = pd.DataFrame()
while 1:
    recv = rq.get(block=True)
    # some converting
    df.append(recv, ignore_index = True)

现在的问题是：如何根据这些数据利用 CPU？所以我充分认识到的局限性GIL，看着成多处理管理器 的命名空间，这里太，但它看起来像有一些缺点关于延迟的centerally保持数据帧。在深入研究之前，我还尝试pool.map了我认为pickle在进程之间应用的方法，这种方法很慢并且开销太大。

所以在所有这一切之后，我终于想知道，如何（如果）每秒 500 行（甚至每秒 50 行）的插入可以传输到不同的进程，同时还有一些 CPU 时间用于对子项中的数据应用统计和启发式流程？

也许在两个进程之间实现自定义 tcp 套接字或排队系统会更好？或者在pandas其他库中是否有一些实现可以真正允许快速访问父进程中的一个大数据框？我爱熊猫！