标签: pool

我正在研究一款需要高性能的视频游戏,所以我试图设置一个好的记忆策略或游戏的特定部分,即游戏"模型",即游戏表现.我有一个包含整个游戏表示的对象,里面有不同的管理器,以保持表示符合游戏规则.每个游戏实体当前都是由特定类型的工厂生成的,所以我有几个工厂允许我按照自己的意愿隔离和更改这些实体的内存管理.

现在,我正在选择这两种选择之间:

1. 为每种类型设置一个内存池:这将允许真正快速的分配/释放和最小的碎片,因为对象池已经知道分配的对象的大小.困扰我的一件事是让几个像这样的池分开,可能会让其他解决方案更有效......
2. 有一个大内存池由一个游戏表示的所有工厂共享 :(使用类似boost :: pool和一些适配器功能)这样我就可以将所有游戏对象内存分配在一起并且可以为游戏分配一个比特我已经知道了总的大小(并非总是如此).我不确定这是一个比A更好的解决方案,因为池内可能存在碎片,因为在同一个池中会有不同大小的对象,但对于内存分析和其他问题修复来说它看起来更容易.

现在,我对A有了一些现实世界的经验,所以我对B没有经验,并希望对这些解决方案有一些建议,对于一个长寿项目.对于长寿项目来说,哪种解决方案似乎更好？为什么？(注意:在这种情况下,确实需要一个池,因为游戏模型也用于游戏编辑,所以会有很多小对象的分配/释放).

编辑澄清:如果(我还不清楚)我正在使用C++

这是关于Java字符串常量池.在我的一个程序中,我正在解密数据库的密码并将其存储在字符串中.我听说Java Strings将存储在一个常量池中,它们不会被VM重新启动或加载String Quits的ClassLoader破坏.

如果是这种情况,我的密码将存储在字符串池中.我非常关心这个问题.有没有其他方法来摧毁这些文字或我能做的任何其他事情.

请在此建议,

此致,Sunny.

我在C中寻找一个好的内存池实现.

它应包括以下内容:

1. 反碎片化.
2. 超级快:)
3. 能够在某些标识符下"捆绑"来自不同大小的多个分配,并删除具有给定标识符的所有分配.
4. 线程安全

当使用Pool.apply_async运行大量任务(具有大参数)时,将分配进程并进入等待状态,并且对等待进程的数量没有限制.这可能最终会占用所有内存,如下例所示:

import multiprocessing
import numpy as np

def f(a,b):
    return np.linalg.solve(a,b)

def test():

    p = multiprocessing.Pool()
    for _ in range(1000):
        p.apply_async(f, (np.random.rand(1000,1000),np.random.rand(1000)))
    p.close()
    p.join()

if __name__ == '__main__':
    test()

我正在寻找一种限制等待队列的方法,以便只有有限数量的等待进程,并且在等待队列已满时阻塞Pool.apply_async.

从最纯粹的角度来看,他们感觉就像是相同的概念.它们都管理一组reosurces /节点,并控制它们对外部组件的访问.

使用池,您可以在池中借用和返回这些资源/节点.

对于群集,您有一个位于资源/节点前面的负载均衡器,您可以通过请求命中负载均衡器.

在这两种情况下,您都无法控制请求/借用映射到哪个资源/节点.

所以我提出了一个问题:"池"模式和负载均衡集群之间的根本区别是什么？

我想在课堂上使用Pool,但似乎有问题.我的代码很长,我创建了一个小型演示变体来说明问题.如果你能给我一个下面代码的变体,那就太棒了.

from multiprocessing import Pool

class SeriesInstance(object):
    def __init__(self):
        self.numbers = [1,2,3]
    def F(self, x):
        return x * x
    def run(self):
        p = Pool()
        print p.map(self.F, self.numbers)


ins = SeriesInstance()
ins.run()

输出:

Exception in thread Thread-2:
Traceback (most recent call last):
  File "/usr/lib64/python2.7/threading.py", line 551, in __bootstrap_inner
    self.run()
  File "/usr/lib64/python2.7/threading.py", line 504, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/usr/lib64/python2.7/multiprocessing/pool.py", line 319, in _handle_tasks
    put(task)
PicklingError: Can't pickle <type 'instancemethod'>: attribute lookup __builtin__.instancemethod failed

然后挂起.

我不确定何时使用工作池与多个进程.

processes = []

for m in range(1,5):
       p = Process(target=some_function)
       p.start()
       processes.append(p)

for p in processes:
       p.join()

VS

if __name__ == '__main__':
    # start 4 worker processes
    with Pool(processes=4) as pool:
        pool_outputs = pool.map(another_function, inputs)

我正在尝试使用concurrent.futures.ProcessPoolExecutorLocks,但是我遇到了运行时错误.(如果相关,我正在使用Windows)

这是我的代码:

import multiprocessing
from concurrent.futures import ProcessPoolExecutor

import time


def f(i, lock):
    with lock:
        print(i, 'hello')
        time.sleep(1)
        print(i, 'world')


def main():
    lock = multiprocessing.Lock()
    pool = ProcessPoolExecutor()
    futures = [pool.submit(f, num, lock) for num in range(3)]
    for future in futures:
        future.result()


if __name__ == '__main__':
    main()

