Я хочу выполнить длинную задачу в пользовательском интерфейсе wxpython без потери отзывчивости пользовательского интерфейса.
Я думал, что использование параллельных фьючерсов с ThreadPoolExecutor позволит мне сделать именно это, но пользовательский интерфейс все равно зависает.
Вот простой код для воспроизведения проблемы. Пользовательский интерфейс зависает на 5 секунд.
Почему это происходит и как решить?
import time
import wx
from concurrent.futures import ThreadPoolExecutor
class MyFrame(wx.Frame):
def __init__(self, *args, **kwds):
super().__init__(*args, **kwds)
self.panel = wx.Panel(self, wx.ID_ANY)
self.longtask_button = wx.Button(self, label = "Long Task")
self.longtask_button.Bind(wx.EVT_BUTTON, on_long_task)
sizer = wx.BoxSizer(wx.VERTICAL)
sizer.Add(self.longtask_button, 0, wx.ALIGN_RIGHT)
self.SetSizer(sizer)
self.Layout()
def on_long_task(event):
with ThreadPoolExecutor() as executor:
executor.submit(block5)
def block5():
time.sleep(5)
return 1
if __name__ == "__main__":
app = wx.App()
frame = MyFrame(None, wx.ID_ANY, "")
frame.Show()
app.MainLoop()
Сложный вопрос, так как вы не говорите, в чем заключается задача.
Выполняется ли он молчаливо, или это серия шагов, которые необходимо выполнить.
Предполагая, что это серия шагов, именно для этого и нужен wx.GetApp().Yield(), временная передача управления обратно основному циклу wx, чтобы он мог оставаться отзывчивым.
В этом случае:
import time
import wx
#from concurrent.futures import ThreadPoolExecutor
class MyFrame(wx.Frame):
def __init__(self, *args, **kwds):
super().__init__(*args, **kwds)
self.panel = wx.Panel(self, wx.ID_ANY)
self.longtask_button = wx.Button(self, label = "Long Task")
self.longtask_button.Bind(wx.EVT_BUTTON, self.on_long_task)
sizer = wx.BoxSizer(wx.VERTICAL)
self.other_button = wx.Button(self, label = "Push Me!")
self.other_button.Bind(wx.EVT_BUTTON, self.other_task)
sizer.Add(self.longtask_button)
sizer.Add(self.other_button)
self.SetSizer(sizer)
self.Layout()
def on_long_task(self, event):
# with ThreadPoolExecutor() as executor:
# executor.submit(self.block5)
self.block5()
def block5(self):
print("sleeping")
for i in range(11):
time.sleep(0.5)
print("yielding")
wx.GetApp().Yield()
print("Awoke")
return 1
def other_task(self, event):
print("other pushed")
if __name__ == "__main__":
app = wx.App()
frame = MyFrame(None, wx.ID_ANY, "")
frame.Show()
app.MainLoop()
выполнит эту работу, не беспокоясь о ThreadPool, который, похоже, не отказывается от управления, пока не завершится.
Примечание. Я никогда не использовал ThreadPool и плохо о нем знаю.
@Matthis Подробности, подробности, подробности. Как вы предлагаете копировать файл и откуда куда?
В вашем коде есть некоторые проблемы.
def on_long_task(event):
with ThreadPoolExecutor() as executor:
executor.submit(block5)
Вы создаете объект ThreadPoolExecutor внутри функции on_long_task. Когда функция завершает работу, ThreadPoolExecutor будет удален, поскольку на него больше нет ссылки. Это было просто локально для вашей функции.
Вам следует создать ThreadPoolExecutor в функции init как член MyFrame, чтобы сохранить ссылку на него. По ссылке вы можете проверить готовность блока 5 и получить результат.
Функция on_long_task должна быть функцией-членом MyFrame для обработки событий, поэтому она должна быть
def on_long_task(self, event):
Я исправил ваш код и добавил кнопку, чтобы проверить, продолжает ли MyFrame отвечать на запросы.
import time
import wx
from concurrent.futures import ThreadPoolExecutor, Future
def block5():
time.sleep(5)
return 1
class MyFrame(wx.Frame):
def __init__(self, *args, **kwds):
super().__init__(*args, **kwds)
self.panel = wx.Panel(self, wx.ID_ANY)
self.longtask_button = wx.Button(self, label = "Long Task")
self.longtask_button.Bind(wx.EVT_BUTTON, self.on_long_task)
self.longtask: Future = None
self.response_button = wx.Button(self, label = "RESPONSE TEST")
self.response_button.Bind(wx.EVT_BUTTON, self.on_response_test)
self.response_count = 0
sizer = wx.BoxSizer(wx.VERTICAL)
sizer.Add(self.longtask_button, 0, wx.ALIGN_RIGHT)
self.SetSizer(sizer)
self.Layout()
self.executor = ThreadPoolExecutor()
def on_long_task(self, event):
self.longtask = self.executor.submit(block5)
self.longtask_button.Enable(False)
def on_response_test(self, event):
if self.longtask:
if self.longtask.done():
self.SetTitle(f'LONG TASK READY: {self.longtask.result()}')
else:
self.response_count += 1
self.SetTitle(f'Response {self.response_count}')
if __name__ == "__main__":
app = wx.App()
frame = MyFrame(None, wx.ID_ANY, "")
frame.Show()
app.MainLoop()
Спасибо за ваш вклад. Задача у меня на самом деле длинная, т. е. скачать файл в пару мегабайт.