我一直在思考这个问题:
您是否可以利用您拥有多个CPU的优势,在多核计算机上构建更快的基础数据结构(即链表,哈希表,集合,跳转列表,布隆过滤器,红黑树等)？
我做了一些pthreads的初步试验,发现pthread_create()的顺序为30us,但是一个简单的hash_map插入所花费的时间远远少于单个核心.因此,我很难想象创建一个更快的hash_map <>,因为同步原语和线程创建是如此之慢.我还可以想象树的遍历和并行平衡,但同样,同步原语似乎会使运行时更长,而不是更短.
对我来说,我仍然觉得"我有更多的CPU,因此,我应该能够更快地做到这一点",但我无法完全围绕证据或反证据证明这一点.我在C++中进行了相当多的实验,但我现在怀疑其他语言可能会为这项任务提供更好的解决方案(erlang？).思考？
编辑细节:我认为有几种经常使用的编程/数据结构范例可能会加速.例如,我发现自己经常编写基本上看起来像这样的代码(其中实际数据已被"rand()"替换)
static const int N = 1000000;
static const int M = 10000000; // 10x more lookups
hash_map<int, int> m;
// batch insert a bunch of interesting data
for (int i = 0; i < N; i++) m[rand()] = rand();
// Do some random access lookups.
for (int i = 0; i < M; i++) m[rand()]++;
这种范例经常用于名称 - 值设置和配置数据,批处理等等.10x(或更多)查找/插入比率使传统的hash_map <>成为这种操作的理想选择.
这可以很容易地分成两半,具有插入阶段和查找阶段,并且在并行世界中,在两半之间可能存在一些"刷新队列"操作.交错插入+查找版本更难:
hash_map<int, int> m;
for (int i = 0; i < N; i++) {
if …我有一个关于以下代码示例的问题(m_value不是volatile,每个线程都在一个单独的处理器上运行)
void Foo() // executed by thread #1, BEFORE Bar() is executed
{
Interlocked.Exchange(ref m_value, 1);
}
bool Bar() // executed by thread #2, AFTER Foo() is executed
{
return m_value == 1;
}
在Foo()中使用Interlocked.Exchange是否保证在执行Bar()时,我会看到值"1"？(即使值已存在于寄存器或缓存行中？)或者在读取m_value的值之前是否需要设置内存屏障？
另外(与原始问题无关),声明一个volatile成员并通过引用InterlockedXX方法传递它是否合法？(编译器警告通过引用传递volatile,所以在这种情况下我应该忽略警告吗？)
请注意,我不是在寻找"更好的做事方式",所以请不要发布建议完全替代方式("使用锁定"等)的答案,这个问题来自于纯粹的兴趣..
我有一个monadic函数getRate:
getRate :: String -> IO Double
我想将这个函数映射到String的列表上.通常情况下,我会这样做:
mapM getRate ["foo", "bar"]
但是由于每次调用getRate进行网络调用,我都希望并行化地图,以便在一个单独的线程中获取每个速率(或者至少在队列中分散).我在想类似的东西
parMapM getRate ["foo", "bar"]
但没有parMapM函数,parMap不适用于monadic函数.
我能做什么？
parallel-processing monads concurrency multithreading haskell
经过大量搜索c中并行快速排序的实现后,我即将潜入并自己编写代码.(我需要对一个大约100万个文本字符串的数组进行排序.)似乎我发现的所有实现都将qsort函数本身的工作分开,这在分割每个线程相对少量的工作时会产生大量的开销.
将100万个字符串除以线程数(在我的情况下是24个线程)并将它们分别放在一个节上,然后进行合并输出会不会快得多？当然,这具有理论上的缺点,即它不是就地排序,但是随着可用内存的大量存在,这不是问题.运行的机器有12个(非常快)物理/ 24逻辑核心和192 GB(是,千兆字节)的内存.目前,即使在这台机器上,排序也需要大约8分钟!
编译器每次都抱怨并行Haskell的不同示例应用程序; 有了这条消息:
Could not find module `Control.Parallel.Strategies'
ghc编译器命令:
ghc -threaded -i/sudo/dir/par-modules/3 -cpp -DEVAL_STRATEGIES -eventlog --make parFib.hs
同样简单
ghc -O2 --make -threaded parFib.hs
我忽略了什么细节？我错过了一些PATH变量.
进口可能如下所示:
module Main where
import System
# if defined(EVAL_STRATEGIES)
import Control.Parallel
import Control.Parallel.Strategies
#endif
干杯
parallel-processing haskell functional-programming compiler-errors ghc
我试图找到大约61亿(自定义)项目的最大重量,我想用并行处理这样做.对于我的特定应用程序,有更好的算法,不需要我迭代超过61亿项,但解释它们的教科书是我的头脑,我的老板希望在4天内完成.我想我的公司的花哨的服务器和并行处理有更好的机会.但是,我所知道的关于并行处理的一切都来自于阅读Python 文档.这就是说我很丢失......
我目前的理论是设置一个馈送器进程,一个输入队列,一大堆(比如说30个)工作进程,以及一个输出队列(在输出队列中找到最大元素将是微不足道的).我不明白的是,馈线进程如何告诉工作进程何时停止等待项目通过输入队列.
我曾经考虑过使用multiprocessing.Pool.map_async我的6.1E9项目的迭代,但是只需要花费将近10分钟来迭代这些项目而不对它们做任何事情.除非我误解了某些东西......在map_async迭代过程中将它们分配给流程可以在流程开始工作时完成.(Pool也提供imap但是文档说它类似于map,它似乎不是异步工作.我想要异步,对吗？)
相关问题:我想用concurrent.futures而不是multiprocessing吗？我不可能是第一个实施双排队系统的人(这正是美国每家熟食店的生产线如何工作......)那么有更多的Pythonic /内置方法吗？
这是我正在尝试做的一个框架.请参阅中间的注释块.
import multiprocessing as mp
import queue
def faucet(items, bathtub):
"""Fill bathtub, a process-safe queue, with 6.1e9 items"""
for item in items:
bathtub.put(item)
bathtub.close()
def drain_filter(bathtub, drain):
"""Put maximal item from bathtub into drain.
Bathtub and drain are process-safe queues.
"""
max_weight = 0
max_item = None
while True:
try: …我编写了以下简短的C++程序来重现Herb Sutter所描述的错误共享效果:
比如说,我们想要执行总量的WORKLOAD整数运算,并且我们希望它们平均分配给多个(PARALLEL)线程.出于此测试的目的,每个线程将从整数数组中递增其自己的专用变量,因此该过程可以理想地并行化.
void thread_func(int* ptr)
{
for (unsigned i = 0; i < WORKLOAD / PARALLEL; ++i)
{
(*ptr)++;
}
}
int main()
{
int arr[PARALLEL * PADDING];
thread threads[PARALLEL];
for (unsigned i = 0; i < PARALLEL; ++i)
{
threads[i] = thread(thread_func, &(arr[i * PADDING]));
}
for (auto& th : threads)
{
th.join();
}
return 0;
}
我认为这个想法很容易理解.如果你设置
#define PADDING 16
每个线程将在单独的缓存行上工作(假设缓存行的长度为64字节).因此,结果将是加速的线性增加,直到PARALLEL> #core.另一方面,如果将PADDING设置为低于16的任何值,则应该遇到严重的争用,因为现在至少有两个线程可能在相同的高速缓存行上运行,但是受到内置硬件互斥锁的保护.我们希望我们的加速不仅在这种情况下是次线性的,而且即使总是<1,因为看不见的锁定车队.
现在,我的第一次尝试几乎满足了这些期望,但PADDING避免错误共享所需的最小值是8左右而不是16分.在我得出明显结论之前,我很困惑半小时,我无法保证我的数组与主内存中的缓存行的开头完全对齐.实际对齐可能根据许多条件而变化,包括阵列的大小.
在这个例子中,当然没有必要让我们以特殊的方式对齐数组,因为我们可以将PADDING保持在16并且一切正常.但人们可以想象一下案例,它确实会产生影响,某个结构是否与缓存行对齐.因此,我添加了一些代码行来获取有关数组实际对齐的一些信息.
int main()
{
int arr[PARALLEL * 16];
thread threads[PARALLEL];
int offset …我在基于Intel i3的计算机上运行以下代码,该计算机具有4个虚拟核心(2个超线程/物理核心,64位)和安装的Ubuntu 14.04:
n = multiprocessing.cpu_count()
executor = ThreadPoolExecutor(n)
tuple_mapper = lambda i: (i, func(i))
results = dict(executor.map(tuple_mapper, range(10)))
代码似乎没有以并行方式执行,因为CPU的使用率仅为25%.在利用率图表中,一次仅100%使用4个虚拟核心中的一个.使用的核心每10秒左右交替一次.
但是并行化在具有相同软件设置的服务器计算机上运行良好.我不知道核心的确切数量,也不知道确切的处理器类型,但我确信它有几个核心,利用率为100%,并且计算速度快(使用并行化后速度提高了10倍)一些实验用它).
我希望,并行化也可以在我的机器上运行,而不仅仅是在服务器上.
为什么不起作用？它与我的操作系统设置有关吗？我需要改变它们吗？
提前致谢!
更新: 有关背景信息,请参阅下面的正确答案.为了完整起见,我想提供一个解决问题的示例代码:
tuple_mapper = lambda i: (i, func(i))
n = multiprocessing.cpu_count()
with concurrent.futures.ProcessPoolExecutor(n) as executor:
results = dict(executor.map(tuple_mapper, range(10)))
在重用之前,请注意您正在使用的所有函数都在模块的顶层定义,如下所述: Python多处理酸洗错误
为什么下面的代码不打印任何输出,而如果我们删除并行,它打印0,1？
IntStream.iterate(0, i -> ( i + 1 ) % 2)
.parallel()
.distinct()
.limit(10)
.forEach(System.out::println);
虽然我知道理想的限制应该放在不同之前,但我的问题更多地与添加并行处理引起的差异有关.
刚刚在新计算机上下载了p4v,我正在尝试拉出我的存储库,我遇到了这个错误.在今天之前从未拉过存储库的问题.知道解决方法的用途是什么吗？
我试着运行命令
p4 configure set net.parallel.max = 5
但这给了我一个你尝试它时没有这个操作的权限