Оператор With и Threading: выполнение функции перед запуском

Этот вопрос является продолжением следующего вопроса: С инструкциями и потоками Python

Я экспериментировал с api потоковой передачи Python. У меня есть этот код, который работает для того, чего я хочу достичь: ----> выполнение функции перед вызовом run в потоке python.

Однако для этого мне обязательно нужно вызвать time.sleep (1) в методе run (), чтобы заставить его перейти к execute (). В противном случае поток завершится без назначения и выполнения функции. Есть ли лучший способ достичь этого типа ожидания?

from __future__ import print_function
import threading
import time
import functools
import contextlib
import thread
from threading import Lock
#import contextlib
#Thread module for dealing with lower level thread operations.Thread is limited use Threading instead.

def timeit(fn):
    '''Timeit function like this doesnot work with the thread calls'''
    def wrapper(*args,**kwargs):
        start = time.time()
        fn(*args,**kwargs)
        end = time.time()
        threadID = ""
        print ("Duration for func %s :%d\n"%(fn.__name__ +"_"+ threading.current_thread().name ,end-start))
    return wrapper

exitFlag = 0

@timeit
def print_time(counter,delay):
    while counter:
        if exitFlag:
            thread.exit()
        time.sleep(delay)
        print("%s : %s_%d"%(threading.current_thread().name,time.ctime(time.time()),counter))
        counter -= 1

class Mythread(threading.Thread):
    def __init__(self,threadID,name):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self._f = None

    def run(self):
        print("Starting%s\n" % self.name)
        time.sleep(1)
        if self._f:
            self._f()
            print("Exiting%s\n" % self.name)
        else:
            print("Exiting%s without function execution\n" % self.name ) 

#     def set_f(self,f):
#         self._f = f

    def execute(self,f,*args,**kwargs):
        self._f=functools.partial(f,*args,**kwargs)

    def __enter__(self):
        self.start()

    def __exit__(self,type,value,traceback):
        self.join()




class ThreadContainer(object):
    def __init__(self,id,name):
        self._t = Mythread(id,name)

    def execute(self,f,*args,**kwargs):
        self._f=functools.partial(f,*args,**kwargs)
        self._t.set_f(self._f)
#        self._t.start()
#         self._t.join()


    def __enter__(self):
        self._t.start()

    def __exit__(self,type,value,traceback):
        self._t.join()




if __name__ == '__main__':
    '''
    print_time(5, 1)
     threadLock = threading.Lock()
     threads = []
     thread1 = Mythread(1,"Thread1",5,1)
     thread2 = Mythread(2,"Thread2",5,2)
     thread1.start()
     thread2.start()
     threads.append(thread1)
     threads.append(thread2)
     for t in threads:
         t.join()
    '''
#     thread1 = Mythread(1,"Thread1")
#     thread2 = Mythread(2,"Thread2")
#     with contextlib.nested(ThreadContainer(1,"Thread1"),ThreadContainer(2,"Thread2")) as (t1,t2):
#         t1.execute(print_time,5,1)
#         t2.execute(print_time,5,2)
    t1 = Mythread(1,"Thread1")
    t2 = Mythread(2,"Thread2")
    with contextlib.nested(t1,t2):
        t1.execute(print_time,5,1)
        t2.execute(print_time,5,2)


    print("Exiting main thread ")

python multithreading asynchronous with-statement contextmanager
person Rahuketu86    schedule 06.05.2013    source источник
comment
На ваши вопросы было бы намного легче ответить, если бы вы избавились от них - избавьтесь от закомментированного кода, классов, которые вы никогда не создаете, и т. Д., Если они не имеют отношения к вопросу, и предоставьте минимальный пример, который показывает, что вы ' спрашиваете (в идеале о том, что не нужно прокручивать, чтобы прочитать).   -  person abarnert    schedule 07.05.2013

Ответы (1)


arrow_upward
1
arrow_downward

