标签: python-multithreading

几个星期前我开始用Python编程,并试图使用Semaphores同步两个简单的线程,用于学习目的.这是我得到的:

import threading
sem = threading.Semaphore()

def fun1():
    while True:
        sem.acquire()
        print(1)
        sem.release()

def fun2():
    while True:
        sem.acquire()
        print(2)
        sem.release()

t = threading.Thread(target = fun1)
t.start()
t2 = threading.Thread(target = fun2)
t2.start()

但它一直打印只有1.如何对照片进行内部缩放？

我正在尝试用imaplib 创建一个邮箱检查器,它与python,队列和多线程没有gui工作得很好.

但是当我试图把一个gui,我所做的每一个功能,让gui冻结直到完成.

我从各种d​​oc(添加qthread,signal,cursorr etcc)尝试了很多东西,而且没有一个教程对我有用.

有人可以帮助我理解如何在运行函数时设置或附加文本到QtextEdit,因为它只在完成后工作.

这是我的代码:

class Checker(QtCore.QThread):
    signal = QtCore.pyqtSignal(object)

    def __init__(self, lignesmailtocheck):
        QtCore.QThread.__init__(self)
        self.lignesmailtocheck = lignesmailtocheck

    def run(self):
            lignemailtocheck = self.lignesmailtocheck.strip()                        
            maillo, passo = lignemailtocheck.split(":",1)
            debmail, finmail = maillo.split("@",1)
            setimap =["oultook.com:imap-mail.outlook.com", "gmail.com:imap.gmail.com"]
            for lignesimaptocheck in sorted(setimap):
                    ligneimaptocheck = lignesimaptocheck.strip()
                    fai, imap = ligneimaptocheck.split(":",1)                                
                    if finmail == fai:
                            passo0 = passo.rstrip()
                            try :
                                    mail = imaplib.IMAP4_SSL(imap)
                                    mail.login(maillo, passo)
                                    mailboxok = open("MailBoxOk.txt", "a+", encoding='utf-8', errors='ignore')
                                    mailboxok.write(maillo+":"+passo+"\n")
                                    mailboxok.close()
                                    totaly = maillo+":"+passo0+":"+imap                                
                                    print(maillo+":"+passo+"\n")

                                    self.send_text.emit(totaly)
                                    time.sleep(1)
                            except imaplib.IMAP4.error:                          
                                           print ("LOGIN …

我有一个程序,我目前正在使用concurrent.futures.ThreadPoolExecutor同时运行多个任务.这些任务通常受I/O限制,涉及对本地数据库和远程REST API的访问.但是,这些任务本身可以分成子任务,这也可以从并发中受益.

我希望在任务中使用concurrent.futures.ThreadPoolExecutor是安全的.我编写了一个玩具示例,它似乎有效:

import concurrent.futures


def inner(i, j):
    return i, j, i**j


def outer(i):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(inner, i, j): j for j in range(5)}
        results = []
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())
    return results


def main():
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(outer, i): i for i in range(10)}
        results = []
        for future in concurrent.futures.as_completed(futures):
            results.extend(future.result())
    print(results)


if __name__ == "__main__":
    main()

尽管这个玩具示例似乎有效,但我还是有信心这是故意的.我希望它是,因为否则使用执行程序执行任意代码是不安全的,以防它也使用concurrent.futures来利用并发.

我正在尝试编写一个在循环中创建新线程的程序,而不是等待它们完成.据我所知,如果我在线程上使用.start(),我的主循环应该继续,而另一个线程将在同一时间完成其工作

但是,一旦我的新线程启动,循环就会阻塞,直到线程完成.我是否误解了线程如何在python中运行,或者我正在做些什么愚蠢的事情.

这是我创建新线程的代码.

def MainLoop():
    print 'started'
    while 1:
        if not workQ.empty():
            newThread = threading.Thread(target=DoWorkItem(), args=())
            newThread.daemon = True
            newThread.start()
        else:
            print 'queue empty'

谢谢大家

joblib中出现此类问题的原因是什么？'多处理支持的并行循环不能嵌套在线程下面,设置n_jobs = 1'我该怎么做才能避免这样的问题？

实际上我需要实现在后台线程中运行大量计算的XMLRPC服务器,并通过从UI客户端轮询来报告当前进度.它使用基于joblib的scikit-learn.

PS:我只是简单地将线程的名称更改为"MainThread"以避免此类警告,并且一切看起来都很好(并行运行并没有问题).这种解决方法将来可能会出现什么问题？

我想在多个线程(每个CPU核心一个)中使用scipy.integrate.ode(或scipy.integrate.odeint)实例,以便一次解决多个IVP.然而文档说:" 这个集成器不是可重入的.你不能同时使用"vode"集成器的两个ode实例. "

(如果文档没有说明,odeint会在多次实例化时导致内部错误.)

知道可以做些什么吗？

操作系统:debian9.
一个名为的简单多进程程序mprocesses.py.

import os
import multiprocessing

def run_task(name):
    print("task %s (pid = %s) is running"  %(name,os.getpid()))
    while True:
        pass

if __name__ == "__main__":
    print("current process %s ." %os.getpid())
    pool = multiprocessing.Pool(processes = 2)
    for i in range(2):
        pool.apply_async(run_task,args=(i,))
    pool.close()
    pool.join()

运行python3 mprocesses.py并获得低于输出.

python3 mprocesses.py
current process 6145 .
task 0 (pid = 6146) is running
task 1 (pid = 6147) is running

获取流程信息.

ps lax |grep 'python3 mprocesses.py' |grep -v grep 
0  1000  6145  5615  20 …

我有一个函数foo,它将一个指向内存的指针作为参数,并且对该内存进行写入和读取:

cdef void foo (double *data):
   data[some_index_int] = some_value_double
   do_something_dependent_on (data)

我data这样分配:

cdef int N = some_int
cdef double *data = <double*> malloc (N * sizeof (double))

cdef int i
for i in cython.parallel.prange (N, nogil=True):
    foo (data)

readout (data)

我现在的问题是:不同的线程如何对待这个？我的猜测是指向的内存data将由所有线程共享,并在函数内部"同时"读取或写入foo.这会弄乱所有结果,因为人们不能依赖先前设定的数据值(内部foo)？我的猜测是正确的还是在cython编译器中实现了一些神奇的安全带？

非常感谢你提前.

我在thread今天的模块文档的"警告"部分中遇到了这个有趣的声明:

并非所有可能阻止等待I/O的内置函数都允许其他线程运行.(最流行的(time.sleep(),file.read(), select.select())正常工作.)

几乎所有其他地方我都见过Python线程,我们总是假设所有执行I/O的内置函数都会释放GIL,这意味着其他线程可以在函数阻塞时运行.据我所知,I/O操作阻止其他线程的唯一风险是,如果它是由一个忽略释放GIL的错误C扩展.

那么,来自thread文档的这句话实际上是真的吗？是否有任何不释放GIL的内置阻塞I/O操作？到目前为止,我还没有找到任何具体的例子.

目前还不清楚如何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 …