How to prevent UI freeze in wxpython when running long task with concurrent future

Question

I want to execute a long task in a wxpython UI, without the UI losing responsiveness.

I thought using concurrent futures with a ThreadPoolExecutor would allow me to do just that, but the UI still freezes.

Here is a simple code to reproduce the problem. The UI freezes for 5 seconds.

Why is this happening and how to solve?

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

Answer 1

There are some issues in your code.

def on_long_task(event):
    with ThreadPoolExecutor() as executor:
        executor.submit(block5)

You create a ThreadPoolExecutor object inside the on_long_task function. When the function exits the ThreadPoolExecutor will be deleted because there is no more reference to it. It was just local to your function.

You should create the ThreadPoolExecutor in the init function as member of MyFrame to keep a reference to it. With the reference you can check the if block5 is ready and get the result.

Function on_long_task must be a member function of MyFrame to handle event so it should be

def on_long_task(self, event):

I have fixed your code and added a button to test if MyFrame stays responsive.

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

Answer 2

Tricky question to answer, as you don't say what the task is.

Does it run silently calculating or is it a series of steps that need to be performed.

Assuming it is a series of steps, that is what wx.GetApp().Yield() is for, temporarily passing control back to the wx mainloop, so it can remain responsive.

In this case:

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

would do the job, without bothering with ThreadPool, which appears not to relinquish control until it ends.

Note: I've never used ThreadPool and know diddly about it.