Проблема здесь в том, что вы хотите, чтобы функция run ожидала вызова функции execute.

Конечно, очевидным решением является вызов execute перед вызовом start:

t1.execute(print_time,5,1)
t2.execute(print_time,5,2)
with contextlib.nested(t1, t2):
    pass

… Или просто execute вызов start, или передайте функцию в конструктор, или вызов start, или ...

Кроме того, ваш предполагаемый дизайн немного странный. Функция потока предназначена для обработки случая, когда _f не был установлен ... но вы хотите, чтобы он ждал, пока _f не будет установлен?

Но вполне возможно, что такая проблема могла бы возникнуть в более реалистичном дизайне, поэтому давайте посмотрим, как ее решить.

Во-первых, добавление sleep для решения проблемы многопоточности почти всегда является признаком того, что вы делаете что-то очень неправильно. Это также отличный способ вызвать ужасные проблемы с производительностью (например: к тому времени, когда вы добавите достаточно sleep в достаточном количестве мест, чтобы все в основном работало, вашему приложению потребуется 30 секунд вместо 30 миллисекунд) - и, что еще хуже, ошибки состояния гонки (конечно, 1 секунда - это всегда достаточно времени, не так ли? если только компьютер не выполняет сбой подкачки, или не выходит из спящего режима, или занят другими программами, использующими весь ЦП, или…).

Если вы пытаетесь синхронизировать действия между потоками, вам необходимо использовать объект синхронизации. Уловка состоит в том, чтобы знать правильный. Прочтите документы для Lock через _ 14_ (а 3.x добавляет _ 15_) и найдите руководство по потоковой передаче в целом, чтобы получить более широкое представление о том, для чего все эти вещи нужны. *

В этом случае у вас есть код, который ожидает некоторых изменений в сохраненном состоянии, и другой код, который вносит это изменение, что является типичным вариантом использования 'Condition'. Так:

class Mythread(threading.Thread):
    def __init__(self, threadID, name, condition):
        self.condition = condition
        # ... same as before

    def run(self):
        # ... setup before checking for _f

        with self.condition:
            while not self._f:
                self.condition.wait()
        self._f()

        # ... anything else you want

Теперь вам нужно создать Condition, передать его потокам и notify это.

Вы можете использовать один Condition:

condition = threading.Condition()
t1 = Mythread(1, "Thread1", condition)
t2 = Mythread(2, "Thread2", condition)
with contextlib.nested(t1,t2):
    with condition:
        t1.execute(print_time, 5, 1)
        t2.execute(print_time, 5, 2)
        condition.notify_all()

В качестве альтернативы вы можете дать каждому потоку свой Condition:

class Mythread(threading.Thread):
    def __init__(self, threadID, name):
        self.condition = Condition()
        # ... same as before

# ...

t1 = Mythread(1, "Thread1")
t2 = Mythread(2, "Thread2")
with contextlib.nested(t1,t2):
    with t1.condition:
        t1.execute(print_time, 5, 1)
        t1.condition.notify()
    with t2.condition:
        t2.execute(print_time, 5, 1)
        t2.condition.notify()

Обратите внимание, что это не позволяет вам явно указать «не устанавливать» _f, но это довольно просто сделать. Например, вы можете добавить атрибут _f_set и проверить его вместо _f, чтобы кто-то мог позвонить execute(None) (а затем notify), чтобы разбудить вас и привести к случаю «нет _f».

* Предупреждение: некоторые наименования несовместимы. Есть другая вещь, также называемая «барьером», и другая вещь, также называемая «забор», и есть много вариантов «события», которые сильно отличаются от Pythons (некоторые из которых больше похожи на состояние, но не фактически может использоваться как таковая), а иногда «условная переменная» является фактическим совместно используемым состоянием, защищенным объектом синхронизации, а не объектом синхронизации, и так далее ...

person abarnert    schedule 07.05.2013