我想multiprocessing在我的代码中使用以获得更好的性能.

但是,我收到如下错误:

Traceback (most recent call last):
  File "D:\EpubBuilder\TinyEpub.py", line 49, in <module>
    e.epub2txt()
  File "D:\EpubBuilder\TinyEpub.py", line 43, in epub2txt
    tempread = self.get_text()
  File "D:\EpubBuilder\TinyEpub.py", line 29, in get_text
    txtlist = pool.map(self.char2text,charlist)
  File "C:\Python34\lib\multiprocessing\pool.py", line 260, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "C:\Python34\lib\multiprocessing\pool.py", line 599, in get
    raise self._value
  File "C:\Python34\lib\multiprocessing\pool.py", line 383, in _handle_tasks
    put(task)
  File "C:\Python34\lib\multiprocessing\connection.py", line 206, in send
    self._send_bytes(ForkingPickler.dumps(obj))
  File "C:\Python34\lib\multiprocessing\reduction.py", line 50, in dumps
    cls(buf, protocol).dump(obj)
TypeError: cannot serialize '_io.BufferedReader' object …

我想基于a中的值创建一个类的实例pandas.DataFrame.我已经失败了.

import itertools
import multiprocessing as mp
import pandas as pd

class Toy:
    id_iter = itertools.count(1)

    def __init__(self, row):
        self.id = self.id_iter.next()
        self.type = row['type']

if __name__ == "__main__":

    table = pd.DataFrame({
        'type': ['a', 'b', 'c'],
        'number': [5000, 4000, 30000]
        })

    for index, row in table.iterrows():
        [Toy(row) for _ in range(row['number'])]

多处理尝试

我已经能够通过添加以下内容来并行化这种(某种程度):

pool = mp.Pool(processes=mp.cpu_count())
m = mp.Manager()
q = m.Queue()

for index, row in table.iterrows():
    pool.apply_async([Toy(row) for _ in range(row['number'])])

如果数字row['number']显着长于长度,这似乎会更快table.但在我的实际情况中,table …

我正在使用tensorflow来预处理一些大图像.我遇到了一个内存迅速崩溃的问题.我转向在python中使用多处理,因此内存将在我想要的时候完全释放.

问题是,我正在使用python的多进程队列,由于某种原因未知,我无法将我的tensorflow会话从我的父进程传递给子进程.使用一些高级调试技术(即每几行打印一些东西)我注意到python在我使用会话的行内空闲,它不会抛出错误消息.

我的代码看起来像这样:

def subprocess(some_image, sess, q):
    with sess.as_default():
        # ... use sess and q ...
        print "All good and well" #This is printed
        some_image.eval() #Nothing happens here in console
        print "Still all good and well" #This is not printed

if __name__ == '__main__':
    # ... some initial operations ...
    some_image = read_some_image()

    sess = tf.Session()

    q = Queue()
    q.put(something)
    p = Process(target=subprocess, args=(some_image, sess, q))
    p.start()
    p.join()

可能是什么问题呢？非常感谢!

我有一个存储对象的字典:

jobs = {}
job = Job()
jobs[job.name] = job

现在我想将它转换为使用管理器dict,因为我想使用多处理并需要共享这个dict amonst进程

mgr = multiprocessing.Manager()
jobs = mgr.dict()
job = Job()
jobs[job.name] = job

只是通过转换为使用manager.dict(),事情变得非常缓慢.

例如,如果使用原生字典,则只需0.65秒即可创建625个对象并将其存储到dict中.

同样的任务现在需要126秒!

我可以做的任何优化使manager.dict()与python {}保持一致？

我对concurrent.futures有两个问题:

如何在python concurrent.futures中打破time.sleep()？

结论:time.sleep()不能被中断.一种解决方案是:您可以围绕它编写循环并进行短暂睡眠.

请参阅如何在python concurrent.futures中中断time.sleep()

concurrent.futures的个别超时？

结论:个人超时需要由用户实施.例如:对于每个超时,您可以调用wait().

请参阅 concurrent.futures的各个超时

题

是否ASYNCIO解决问题的论文？

我试图将共享秘密传递给子进程.在Linux环境中,这可行.在Windows环境中,孩子不会收到共享密钥.下面的三个文件是我正在尝试做的一个简单示例:

main.py

import multiprocessing
import module1
import module2

if __name__ == "__main__":
    module1.init()
    process = multiprocessing.Process(target=module2.start)
    process.start()
    process.join()

module1.py

import ctypes
import multiprocessing

x = None

def init():
    global x
    x = multiprocessing.Value(ctypes.c_wchar_p, "asdf")

module2.py

import module1

def start():
    print(module1.x.value)

在Ubuntu 14.04环境中,在Python 3.5上,我收到以下输出:

$ python3 main.py
asdf

在CentOS 7环境中,我收到以下输出:

$ python3 main.py
asdf

在Windows 10上使用Windows子系统Linux(在创建者更新之前和之后,所以Ubuntu 14.04和16.04)我得到以下输出:

$ python3 main.py
asdf

但是,在Windows 7和Windows 10环境中,使用3.5或3.6,我得到的AttributeError不是上面的输出:

Process Process-1:
Traceback (most recent call last):
  File "C:\Python\Python35\lib\multiprocessing\process.py", line 249, in _bootstrap
    self.run()
  File …

