标签: python-multithreading

我一直在研究 python 中的多线程一段时间，但是我对一些问题感到困惑-

1. 首先，python线程库创建的线程是用户级线程还是内核级线程？

书中说用户级线程必须映射到内核线程，操作系统只创建和维护内核级线程。

python线程库中将使用哪种线程模型？此外，谁在内核级线程和用户级线程之间做出选择？是操作系统还是程序员有发言权？

如果使用多对一模型（如图所示），我认为它不是真正的多线程，因为所有线程都映射到单个内核线程。

2. 有没有办法让操作系统在我的 python 程序中遵守某个线程模型？

3. 是否可以显示进程的所有正在运行的线程，并将它们的状态分别标记为内核级或用户级。还可以显示两个级别（用户和内核）之间的映射吗？

目前还不清楚如何Parallel在python中正确地删除joblib的工作者.其他人在这里,这里,这里和这里有类似的问题.

在我的例子中,我正在使用一个由50 joblib名工人组成的 threading后端池.

并行调用(线程):

output = Parallel(n_jobs=50, backend  = 'threading')
    (delayed(get_output)(INPUT) 
        for INPUT in list)

在这里,Parallel挂起没有错误,len(list) <= n_jobs但只有在n_jobs => -1.

为了克服这个问题,人们给予 说明如何创建一个超时装饰的Parallel功能(get_output(INPUT)使用)在上面的例子)multiprocessing:

主要功能(装饰):

@with_timeout(10)    # multiprocessing
def get_output(INPUT):     # threading
    output = do_stuff(INPUT)
    return output

多处理装饰器:

def with_timeout(timeout):
    def decorator(decorated):
        @functools.wraps(decorated)
        def inner(*args, **kwargs):
            pool = multiprocessing.pool.ThreadPool(1)
            async_result = pool.apply_async(decorated, args, kwargs)
            try:
                return …

我正在尝试对 MODIS 卫星数据进行一些分析。我的代码主要读取尺寸为 1200 x 1200 (806*1200*1200) 的大量文件 (806)。它使用 afor loop并执行数学运算。

以下是我读取文件的一般方式。

mindex=np.zeros((1200,1200))
for i in range(1200):
    var1 = xray.open_dataset('filename.nc')['variable'][:,i,:].data
    for j in range(1200):
        var2 = var1[:,j]
        ## Mathematical Calculations to find var3[i,j]## 
        mindex[i,j] = var3[i,j]

由于要处理大量数据，该过程非常缓慢，我正在考虑对其进行并行化。我尝试用 做一些事情joblib，但我一直无法做到。

我不确定如何解决这个问题。

我有两个进程(参见示例代码),每个进程都尝试访问threading.local对象.我希望下面的代码打印"a"和"b"(按任意顺序).相反,我得到"a"和"a".当我启动整个新进程时,如何优雅而强大地重置threading.local对象？

import threading
import multiprocessing
l = threading.local()
l.x = 'a'
def f():
    print getattr(l, 'x', 'b')
multiprocessing.Process(target=f).start()
f()

编辑:作为参考,当我使用threading.Thread而不是multiprocessing.Process时,它按预期工作.

我有一个调用API的脚本.为了加快脚本速度,我试图实现线程化.

当我在IDLE时,下面的脚本工作,但是当我尝试从命令行使用sys argv运行它时,我收到了下面列出的两种类型的错误.

错误1

Fatal Python error: PyImport_GetModuleDict: no module dictionary!

This application has requests the Runtime to terminate it in an unusual way.  Please         contact the application's support team for more information.

错误2

Exception in thread Thread-1 (most likely raised during iterpreter shutdown): 
Exception in thread Thread-2 (most likely raised during iterpreter shutdown):
Exception in thread Thread-3 (most likely raised during iterpreter shutdown):
Exception in thread Thread-5 (most likely raised during iterpreter shutdown):

我找不到任何关于这些错误的信息.所以,任何帮助表示赞赏.下面是处理线程的脚本部分.

import threading
import diffbot

urls …

我试图创建86个task.py实例同时运行.

import sys
import subprocess

for file in range(86):
    subprocess.call([sys.executable,'task.py',str(file)+'in.csv',str(filen)+'out.csv'])

在我开始描述我的问题之前,值得一提的是我正在使用Python 2.7.我没有检查过,但这可能与Python 3.x无关.

