Is this the Correct/Safe Way to Use Threading and wx.CallAfter

[PyWoody]

Hi, I want to make sure I’m not shooting myself in the foot before I get too far into my current project with wxPython. The code listed below works and meets my needs but I wanted to make sure I wasn’t just getting lucky each time.

import random
import threading
import time
import wx

from concurrent.futures import ThreadPoolExecutor


class MyForm(wx.Frame):

    def __init__(self):
        wx.Frame.__init__(self, None, wx.ID_ANY, title='Thread Testing')
        self.SetInitialSize(wx.Size(600, 400))
        self.threads_created = 0
        self.Bind(wx.EVT_CLOSE, self.on_exit)
        self.executor = ThreadPoolExecutor(max_workers=3)
        self.panel = wx.Panel(self, wx.ID_ANY)
        self.results_sizer = wx.BoxSizer(wx.VERTICAL)
        self.sizer = wx.BoxSizer(wx.VERTICAL)
        btn_sizer = wx.BoxSizer(wx.HORIZONTAL)
        run_btn = wx.Button(self.panel, wx.ID_ANY, 'Add Thread')
        self.Bind(wx.EVT_BUTTON, self.run, run_btn)
        btn_sizer.Add(run_btn)
        print_btn = wx.Button(self.panel, wx.ID_ANY, 'Print')
        self.Bind(wx.EVT_BUTTON, self.print_to_terminal, print_btn)
        btn_sizer.Add(print_btn)
        self.sizer.Add(btn_sizer)
        self.sizer.Add(self.results_sizer, 0, wx.EXPAND)
        self.panel.SetSizer(self.sizer)
        self.exit = False

    def run(self, evt):
        print('Run btn pressed.')
        # Occurs in the main thread so no need for locks
        self.threads_created += 1
        # This could be a very long running task
        t = threading.Thread(
            target=self.create_and_initiate_gauge, args=(self.threads_created,)
        )
        t.start()

    def create_and_initiate_gauge(self, thread_num):
        # Doesn't modify main GUI; no need for CallAfter
        sizer = wx.BoxSizer(wx.HORIZONTAL)

        # Add Spacer
        wx.CallAfter(sizer.AddSpacer, 15)

        # Two-step Create Gauge
        gauge = wx.Gauge()
        wx.CallAfter(gauge.Create, self.panel)
        wx.CallAfter(sizer.Add, gauge, 1, wx.EXPAND)

        # Add Spacer
        wx.CallAfter(sizer.AddSpacer, 35)

        # Two-step create StaticText
        text = wx.StaticText()
        wx.CallAfter(text.Create, self.panel)
        wx.CallAfter(text.SetLabel, f'Job: {thread_num}')
        wx.CallAfter(sizer.Add, text, 0, wx.EXPAND)

        # Add Spacer
        wx.CallAfter(sizer.AddSpacer, 15)

        # Add thread created Gauge and StaticText to main GUI
        wx.CallAfter(self.results_sizer.Add, sizer, 0, wx.EXPAND)

        # Update Layout
        wx.CallAfter(self.sizer.Layout)

        # Demonstrate widget passing between threads
        self.executor.submit(self.update_gauge, gauge, text)

    def update_gauge(self, gauge, text):
        sleep = random.randint(1, 10) / 10
        rate = random.randint(10, 100) / 10
        completed = 0
        while completed < 100:
            if self.exit:
                return
            time.sleep(sleep)
            wx.CallAfter(gauge.SetValue, int(completed))
            completed += rate
        if gauge.GetValue() != 100:
            wx.CallAfter(gauge.SetValue, 100)
        wx.CallAfter(text.SetLabel, 'COMPLETED')
        wx.CallAfter(self.results_sizer.Layout)

    def print_to_terminal(self, evt):
        # Demonstrates the main GUI is not blocked
        print('Print button pressed.')

    def on_exit(self, evt):
        # Stop and cleanup all executor tasks on Exit
        self.exit = True
        self.executor.shutdown(wait=False, cancel_futures=True)
        self.Destroy()


