Как предотвратить зависание пользовательского интерфейса в wxpython при выполнении длинной задачи с параллельным будущим

Я хочу выполнить длинную задачу в пользовательском интерфейсе 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()
Почему в Python есть оператор "pass"?
Почему в Python есть оператор "pass"?
Оператор pass в Python - это простая концепция, которую могут быстро освоить даже новички без опыта программирования.
Некоторые методы, о которых вы не знали, что они существуют в Python
Некоторые методы, о которых вы не знали, что они существуют в Python
Python - самый известный и самый простой в изучении язык в наши дни. Имея широкий спектр применения в области машинного обучения, Data Science,...
Основы Python Часть I
Основы Python Часть I
Вы когда-нибудь задумывались, почему в программах на Python вы видите приведенный ниже код?
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
LeetCode - 1579. Удаление максимального числа ребер для сохранения полной проходимости графа
Алиса и Боб имеют неориентированный граф из n узлов и трех типов ребер:
Оптимизация кода с помощью тернарного оператора Python
Оптимизация кода с помощью тернарного оператора Python
И последнее, что мы хотели бы показать вам, прежде чем двигаться дальше, это
Советы по эффективной веб-разработке с помощью Python
Советы по эффективной веб-разработке с помощью Python
Как веб-разработчик, Python может стать мощным инструментом для создания эффективных и масштабируемых веб-приложений.
0
0
85
2
Перейти к ответу Данный вопрос помечен как решенный

Ответы 2

Сложный вопрос, так как вы не говорите, в чем заключается задача.

Выполняется ли он молчаливо, или это серия шагов, которые необходимо выполнить.

Предполагая, что это серия шагов, именно для этого и нужен 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 23.02.2024 08:41

@Matthis Подробности, подробности, подробности. Как вы предлагаете копировать файл и откуда куда?

Rolf of Saxony 23.02.2024 10:00
Ответ принят как подходящий

В вашем коде есть некоторые проблемы.

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()

Другие вопросы по теме