假设我们有一个list(mylist)用作lapply函数的输入对象.有没有办法知道mylist正在评估哪个元素？该方法应该起作用,lapply并且snowfall::sfApply(也可能是其他人也适用于家庭成员).
在聊天时,Gavin Simpson提出了以下方法.这很有用,lapply但不是很重要sfApply.我想避免额外的包或摆弄列表.有什么建议？
mylist <- list(a = 1:10, b = 1:10)
foo <- function(x) {
deparse(substitute(x))
}
bar <- lapply(mylist, FUN = foo)
> bar
$a
[1] "X[[1L]]"
$b
[1] "X[[2L]]"
这是没有削减它的并行版本.
library(snowfall)
sfInit(parallel = TRUE, cpus = 2, type = "SOCK") # I use 2 cores
sfExport("foo", "mylist")
bar.para <- sfLapply(x = mylist, fun = foo)
> bar.para
$a
[1] "X[[1L]]"
$b
[1] …我有兴趣使用Microsoft的Direct Compute框架进行数值计算.整个概念似乎是一个相当糟糕的业务.是否有任何有用的资源,例如网络教程或电子书,我可以学习使用这个框架？
谢谢!
我的任务是将一个进程分成许多并行运行的小进程,分发给许多从机.请求通过HTTP传入,服务器将其分解为发送到从属计算机的多个子进程,等待所有从属请求返回响应,然后将聚合结果整理为单个数据对象,作为结果返回顶级请求.我认为node.js会很有用,但由于它是单线程的,我无法确定这是否可能,或者它是否会阻止等待每个请求返回之后再转到下一个请求.这可能与node.js一起使用吗？如果是这样,有人会指出我正确的方向吗？即一个节点模块使用或概述它将如何完成？
谢谢你的帮助.
技术堆栈:C#/ .NET 4/WinForms
背景:
我正在工作的项目是一系列图像堆栈的可视化应用程序.具体地,每个图像堆栈与网格对齐,在任何时间显示相同的图像,并且处理功能应用于当前在视图中的图像.图像堆栈本身为150-300 MB,每个映像为512KB-1MB.典型的数据集将包含~100个图像堆栈.
题:
为了尝试使用这些数据,我使用了几种技术:
然而,即使使用上述所有内容,重复性仍然有很多不足之处.使用SysInternals Process Explorer,CPU利用率很低(<25%),而内存使用率在垃圾收集发生之前会上升到极限.
分析表明,大部分执行时间都花费在从内存映射文件中获取数据.我假设它等待操作系统将请求的内存页面重新打开到活动内存中？
我还能做些什么来提高性能？
注意:
更新1:
.net c# parallel-processing memory-management image-processing
我是java的新手,非常喜欢学习它.我已经制作了一个运行良好的程序,但是当我添加更多数据进行处理时需要一些时间.我把它做了线程,它确实加速了很多,但现在我正在考虑加快速度(显然它需要处理的数据越多,所需的时间越长).只是一个fyi,我的程序不会在线程之间共享任何数据,它会获取列表中的一项并进行一些数学运算并将结果上传到数据库.理想情况下,一些工作计算机会获得列表中的一些项目然后完成其工作,然后在完成之前获得更多工作
我做了一些研究并找到了队列,我不确定它是否是我需要的东西,或者是否有其他的东西(我也在考虑保持工人的诚信/监督对我来说可能对于作为新手来说太多了).我家里有4台电脑(一些Linux,Mac和Windows ..但我可以在所有非Linux系统上安装linux vm,如果这些解决方案特定于操作系统)并且想要让他们也开始处理这项任务.我想创建Java队列,其他客户端采取一个部分和过程,但我也看到了库(rabbitmq).我也简要介绍过网格计算.
这是要走的路还是有更好的方法？我不需要任何代码或任何东西只是想知道分发线程的解决方案是什么,或者在评估时使用哪些因素.
我有一个使用64位Python 3.3.0 CPython解释器在64位Linux(内核版本2.6.28.4)机器上运行的自定义模拟器(用于生物学).
因为模拟器依赖于许多独立实验来获得有效结果,所以我建立了并行处理来运行实验.线程之间的通信主要发生在具有托管multiprocessing Queues(doc)的生产者 - 消费者模式下
.该体系结构的破坏如下:
Processes以及各种Queues 的主进程主进程和工作进程通过输入进行通信Queue.类似地,工作进程将结果放在Queue结果消费者进程从中消耗项目的输出中.最终的ResultConsumer对象通过multiprocessing Pipe
(doc)传递回主进程.
一切正常,直到它试图通过以下方法将ResultConsumer对象传递回主进程Pipe:
Traceback (most recent call last):
File "/home/cmccorma/.local/lib/python3.3/multiprocessing/process.py", line 258, in _bootstrap
self.run()
File "/home/cmccorma/.local/lib/python3.3/multiprocessing/process.py", line 95, in run
self._target(*self._args, **self._kwargs)
File "DomainArchitectureGenerator.py", line 93, in ResultsConsumerHandler
pipeConn.send(resCon)
File "/home/cmccorma/.local/lib/python3.3/multiprocessing/connection.py", line 207, in send
self._send_bytes(buf.getbuffer())
File "/home/cmccorma/.local/lib/python3.3/multiprocessing/connection.py", line 394, in _send_bytes
self._send(struct.pack("!i", n))
struct.error: 'i' format requires -2147483648 …我在读C++并发在行动由安东尼·威廉姆斯.在关于设计并发代码的章节中,有std :: for_each algorihtm的并行版本.以下是本书略有修改的代码:
join_thread.hpp
#pragma once
#include <vector>
#include <thread>
class join_threads
{
public:
explicit join_threads(std::vector<std::thread>& threads)
: threads_(threads) {}
~join_threads()
{
for (size_t i = 0; i < threads_.size(); ++i)
{
if(threads_[i].joinable())
{
threads_[i].join();
}
}
}
private:
std::vector<std::thread>& threads_;
};
parallel_for_each.hpp
#pragma once
#include <future>
#include <algorithm>
#include "join_threads.hpp"
template<typename Iterator, typename Func>
void parallel_for_each(Iterator first, Iterator last, Func func)
{
const auto length = std::distance(first, last);
if (0 == …我有一个MyReader实现Iterator并生成Buffers的地方Buffer : Send.很快就会MyReader生成很多东西Buffer,但我有一个CPU密集型的工作要对每个Buffer(.map(|buf| ...))这是我的瓶颈,然后收集结果(有序).我希望将CPU密集型工作并行化 - 希望能够使用工作窃取来执行它们,以及核心数量允许的速度.
编辑:更准确.我正在努力rdedup.MyStruct是Chunker哪些读取io::Read(通常是stdio),查找数据的部分(块)并产生它们.然后map()假设,对于每个块,计算它的sha256摘要,压缩,加密,保存并返回摘要作为结果map(...).已保存数据的摘要用于构建index数据.正在处理的块之间的顺序map(...)无关紧要,但是从每个块 返回的摘要map(...)需要以与找到块相同的顺序收集.实际save到文件步骤被卸载到另一个线程(写入线程).有问题的PR的实际代码
我希望我可以使用rayon它,但rayon期望一个已经可以并行的迭代器 - 例如.一个Vec<...>或类似的东西.我发现没有办法得到一个par_iter从MyReader-我的读者是非常单线程的性质.
有simple_parallel,但文件说,它不推荐用于一般用途.我想确保一切都能正常运作.
我可以采用spmc队列实现和自定义thread_pool,但我正在寻找经过优化和测试的现有解决方案.
还有pipeliner但不支持有序地图.
API通常具有用户必须遵循的速率限制。例如，让我们接受50个请求/秒。顺序请求需要0.5-1秒，因此太慢了，无法接近该限制。但是，使用aiohttp的并行请求超出了速率限制。
为了尽可能快地轮询API，需要对并行调用进行速率限制。
到目前为止，我发现的示例装饰了session.get，大致像这样：
session.get = rate_limited(max_calls_per_second)(session.get)
这对于顺序调用非常有效。尝试在并行调用中实现此功能无法按预期进行。
这是一些示例代码：
async with aiohttp.ClientSession() as session:
session.get = rate_limited(max_calls_per_second)(session.get)
tasks = (asyncio.ensure_future(download_coroutine(
timeout, session, url)) for url in urls)
process_responses_function(await asyncio.gather(*tasks))
问题在于它将限制任务的排队速度。与的执行gather将或多或少地同时发生。两全其美；-)。
是的，我在aiohttp处发现了一个类似的问题：设置每秒的最大请求数，但没有答复回答限制请求速率的实际问题。同样，来自Quentin Pradet的博客文章仅在限制队列速率上起作用。
总结一下：如何限制并行请求的每秒aiohttp请求数？
我使用tqdm将进度条添加到了2.7 python代码中，但是它大大降低了我的代码速度。如果没有进度条，则需要12秒，而使用进度条则需要57秒。
没有进度条的代码如下所示：
p = mp.Pool()
combs = various combinations
result = p.map(self.parallelize, combs)
p.close()
p.join()
带有进度条的代码如下：
from tqdm import tqdm
p = mp.Pool()
combs = various combinations
result = list(tqdm(p.imap(self.parallelize, combs), total = 5000))
p.close()
p.join()
有没有一种更好的方法不会降低我的代码速度呢？