Trying to send a message from background worker to main thread - keeps crashing

[davepruitt]

Hey all,

I have a pretty simple app designed. The user enters a "command", a worker (running in a thread) is supposed to handle the command, and then the worker reports back its progress to the main thread.

In the current incarnation of the app, the worker thread is pretty empty. I don't have it doing anything except for periodically calling post_message. According to the documentation, post_message is thread safe and is the proper way to send messages from a thread worker to the UI (https://textual.textualize.io/guide/workers/#posting-messages).

Unfortunately, it simply isn't working. Every single time I call post_message from the worker thread, I get an exception: "Attribute Error: _prevent".

Here is my entire app as of right now:

from textual.app import App, ComposeResult
from textual.widgets import Header, Footer, TextLog, Input
from textual import on
from textual.worker import Worker, WorkerState
from textual.message import Message

import queue
import time

class LogMessage(Message):
    def __init__(self, message_obj: str) -> None:
        self.message_obj = message_obj

class TextualSample(App):

    def compose(self) -> ComposeResult:
        self.background_worker = None
        self.background_worker_msg_queue_fg_to_bg = queue.Queue()
        
        yield Header()
        yield Input(placeholder="Enter a command...",  name="PrimaryCommandInput")
        yield TextLog(name = "PrimaryTextLog", id="text_log", wrap=True)
        yield Footer()

    @on(Input.Submitted)
    def handle_input_command_submitted(self, event: Input.Submitted) -> None:
        #Retrieve the command that was submitted by the user
        user_command = event.value

        #Clear the user input in the GUI
        input = self.query_one(Input)
        input.value = ""    

        #Get the text log in the GUI
        text_log = self.query_one(TextLog)                    

        #Check to see if the user actually entered anything in their command
        if (len(user_command) > 0):
            #Now let's handle the command entered by the user
            if (user_command == "start"):
                self.start_background_worker()
            elif (user_command == "stop") or (user_command == "exit"):
                self.stop_background_worker()
            else:
                text_log.write(f"Unrecognized command: {user_command}")

    def on_worker_state_changed(self, event: Worker.StateChanged) -> None:
        #Get the text log in the GUI
        text_log = self.query_one(TextLog)
        text_log.write(f"Background worker event state changed: {event.state}")

    def on_log_message(self, event: LogMessage) -> None:
        #Get the text log in the GUI
        text_log = self.query_one(TextLog)
        text_log.write(event.message_obj)

    def start_background_worker(self) -> None:
        #Get the text log in the GUI
        text_log = self.query_one(TextLog)

        #Check to see if a background worker already exists
        if ((hasattr(self, "background_worker")) and 
            ((self.background_worker is None) or 
             (self.background_worker.state == WorkerState.CANCELLED) or
             (self.background_worker.state == WorkerState.SUCCESS) or 
             (self.background_worker.state == WorkerState.ERROR))):
            
            #Display a message to the user
            text_log.write("Starting background worker...")

            #Launch the background worker
            self.background_worker = self.run_worker(self.background_worker_dowork, thread=True, exclusive=True)
        else:
            #Display a message to the user
            text_log.write("Error: unable to start because the worker is already running.")

    def stop_background_worker(self) -> None:
        #Get the text log in the GUI
        text_log = self.query_one(TextLog)

        if (hasattr(self, "background_worker") and (self.background_worker is not None) and (self.background_worker.state == WorkerState.RUNNING)):
            #Display a message to the user
            text_log.write("Stopping background worker...")

            #Cancel the background worker
            self.background_worker_msg_queue_fg_to_bg.put("stop")
        else:
            #Display a message to the user
            text_log.write("Error: unable to stop worker because no worker is running.")

    def background_worker_dowork(self) -> None:
        while(True):
            try:
                data = self.background_worker_msg_queue_fg_to_bg.get(block=False)
                if (data == "stop"):
                    break
            except:
                pass

            self.post_message(LogMessage("Background thread debugging pulse"))
            time.sleep(1)
            

if __name__ == "__main__":
    app = TextualSample()
    app.run()     

Any ideas on how to get this to work properly?

Also: I have tried moving the LogMessage class to be inside of the TextualSample class (and then changing the handler function name accordingly), but that did nothing.

Thanks for any help y'all can provide!

[willmcgugan]

Do you have a traceback you could share?

[willmcgugan]

NM. You for got to add super().__init__() to your custom message initialiser. If you do that, it should unblock you!

[davepruitt]

Ah, thanks for noticing that! That should help!

[willmcgugan]

BTW, you can create custom messages with dataclasses, which is less error prone:

from dataclasses import dataclass

@dataclass
class LogMessage(Message):
    message_obj: str