我有一个我想要使用多处理修改的元素列表.问题是对于某些特定的输入(在尝试之前不可观察),我的部分功能会停止.我已经在概念上用下面的代码展示了这个函数,其中函数sometimes_stalling_processing()偶尔会无限期地停止.

为了将其置于上下文中,我正在使用Web scraper处理一堆链接,并且即使在请求模块中使用超时,其中一些链接也会停止.我尝试了不同的方法(例如使用eventlet),但得出的结论是,在多处理级别处理它可能更容易.

def stable_processing(obs):
    ...
    return processed_obs

def sometimes_stalling_processing(obs):
    ...
    return processed_obs

def extract_info(obs):
    new_obs = stable_processing(obs)
    try:
        new_obs = sometimes_stalling_processing(obs)
    except MyTimedOutError: # error doesn't exist, just here for conceptual purposes
        pass
    return new_obs

pool = Pool(processes=n_threads)
processed_dataset = pool.map(extract_info, dataset)
pool.close()
pool.join()

这个问题(如何在超时后中断多处理.Pool中的任务？)似乎非常相似,但我一直无法将其转换为使用map而不是apply.我也尝试过使用该eventlet软件包,但这不起作用.请注意,我使用的是Python 2.7.

如何pool.map()对个别观察结果进行超时并杀死sometimes_stalling_processing？

我在具有8 GB RAM的macOS上具有4个内核(8线程超线程)的Intel i7并行生成大约400,000,000(4亿)个随机数.

但是,我也在DigitalOcean服务器上生成400,000,000个随机数,Debian上有20个内核,64 GB RAM.

这是代码:

import multiprocessing
import random

rangemin = 1
rangemax = 9

def randomGenPar_backend(backinput):
    return random.randint(rangemin, rangemax)

def randomGenPar(num):
    pool = multiprocessing.Pool()
    return pool.map(randomGenPar_backend, range(0, num))

randNum = 400000000

random.seed(999)
randomGenPar(randNum)

这些是基准测试的结果:

5,000,000 Random Numbers:
1 Core: 5.984
8 Core: 1.982

50,000,000 Random Numbers:
1 Core: 57.28
8 Core: 19.799
20 Core: 18.257
Times Benefit (20 core vs. 8 core) = 1.08

100,000,000 Random Numbers:
1 Core: 115
8 Core: 40.434
20 Core: 31.652 …

我按照下面的代码在postgres数据库上实现并行选择查询:

https://tech.geoblink.com/2017/07/06/parallelizing-queries-in-postgresql-with-python/

我的基本问题是我有大约6k个需要执行的查询,我正在尝试优化这些选择查询的执行.最初它是一个where id in (...)包含所有6k谓词ID 的单个查询,但是我在运行它的机器上使用> 4GB的RAM遇到了查询问题,因此我决定将其拆分为6k个别查询,当同步保持时稳定的内存使用.然而,运行时间需要更长的时间,这对我的用例来说不是一个问题.即便如此,我也在尽量减少时间.

这就是我的代码:

class PostgresConnector(object):
    def __init__(self, db_url):
        self.db_url = db_url
        self.engine = self.init_connection()
        self.pool = self.init_pool()

    def init_pool(self):
        CPUS = multiprocessing.cpu_count()
        return multiprocessing.Pool(CPUS)

    def init_connection(self):
        LOGGER.info('Creating Postgres engine')
        return create_engine(self.db_url)

    def run_parallel_queries(self, queries):
        results = []
        try:
            for i in self.pool.imap_unordered(self.execute_parallel_query, queries):
                results.append(i)
        except Exception as exception:
            LOGGER.error('Error whilst executing %s queries in parallel: %s', len(queries), exception)
            raise
        finally:
            self.pool.close()
            self.pool.join()

        LOGGER.info('Parallel query ran producing %s sets of results of type: …

我经常发现自己用Python编写程序来构造一个大的(兆字节)只读数据结构,然后使用该数据结构来分析一个非常大的(总共几百兆字节)小记录列表.每个记录都可以并行分析,因此自然模式是设置只读数据结构并将其分配给全局变量,然后创建一个multiprocessing.Pool(通过隐式将数据结构复制到每个工作进程中fork)和然后用来imap_unordered并行处理记录.这种模式的骨架看起来像这样:

classifier = None
def classify_row(row):
    return classifier.classify(row)

def classify(classifier_spec, data_file):
    global classifier
    try:
        classifier = Classifier(classifier_spec)
        with open(data_file, "rt") as fp, \
             multiprocessing.Pool() as pool:
            rd = csv.DictReader(fp)
            yield from pool.imap_unordered(classify_row, rd)
    finally:
        classifier = None

我因为全局变量之间的隐式耦合的不满意这点classify和classify_row.理想情况下,我想写

def classify(classifier_spec, data_file):
    classifier = Classifier(classifier_spec)
    with open(data_file, "rt") as fp, \
         multiprocessing.Pool() as pool:
        rd = csv.DictReader(fp)
        yield from pool.imap_unordered(classifier.classify, rd)

但是这不起作用,因为Classifier对象通常包含无法pickle的对象(因为它们是由作者不关心的扩展模块定义的); 我还读过如果它确实有效会很慢,因为在每次调用绑定方法时,Classifier对象都会被复制到工作进程中.

还有更好的选择吗？我只关心3.x.