if __name__ == '__main__':
    app = wx.App()
    frame = MyForm().Show()
    app.MainLoop()```


Thanks for any help or advice.

[Robin]

There’s probably more than a bit of luckiness happening…

Most of what’s in your create_and_initiate_gauge should be done in the main/GUI thread. In addition, it’s usually not a good idea to do many wx.CallAfter calls one after another as there are some (rare-ish) cases where they may actually be executed out of order. It’s better to refactor them all into a single method and use just one wx.CallAfter to invoke it.

So to summarize, I would move all the GUI work currently in create_and_initiate_gauge to a method that is called in run before you create the thread. You can pass the gauge to the thread if you want to manage it that way. If you really want the gauge to be conceptually part of the thread then you can use a single wx.CallAfter to invoke that method.

Similarly, in update_gauge I would move everything after the loop to a separate function (including the gauge.GetValue) into a separate method that is invoked with wx.CallAfter.

[da-dada]

@PyWoody sorry I got it wrong (must have been too late yesterday)

but there are times when the CallAfter gets puffed (nice example! )

[PyWoody]

Thank you for the responses and advice.

I put this quick demo together to work as a test-case and it’s certainly proved fruitful. I was concerned about using (or abusing) CallAfter and it showed my approach was flawed.

I ended up reevaluating my use-case and will be using another GUI Library instead.

[da-dada]

well, @PyWoody wx.CallAfter is a convenience and not a workaholic!

About your use-case I could not imagine the use, because using a future in a thread only makes sense if it does not manipulate the GUI: the responsiveness of the GUI only needs one thread level

please have a look at this somewhat more natural design of using CallAfter

import random
from threading import Thread, Event
import wx

class MyForm(wx.Frame):

    def __init__(self):
        wx.Frame.__init__(self, None, wx.ID_ANY, title='Thread Testing')
        self.SetInitialSize(wx.Size(600, 400))
        self.threads_created = 0
        self.panel = wx.Panel(self, wx.ID_ANY)
        self.results_sizer = wx.BoxSizer(wx.VERTICAL)
        self.sizer = wx.BoxSizer(wx.VERTICAL)
        btn_sizer = wx.BoxSizer(wx.HORIZONTAL)
        run_btn = wx.Button(self.panel, wx.ID_ANY, 'Add Thread')
        self.Bind(wx.EVT_BUTTON, self.create_and_initiate_gauge, run_btn)
        btn_sizer.Add(run_btn)
        print_btn = wx.Button(self.panel, wx.ID_ANY, 'Print')
        self.Bind(wx.EVT_BUTTON, self.print_to_terminal, print_btn)
        btn_sizer.Add(print_btn)
        self.sizer.Add(btn_sizer)
        self.sizer.Add(self.results_sizer, 0, wx.EXPAND)
        self.panel.SetSizer(self.sizer)
        self.exit = Event()
        self.Bind(wx.EVT_WINDOW_DESTROY, lambda _: self.exit.set())
        self.Show()

    def create_and_initiate_gauge(self, _):
        if 'thread_num' in vars(self):
            self.thread_num += 1
        else:
            self.thread_num = 1
        # Doesn't modify main GUI; no need for CallAfter
        sizer = wx.BoxSizer(wx.HORIZONTAL)

        # one-step Create Gauge
        gauge = wx.Gauge(self.panel)
        sizer.Add(gauge, 1, wx.EXPAND|wx.RIGHT, 35)

        # one-step create StaticText
        text = wx.StaticText(self.panel)
        text.SetLabel(f'Job: {self.thread_num}')
        sizer.Add(text, 0, wx.EXPAND|wx.RIGHT, 15)

        # Add thread created Gauge and StaticText to main GUI
        self.results_sizer.Add(sizer, 0, wx.EXPAND|wx.LEFT, 15)

        # Update Layout
        self.sizer.Layout()

        # Demonstrate widget passing between threads
        Counter(self.exit, self.layout_result, gauge, text).start()

    def layout_result(self):
        self.results_sizer.Layout()

    def print_to_terminal(self, _):
        # Demonstrates the main GUI is not blocked
        print('Print button pressed.')

class Counter(Thread):
    def __init__(self, evt, lot_res, gauge, text):
        Thread.__init__(self)
        self.exit = evt
        self.layout_result = lot_res
        self.gauge = gauge
        self.text = text

    def run(self):
        sleep = random.randint(1, 10) / 10
        rate = random.randint(10, 100) / 10
        completed = 0
        while True:
            self.exit.wait(sleep)
            if self.exit.is_set():
                break
            if completed < 100:
                self.gauge.SetValue(int(completed))
                completed += rate
            else:
                if self.gauge.GetValue() != 100:
                    self.gauge.SetValue(100)
                self.text.SetLabel('COMPLETED')
                wx.CallAfter(self.layout_result)
                break

app = wx.App()
MyForm()
app.MainLoop()

[PyWoody]

The benefit of the TheadPoolExecutor (TPE) is you can set max_workers=n and at max only n Threads will be created and running at any given time. You’ll notice in my demo only 3 gauges will be running at once no matter how many times you press the button. The TPE also manages thread reuse, error capturing, and futures monitoring with as_completed.

Again, I thank everyone for taking the time to help but it’s clear that my potential use case does not mesh with wxPython, which is not a knock on wxPython.

[da-dada]

well, @PyWoody, if I may be allowed to ask, where is the problem in modifying your TPE sample the same way as I factored out the CallAfter ?

and I’m sure instead of using the GUI controls directly from the Executor (which is no problem as long as things are kept apart) that can all be done via the event loop of wx: just a few wx.PyEvents plus their handlers

so what I would love to know is the potential gap left, may be helpful for my future

[da-dada]

well, if the customers don’t know what they want (as a highlight the TPE is managing thread reuse), the magic is all yours (may not all be fun though, but comments should always be welcome)

import threading as thd
import random
import wx

class Gui(wx.Frame):

    def __init__(self):
        super().__init__(None, wx.ID_ANY, size=(600, 400),
                                                            title='Thread Testing')
        self.sb = self.CreateStatusBar(style=wx.SB_FLAT)
        self.sb_timer = None
        self.pnl = wx.Panel(self)
        vbox = wx.BoxSizer(wx.VERTICAL)

        hbox = wx.BoxSizer(wx.HORIZONTAL)
        btn = wx.Button(self.pnl, wx.ID_ANY, 'Add Thread')
        btn.Bind(wx.EVT_BUTTON, self.add_gauge)
        hbox.Add(btn)
        btn = wx.Button(self.pnl, wx.ID_ANY, 'Thread Info')
        btn.Bind(wx.EVT_BUTTON, self.thread_info)
        hbox.Add(btn)
        vbox.Add(hbox)
        self.results_vbox = wx.BoxSizer(wx.VERTICAL)
        vbox.Add(self.results_vbox, 0, wx.EXPAND)

        self.pnl.SetSizer(vbox)

        self.Bind(wx.EVT_CLOSE, self.evt_close)
        self.exit = thd.Event()
        self.thread_num, self.ref_cnt = 0, 0

        self.Show()

    def add_gauge(self, _):
        hbox = wx.BoxSizer(wx.HORIZONTAL)
        gauge = wx.Gauge(self.pnl)
        hbox.Add(gauge, 1, wx.EXPAND|wx.RIGHT, 35)
        text = wx.StaticText(self.pnl)
        self.thread_num += 1
        text.SetLabel(f'Job: {self.thread_num}')
        hbox.Add(text, 0, wx.EXPAND|wx.RIGHT, 15)
        self.results_vbox.Add(hbox, 0, wx.EXPAND|wx.LEFT, 15)

        self.pnl.Layout()

        self.ref_cnt += 1
        Job(self.exit, self.layout_result, gauge, text).start()

    def layout_result(self):
        self.results_vbox.Layout()
        self.ref_cnt -= 1

    def thread_info(self, _):
        for entry in thd.enumerate():
            print(entry)

    def evt_close(self, _):
        if self.ref_cnt:
            if not self.sb_timer:
                self.sb.SetBackgroundColour('red')
                self.sb.SetStatusText('jobs are still running..')
                self.sb_timer = thd.Timer(1.5, self.auto_reset_sb)
                self.sb_timer.start()
        else:
            self.exit.set()
            if self.sb_timer:
                self.sb_timer.cancel()
            self.Destroy()

    def auto_reset_sb(self):
        self.sb.SetBackgroundColour(None)
        self.sb.SetStatusText('')
        self.sb_timer = None

class Job(thd.Thread):
    def __init__(self, evt, lot_res, gauge, text):
        super().__init__()
        self.exit = evt
        self.layout_result = lot_res
        self.gauge = gauge
        self.text = text

    def run(self):
        sleep = random.randint(1, 10) / 10
        rate = random.randint(10, 100) / 10
        completed = 0
        while True:
            self.exit.wait(sleep)
            if self.exit.is_set():
                break
            if completed < 100:
                self.gauge.SetValue(int(completed))
                completed += rate
            else:
                if self.gauge.GetValue() != 100:
                    self.gauge.SetValue(100)
                self.text.SetLabel('COMPLETED')
                wx.CallAfter(self.layout_result)
                break

app = wx.App()
Gui()
app.MainLoop()

[da-dada]

@PyWoody The proof of the pudding is in the eating (meshing or not)
NB I still muse about the use case, but the responsiveness hasn’t suffered (CallAfter & the like)
PS well, a rough ride gets the Layout into a tangle and, I’m afraid, must be queued! (that’s where the convenience of CallAfter is unbeatable)
and, naturally, there is an almost C & P version for the PPE
(but it’s always good to keep separate concerns in different modules before the project grows)
PPE.py (5.1 KB) task.py (634 Bytes)

import threading as thd
from concurrent.futures import ThreadPoolExecutor
import time
import random
import wx

def task(abort, gauge):
    sleep = random.randint(1, 10) / 10
    rate = int(random.randint(10, 100) / 10)
    completed = 0
    while completed < 100:
        time.sleep(sleep)
        completed = completed + rate
        gauge.SetValue(completed)
        if abort.is_set():
            break
    return completed

def sizer_layout(sizer):
    sizer.Layout()

class Gui(wx.Frame):

    def __init__(self):
        super().__init__(None, wx.ID_ANY, size=(600, 400),
                                                title='Thread Executor Testing')
        self.sb = self.CreateStatusBar(style=wx.SB_FLAT)
        self.sb_timer = None
        self.pnl = wx.Panel(self)
        vbox = wx.BoxSizer(wx.VERTICAL)
        self.info = wx.InfoBar(self.pnl)
        vbox.Add(self.info, 0, wx.EXPAND)

        hbox = wx.BoxSizer(wx.HORIZONTAL)
        self.btn_add_task = wx.Button(self.pnl, wx.ID_ANY, 'Add Task')
        self.btn_add_task.Bind(wx.EVT_BUTTON, self.add_task)
        hbox.Add(self.btn_add_task)
        btn = wx.Button(self.pnl, wx.ID_ANY, 'Thread Info')
        btn.Bind(wx.EVT_BUTTON, self.thread_info)
        hbox.Add(btn)
        self.btn_abort = wx.Button(self.pnl, wx.ID_ANY, 'Abort?')
        self.btn_abort.Bind(wx.EVT_BUTTON, self.abort)
        hbox.Add(self.btn_abort, 0, wx.LEFT, 50)
        vbox.Add(hbox)
        self.results_vbox = wx.BoxSizer(wx.VERTICAL)
        vbox.Add(self.results_vbox, 0, wx.EXPAND|wx.LEFT, 15)

        self.pnl.SetSizer(vbox)

        self.Bind(wx.EVT_CLOSE, self.evt_close)
        self.task_num = 0
        self.executor = ThreadPoolExecutor(max_workers=3)
        self.futures = {}                   # future: sizer
        self.evt_abort = thd.Event()

        self.Show()
        self.btn_abort.Hide()

    def add_task(self, _):
        hbox = wx.BoxSizer(wx.HORIZONTAL)
        gauge = wx.Gauge(self.pnl)
        hbox.Add(gauge, 1, wx.EXPAND|wx.RIGHT, 35)
        self.task_num += 1
        hbox.Add(wx.StaticText(
            self.pnl, label=f'task: {self.task_num}'),
            0, wx.EXPAND|wx.RIGHT, 15)
        self.results_vbox.Add(hbox, 0, wx.EXPAND)

        self.pnl.Layout()

        f = self.executor.submit(task, self.evt_abort, gauge)
        f.add_done_callback(self.done)
        self.futures[f] = hbox

    def done(self, future):
        if sizer := self.futures.get(future):
            if future.cancelled():
                val = 0
                txt = 'cancelled'
            else:
                res = future.result()
                if res < 100:
                    val = res
                    txt = 'aborted'
                else:
                    val = 100
                    txt = 'COMPLETED'
            ctrl = sizer.GetChildren()
            ctrl[0].GetWindow().SetValue(val)
            ctrl[1].GetWindow().SetLabel(txt)
            wx.CallAfter(sizer_layout, sizer)
            if val == 100 and not self.still_running():
                self.btn_abort.Hide()

    def thread_info(self, _):
        for entry in thd.enumerate():
            print(entry)

    def abort(self, _):
        self.btn_abort.Disable()
        self.evt_abort.set()
        if self.sb_timer:
            self.sb_timer.cancel()
        self.btn_abort.SetLabel('ABORTED !!!')
        self.sb.SetBackgroundColour('yellow')
        self.sb.SetStatusText("that's it!")

    def evt_close(self, _):
        if self.btn_add_task.IsEnabled():
            self.btn_add_task.Disable()
            for entry in self.futures:
                entry.cancel()
        if self.still_running():
            if not self.sb_timer:
                self.btn_abort.Show()
                self.sb.SetBackgroundColour('red')
                self.sb.SetStatusText('tasks are still running..')
                self.sb_timer = thd.Timer(1.5, self.auto_reset_sb)
                self.sb_timer.start()
        else:
            if self.sb_timer:
                self.sb_timer.cancel()
            self.thread_info(None)
            self.executor.shutdown(cancel_futures=True)
            self.info.SetShowHideEffects(
                                    wx.SHOW_EFFECT_ROLL_TO_BOTTOM,
                                    wx.SHOW_EFFECT_NONE)
            self.info.SetEffectDuration(2000)
            self.info.ShowMessage(
                'Executor shut down..', wx.ICON_INFORMATION)
            self.thread_info(None)
            self.Destroy()

    def still_running(self):
        for entry in self.futures:
            if not entry.done():
                return True

    def auto_reset_sb(self):
        self.sb.SetBackgroundColour(None)
        self.sb.SetStatusText('')
        self.sb_timer = None

app = wx.App()
Gui()
app.MainLoop()      

[da-dada]

in case anybody thought all along a pool of the multiprocessing module may be much the same as the PPE, or what is all this redundant stuff for (unusual for python ?) here is a nice discussion about it (I hope )

[da-dada]

and there are more interesting cases for using CallAfters

scrwin_pnl_test.py (414 Bytes)
scrwin_pnl_win_fgs.py (12.3 KB)