在使用Python的队列时,我发现了一些奇怪的东西.通常,当从队列中获取对象时,我允许长但有限的超时(例如几秒),以便在没有找到对象的情况下允许调试和错误报告.我发现的是,有时在将对象插入先前空的队列的时间与get同一个队列的方法返回该对象的时间之间存在奇怪的差距,即使该方法在该对象之前被调用put被称为该对象.

挖了一点我发现睡眠填补了这个空白.在Queue模块中,如果timeout正被传递给自变量get的方法没有None,并且是肯定的,则non_empty Condition的wait方法被称为具有正参数(即不是100%精确的;事实上,所述Queue的' _qsize’方法,该方法返回底层的长度deque首先被验证为返回0,但只要队列在第一个位置为空,接下来就是条件的等待.

该Conditions的wait方法的行为不同,如果它得到超时或不.如果它没有得到任何超时,它只是调用waiter.acquire.这是在C我理解的范围内定义的,但它似乎正常.然而,如果超时被给予,出现睡的离奇序列相反,当睡眠时间在一些任意大小(1个milisecond)开始,并随着时间的推移变得更长.这是运行的确切代码:

# Balancing act:  We can't afford a pure busy loop, so we
# have to sleep; but if we sleep the whole timeout time,
# we'll be unresponsive.  The scheme here sleeps very
# little at first, …

当调用通过Python subprocess模块花费相对较长时间的linux二进制文件时,这会释放GIL吗？

我想并行化一些从命令行调用二进制程序的代码.是否更好地使用线程(通过threading和a multiprocessing.pool.ThreadPool)或multiprocessing？我的假设是,如果subprocess发布GIL,那么选择该threading选项会更好.

更新:

这个答案表明我在2013年4月所做的事情是不可能的.然而,这似乎与Alex Martelli在Python Cookbook中所说的相矛盾(第624页,第3版):

返回时,PyGILState_Ensure()始终保证调用线程具有对Python解释器的独占访问权.即使调用C代码正在运行解释器未知的其他线程,也是如此.

文档似乎也建议GIL可以获得,这会给我带来希望(除了我不认为我可以PyGILState_Ensure()从纯python代码调用,如果我创建一个C扩展来调用它,我不知道如何嵌入我memory_daemon()的那个).

也许我误读了答案或Python Cookbook和文档.

原始问题:

我想要一个给定的线程(来自threading模块)来阻止任何其他线程在其代码的某个段执行时运行.实现它的最简单方法是什么？

显然,最大限度地减少其他线程中的代码更改,避免使用C和直接操作系统调用,并使其成为Windows和Linux的跨平台.但实际上,我很乐意为我的实际环境提供任何解决方案(见下文).

环境:

• CPython的
• python 3.4(如果有帮助,可以升级到3.5)
• Ubuntu 14.04

使用案例:

出于调试目的,我计算所有对象使用的内存(如报告所示gc.get_objects()),并打印一些摘要报告sys.stderr.我在一个单独的线程中执行此操作,因为我希望从其他线程异步传递此摘要; 我放在循环time.sleep(10)的末尾,while True进行实际的内存使用计算.但是,内存报告线程需要一段时间来完成每个报告,并且我不希望所有其他线程在内存计算完成之前继续前进(否则,内存快照将很难解释).

示例(澄清问题):

import threading as th
import time

def report_memory_consumption():
  # go through `gc.get_objects()`, check their size and print a summary
  # takes ~5 min to run

def memory_daemon():
  while True:
    # all other threads should not do anything until this call …

我浪费了很多时间,但找不到解决方案.
如果我在使用uwsgi部署的应用程序中使用线程,则它们不同步.

这里是一个简单的代码示例(wsgi.py):

from time import sleep
import threading

i = 0
def daemon():
  global i
  while True:
    i += 1
    print(i)
    sleep(3)
th = threading.Thread(target=daemon, args=())
th.start()

def application(environ, start_response):
  start_response('200 OK', [('Content-Type','text/html')])
  return [str(i).encode()]

当我运行这个应用程序i增加日志,但我总是得到1来自浏览器的make请求.(或者0如果我sleep(3)在i第一次增量之前移动)
我尝试了uwsgi.thread装饰,但得到了相同的结果.

uwsgi配置:

[uwsgi]
socket = 127.0.0.1:3034
plugins-dir = /srv/uwsgi
plugin = python34
uid = py3utils
gid = py3utils
chdir = /srv/python/3/py3utils/tht/app/
wsgi-file = wsgi.py
enable-threads = true
daemonize = %(chdir)/../uwsgi.log
master = true …