A thread-safe queue lets multiple threads exchange data safely. A worker thread can place progress updates into it while the main thread retrieves them and updates the user interface, avoiding direct shared-data races. section 7.4.2
A practical worker-to-UI pattern is:
- Create a shared Queue from Python’s queue module. section 7.4.2
- Have the worker thread put progress updates into the queue. section 7.4.2
- Poll the queue from the UI’s main thread and update the progress bar there. section 7.4.2
- Use Tkinter’s after method to poll approximately every 25 milliseconds. section 7.4.2
Listing 7.14 The Tkinter GUI
from queue import Queue
from tkinter import Tk
from tkinter import Label
from tkinter import Entry
from tkinter import ttk
from typing import Optional
from chapter_07.listing_7_13 import StressTest
class LoadTester(Tk):
def __init__(self, loop, *args, **kwargs): #1
Tk.__init__(self, *args, **kwargs)
self._queue = Queue()
self._refresh_ms = 25
self._loop = loop
self._load_test: Optional[StressTest] = None
self.title('URL Requester')
self._url_label = Label(self, text="URL:")
self._url_label.grid(column=0, row=0)
self._url_field = Entry(self, width=10)
self._url_field.grid(column=1, row=0)
self._request_label = Label(self, text="Number of requests:")
self._request_label.grid(column=0, row=1)
self._request_field = Entry(self, width=10)
self._request_field.grid(column=1, row=1)
self._submit = ttk.Button(self, text="Submit", command=self._start)#2
self._submit.grid(column=2, row=1)
self._pb_label = Label(self, text="Progress:")
self._pb_label.grid(column=0, row=3)
self._pb = ttk.Progressbar(self, orient="horizontal", length=200, mode="determinate")
self._pb.grid(column=1, row=3, columnspan=2)
def _update_bar(self, pct: int): #3
if pct == 100:
self._load_test = None
self._submit['text'] = 'Submit'
else:
self._pb['value'] = pct
self.after(self._refresh_ms, self._poll_queue)
def _queue_update(self, completed_requests: int, total_requests: int): #4
self._queue.put(int(completed_requests / total_requests * 100))
def _poll_queue(self): #5
if not self._queue.empty():
percent_complete = self._queue.get()
self._update_bar(percent_complete)
else:
if self._load_test:
self.after(self._refresh_ms, self._poll_queue)
def _start(self): #6
if self._load_test is None:
self._submit['text'] = 'Cancel'
test = StressTest(self._loop,
self._url_field.get(),
int(self._request_field.get()),
self._queue_update)
self.after(self._refresh_ms, self._poll_queue)
test.start()
self._load_test = test
else:
self._load_test.cancel()
self._load_test = None
self._submit['text'] = 'Submit'
Inspect the Tkinter load tester implementation, including its shared queue and refresh interval. section 7.4.2