这是我得到的错误:

    Traceback (most recent call last):
  File "F:\WinPython-64bit-3.4.3.2\python-3.4.3.amd64\Lib\multiprocessing\queues.py", line 242, in _feed
    obj = ForkingPickler.dumps(obj)
  File "F:\WinPython-64bit-3.4.3.2\python-3.4.3.amd64\Lib\multiprocessing\reduction.py", line 50, in dumps
    cls(buf, protocol).dump(obj)
  File "F:\WinPython-64bit-3.4.3.2\python-3.4.3.amd64\Lib\multiprocessing\synchronize.py", line 102, in __getstate__
    context.assert_spawning(self)
  File …

我花了几个小时来尝试并行化我的数字运算代码,但是当我这样做时它只会变慢.不幸的是,当我尝试将其减少到下面的示例时,问题就消失了,我真的不想在这里发布整个程序.所以问题是:在这类程序中我应该避免哪些陷阱？

(注意:Unutbu的答案在底部后跟进.)

以下是情况:

• 它是关于一个模块,它定义了一个BigData包含大量内部数据的类.在该示例中,存在一个ff插值函数列表; 在实际的程序,还有更多,例如ffA[k],ffB[k],ffC[k].
• 计算将被归类为"令人尴尬的并行":可以一次在较小的数据块上完成工作.在这个例子中,那是do_chunk().
• 在我的实际程序中,示例中显示的方法将导致最差的性能:每个块大约1秒(在单个线程中完成的实际计算时间的0.1秒左右).因此,对于n = 50,do_single()将在5秒内do_multi()运行并且将在55秒内运行.
• 我还尝试通过将xi和yi数组切割成连续的块并迭代k每个块中的所有值来分解工作.这工作得更好一点.现在,无论是使用1,2,3或4个线程,总执行时间都没有差别.但当然,我希望看到实际的加速!
• 这可能是相关的:Multiprocessing.Pool使Numpy矩阵乘法更慢.但是,在程序的其他地方,我使用了一个多处理池进行更加孤立的计算:一个看起来类似的函数(没有绑定到类),def do_chunk(array1, array2, array3)并对该数组进行仅限于numpy的计算.在那里,有显着的速度提升.
• CPU使用率随着预期的并行进程数量而变化(三个线程的CPU使用率为300%).

#!/usr/bin/python2.7

import numpy as np, time, sys
from multiprocessing import Pool
from scipy.interpolate import RectBivariateSpline

_tm=0
def stopwatch(msg=''):
    tm = time.time()
    global _tm
    if _tm==0: _tm = tm; return
    print("%s: %.2f seconds" % (msg, tm-_tm))
    _tm = tm

class …

我正在运行40GB数据的计算.每个文件都是一个包含json行的压缩gzip文件.每个文件最多有500,000行,或大约500MB.我有一个运行128 cpu和1952 GB内存的亚马逊实例.我要做的是尽快处理每个文件.

我正在使用这样的多处理池:

def initializeLock(l):

    global lock
    lock = l

if __name__ == '__main__':
    directory = '/home/ubuntu/[directory_containing_files]/*.gz'
    file_names = glob.glob(directory)

    lock = Lock()
    pool = Pool(initializer=initializeLock, initargs=(lock,))
    pool.map(do_analysis, file_names)
    pool.close()
    pool.join()

我期望发生的是创建大量进程,每个进程处理一个文件.实际发生的是最初创建了100多个进程.在这一点上,我使用了大约85%的记忆力,这太棒了!然后每个完成.最终运行的进程数量下降到大约10.此时我只使用了5%的内存.定期启动其他进程,但它永远不会恢复运行100左右.所以我拥有这个拥有所有这些空闲内存的大型CPU,但我大部分时间都在运行大多数10个进程.

任何想法,如何让它继续运行100个进程,直到所有文件都完成？

编辑:

我在应用程序中添加了一些日志记录 最初它加载了127个进程,我认为这是因为我有128个CPU,其中一个在加载进程时正在使用.某些过程成功完成,结果已保存.然后在某些时候,只有少数正在运行的进程结束.当我检查完成了多少文件时,127个中的22个完成了.然后它只使用5-10个进程运行,所有这些都成功完成.我想也许它会耗尽内存和崩溃.但为什么？我有很多内存和很多CPU.

编辑2:

所以我发现了这个问题.问题是我在do_analysis方法中设置了一个锁,并且所有进程大约在同一时间完成并等待锁被释放.这些过程没有停止,他们正在睡觉.所以这让我想到另一个问题:我的主要目标是获取具有许多json行的每个文件,从json行获取ID属性,然后将其附加到包含具有相同id的其他行的文件.如果文件不存在,我创建它.我所做的是在访问文件时设置锁定,以避免被另一个进程访问.这是我的代码.

for key, value in dataframe.iteritems():
    if os.path.isfile(file_name):
        lock.acquire()
        value.to_csv(filename), mode='a', header=False, encoding='utf-8')
        lock.release()
    else:
        value.to_csv(filename), header=True, encoding='utf-8')

所以现在我正在尝试一种创造性的方式来附加到文件,但不会阻止其他所有进程.我正在处理大量数据,并且需要同时访问两个文件的可能性很低,但它仍然会发生.所以我需要确保在附加文件时,另一个进程不会尝试打开该文件.