标签: parallel-processing

所以我意识到这个问题听起来很愚蠢(是的,我使用的是双核),但是我尝试了两个不同的库(Grand Central Dispatch和OpenMP),并且在使用clock()来计算带有和不带有线条的代码时它平行,速度是一样的.(为了记录,他们都使用自己的平行形式).他们报告说在不同的线程上运行,但也许它们运行在同一个核心上？有没有办法检查？(这两个库都用于C,我在较低层时感到不舒服.)这非常奇怪.有任何想法吗？

我想开始使用任务并行库,因为这是执行异步操作的推荐框架.我无法找到的一件事是强制中止的任何方法,例如Thread.Abort提供的内容.

我特别关注的是我安排运行代码的任务,我不希望完全信任.特别是,我不能确定这个不受信任的代码不会死锁,因此我无法确定使用此代码的任务我是否会完成.我想远离真正的AppDomain隔离(由于编组的开销和复杂性),但我也不想让任务线程挂起,死锁.有没有办法在TPL中做到这一点？

我正在寻找Python 的fork-join 模型的实现。作为 Java 的 ForkJoinPool，它应该允许将一个任务的工作递归地拆分（fork）为多个子任务。子任务完成后，结果将被连接并返回。理想情况下，它应该支持类似于concurrent.futures中的ThreadPoolExecutor和ProcessPoolExecutor的线程和进程，但目前线程更重要。它必须允许限制线程数量（我希望每个核心有一个线程）。我知道这只有在代码释放 GIL 时才有用。

维基百科的示例用于阐明 fork-join 模型：

solve(problem):
    if problem is small enough:
        solve problem directly (sequential algorithm)
    else:
        for part in subdivide(problem)
            fork subtask to solve(part)
        join all subtasks spawned in previous loop
        return combined results

Python中有这样的库吗？我找不到。

我有一个长时间运行的进程，它在数百万条记录之间执行匹配，我使用服务总线调用此代码，但是，当我的进程超过 5 分钟限制时，Azure 会再次从头开始处理已处理的记录。

我怎样才能避免这种情况

这是我的代码：

private static async Task ProcessMessagesAsync(Message message, CancellationToken token)
 {
   long receivedMessageTrasactionId = 0;
   try
   {
     IQueueClient queueClient = new QueueClient(serviceBusConnectionString, serviceBusQueueName, ReceiveMode.PeekLock);

     // Process the message
     receivedMessageTrasactionId = Convert.ToInt64(Encoding.UTF8.GetString(message.Body));

     // My Very Long Running Method
     await DataCleanse.PerformDataCleanse(receivedMessageTrasactionId);
            //Get Transaction and Metric details

     await queueClient.CompleteAsync(message.SystemProperties.LockToken);
   }
   catch (Exception ex)
   {
     Log4NetErrorLogger(ex);
     throw ex;
   }
}

我定义了一个运行 bash 脚本的 python 函数。假设该函数是：calc(x,y,z)。如果我在 python 中使用一些变量运行这个函数，

>>> calc(1,2,3)

它生成一个使用变量模拟某些内容的 C 代码(x=1, y=2, z=3)，编译 C 代码并执行编译后的输出文件。

我想在 jupyter 笔记本中同时运行多个calc(x,y,z)具有不同 s 的 s 。(x,y,z)您可能已经注意到，问题在于 jupyter Notebook 中的单元格是按顺序执行的。如果我运行三个calc函数，则需要比运行一个函数的时间长三倍的时间。

我尝试了两种方法，但效果不佳。

1. 使用multiprocessing模块：通过使用模块，可以calc在“一个单元”中同时执行多个操作。但为了以后的分析，我想同时执行多个单元，其中每个单元仅calc使用不同的处理器（或 CPU 内核）。

2. 使用ipyparallel细胞魔法（受此答案启发）：导入后我尝试如下ipyparallel

   # Cell 1
   %%px --targets 0 # use processor 0
   calc(1,1,1)
   

   Run Code Online (Sandbox Code Playgroud)

   。

   # Cell 2
   %%px --targets 1 # use processor 1
   calc(2,2,2)        
   

   Run Code Online (Sandbox Code Playgroud)

   。

   # Cell 3
   %%px …

   Run Code Online (Sandbox Code Playgroud)

如果微服务是可扩展的，例如部署为 AWS 上的 ECS，那么在微服务开发中是否应该使用并行编程？

如果是，那么与 N 个实例消耗相同资源相比，一个实例消耗更多资源有什么好处？

并行编程如何匹配https://12factor.net/

PS 更具体地说 - 我应该在概念上使用并行流而不是简单流吗？

如果在任何一个进程中parallel::mclapply()遇到错误（例如， a ） ，是否可以请求尽快放弃所有进一步的处理？stop()

在 Jupyter 笔记本中使用 Parallel 时是否可以打印内容或进行调试。

这是我的代码

import pandas as pd
from sklearn.model_selection import ParameterGrid
from joblib import Parallel, delayed

def my_func(a,b):
    print("hi")
    return {"a":a,"b":b},a + b

grid = ParameterGrid({"a": [1, 2],
                      "b": [3, 4]})
resList = Parallel(n_jobs=-1)(delayed(my_func)(**params) for params in grid)
cols = ['params', 'results']
resDf = pd.DataFrame(resList,columns=cols)

数据框包含正确的结果，但函数内的“hi”行不打印

这可能是一个相当基本的问题。我正在学习网络应用程序开发的基础知识以及在这方面的并发性和并行性。

如果我启动 Flask Web 应用程序并将 Gunicorn 设置为 1 个工作线程：

gunicorn -w 1 server:app

这是否总是意味着只有一个进程在运行我的应用程序？或者这不是确定的，有些工作人员可能不仅使用线程、greenlet 等，而且还会为我的应用程序生成一个完整的附加进程？

对于后者，这是否也适用于sync或gevent工人，或者这些总是一个过程？

我必须从 Web API (NCBI entrez) 检索大型数据集，该数据集将我每秒的请求数限制为一定数量，例如 10 个（示例代码将在没有 API 密钥的情况下将您限制为 3 个）。我使用 Furrr 的 future_* 函数来并行化请求以尽快获取它们，如下所示：

library(tidyverse)
library(rentrez)
library(furrr)

plan(multiprocess)

api_key <- "<api key>"
# this will return a crap-ton of results
srch <- entrez_search("nuccore", "Homo sapiens", use_history=T, api_key=api_key)

total <- srch$count
per_request <- 500 # get 500 records per parallel request
nrequest <- total %/% per_request + as.logical(total %% per_request)

result <- future_map(seq(nrequest),function(x) {
  rstart <- (x - 1) * per_request
  return(entrez_fetch(
    "nuccore",
    web_history = srch$web_history,
    rettype="fasta",
    retmode="xml",
    retstart=rstart, …