Running a blocking API inside textual and updating the main app thread.

[chaudron]

Hi All,

I'm trying to wrap a blocking API inside a textual GUI, and having problems communicating updates from the to_thread() to the application loop.

I've included the code below, but mainly it's the self.update_count() call from bcc_ringbuffer_task(), which, if I understand async correctly, runs from a different thread and is not updating the main event loop. I might be wrong here, as I'm a noob with async operations.

If it's called from action_togle_dark() it works fine. Anyone any idea what I'm doing wrong, and how to make this work?

Thanks,

Eelco

#!/usr/bin/python3

import asyncio

from textual.app import App, ComposeResult
from textual.widgets import Header, Footer, Static
from bcc import BPF

src = r"""
BPF_RINGBUF_OUTPUT(buffer, 1 << 4);
struct event {
    char filename[16];
    int dfd;
    int flags;
    int mode;
};

TRACEPOINT_PROBE(syscalls, sys_enter_openat) {
    int zero = 0;
    struct event event = {};
    bpf_probe_read_user_str(event.filename, sizeof(event.filename),
                            args->filename);
    event.dfd = args->dfd;
    event.flags = args->flags;
    event.mode = args->mode;
    buffer.ringbuf_output(&event, sizeof(event), 0);
    return 0;
}
"""


class UCMon(App):
    BINDINGS = [("d", "toggle_dark", "Toggle dark mode"),
                ("q", "quit", "Quit")]

    def __init__(self, bcc_callback):
        super().__init__()

        self.bcc_events = []
        self.bcc_bpf = BPF(text=src)
        self.bcc_bpf['buffer'].open_ring_buffer(bcc_callback)

    def bcc_ringbuffer_task(self) -> None:
        while True:
            self.bcc_bpf.ring_buffer_poll(timeout=500)
            self.update_count()
            if self._exit:
                break

    def compose(self) -> ComposeResult:
        yield Header()
        yield Footer()
        yield Static("START", id="event")

        # Start the bcc poll loop !
        self.bcc_task = asyncio.create_task(asyncio.to_thread(
            self.bcc_ringbuffer_task))

    def update_count(self):
        self.query_one("#event", Static).update(
            f"Xevent count {len(self.bcc_events)}")

    def action_toggle_dark(self) -> None:
        # self.dark = not self.dark
        self.update_count()


def bpf_callback(ctx, data, size):
    app.bcc_events.append(app.bcc_bpf['buffer'].event(data))


app = UCMon(bpf_callback)
app.run()


for event in app.bcc_events:
    print("%-16s %10d %10d %10d" % (event.filename.decode('utf-8'),
                                    event.dfd, event.flags, event.mode))

print(len(app.bcc_events))

[mharig]

AFAIR, I never got the asyncio.create_task working as I expected. What did work for me was my answer to my question

#264

May it help you.

Another approach, that I didn't try yet, may work, too:
Using several screens and callback_after_refresh() like in this project:

https://github.com/davep/gridinfo

[chaudron]

Thanks for your input. Unfortunately, your approach did not work. It's not updating the main refresh loop somehow :( I've even tried to store the local loop on initialisation, self.main_loop = asyncio.get_event_loop(), and use that in the other threads update function, self.main_loop.run_in_executor(None, self.update_count)

Is there something special we need to get the main loop to update the view in this case?

[chaudron]

Looks like I found the solution :) I should not use the get_event_loop() but the get_running_loop().

So in 'compose()' I do:

    def compose(self) -> ComposeResult:
        yield Header()
        yield Footer()
        yield Static("START", id="event")

        self.running_loop = asyncio.get_running_loop()

        # Start the bcc poll loop !
        self.bcc_task = asyncio.create_task(asyncio.to_thread(
            self.bcc_ringbuffer_task))

And in the bcc_ringbuffer_task() I do the following:

    def bcc_ringbuffer_task(self) -> None:
        while True:
            self.bcc_bpf.ring_buffer_poll(timeout=500)
            self.running_loop.run_in_executor(None, self.update_count)
            if self._exit:
                break

This seems to work :) I'll try to add a more working example later, and close this.

[chaudron]

I tried this with the ListView widget, and it seems like the item from the add_upcall call are not added. Any idea what is different with the Static vs ListView widget?

Here is the full code:

#!/usr/bin/python3

import asyncio

from textual.app import App, ComposeResult
from textual.widgets import Header, Footer, Static, ListView, ListItem, Label
from bcc import BPF

src = r"""
BPF_RINGBUF_OUTPUT(buffer, 1 << 4);
struct event {
    char filename[16];
    int dfd;
    int flags;
    int mode;
};

TRACEPOINT_PROBE(syscalls, sys_enter_openat) {
    int zero = 0;
    struct event event = {};
    bpf_probe_read_user_str(event.filename, sizeof(event.filename),
                            args->filename);
    event.dfd = args->dfd;
    event.flags = args->flags;
    event.mode = args->mode;
    buffer.ringbuf_output(&event, sizeof(event), 0);
    return 0;
}
"""


class UCMon(App):
    BINDINGS = [("d", "toggle_dark", "Toggle dark mode"),
                ("q", "quit", "Quit")]

    def __init__(self, bcc_callback):
        super().__init__()

        self.bcc_events = []
        self.bcc_bpf = BPF(text=src)
        self.bcc_bpf['buffer'].open_ring_buffer(bcc_callback)

    def bcc_ringbuffer_task(self) -> None:
        while True:
            self.bcc_bpf.ring_buffer_poll(timeout=500)
            self.running_loop.run_in_executor(None, self.update_count)
            self.running_loop.run_in_executor(None, self.add_upcall)
            if self._exit:
                break

    def compose(self) -> ComposeResult:
        yield Header()
        yield Footer()
        yield Static("START", id="event")
        yield ListView(ListItem(Label("One")),
                       id="upcalls")

        self.running_loop = asyncio.get_running_loop()

        # Start the bcc poll loop !
        self.bcc_task = asyncio.create_task(asyncio.to_thread(
            self.bcc_ringbuffer_task))

    def update_count(self) -> None:
        self.query_one("#event", Static).update(
            f"Xevent count {len(self.bcc_events)}")

    def add_upcall(self) -> None:
        self.query_one("#upcalls", ListView).append(ListItem(
            Label("ADD_UPCALL")))

    def action_toggle_dark(self) -> None:
        # self.dark = not self.dark
        self.update_count()
        self.query_one("#upcalls", ListView).append(ListItem(
            Label("TOGGLE_DARK")))


def bpf_callback(ctx, data, size):
    app.bcc_events.append(app.bcc_bpf['buffer'].event(data))


app = UCMon(bpf_callback)
app.run()

print(f"Total events processed: {len(app.bcc_events)}")

@willmcgugan do you have any idea if this should work, or should I use a different approach for non-asyncio aware APIs?

[mharig]

Maybe you have to call an update or refresh in the add_upcall?

[willmcgugan]

You should probably use App.call_from_thread if you are updating your app from a thread. For instance, self.call_from_thread(self.self.my_callback). Asyncio apps are not threadsafe, so this insures that the callback runs in the apps own thread.

[chaudron]

Cool this solved the problem completely, no need for doing run_in_executor()

[chaudron]

Based on the last comment, here is the full working code if someone else needs threading support for none asynchronous API.

#!/usr/bin/python3

import asyncio

from textual.app import App, ComposeResult
from textual.widgets import Header, Footer, Static, ListView, ListItem, Label
from bcc import BPF

src = r"""
BPF_RINGBUF_OUTPUT(buffer, 1 << 4);
struct event {
    char filename[16];
    int dfd;
    int flags;
    int mode;
};

TRACEPOINT_PROBE(syscalls, sys_enter_openat) {
    int zero = 0;
    struct event event = {};
    bpf_probe_read_user_str(event.filename, sizeof(event.filename),
                            args->filename);
    event.dfd = args->dfd;
    event.flags = args->flags;
    event.mode = args->mode;
    buffer.ringbuf_output(&event, sizeof(event), 0);
    return 0;
}
"""


class UCMon(App):
    BINDINGS = [("d", "toggle_dark", "Toggle dark mode"),
                ("q", "quit", "Quit")]

    def __init__(self, bcc_callback):
        super().__init__()

        self.bcc_events = []
        self.bcc_bpf = BPF(text=src)
        self.bcc_bpf['buffer'].open_ring_buffer(bcc_callback)

    def bcc_ringbuffer_task(self) -> None:
        while True:
            self.bcc_bpf.ring_buffer_poll(timeout=500)
            self.call_from_thread(self.update_count)
            self.call_from_thread(self.add_upcall)
            if self._exit:
                break

    def compose(self) -> ComposeResult:
        yield Header()
        yield Footer()
        yield Static("START", id="event")
        yield ListView(ListItem(Label("One")),
                       id="upcalls")

        # Start the bcc poll loop !
        self.bcc_task = asyncio.create_task(asyncio.to_thread(
            self.bcc_ringbuffer_task))

    def update_count(self) -> None:
        self.query_one("#event", Static).update(
            f"Xevent count {len(self.bcc_events)}")

    def add_upcall(self) -> None:
        self.query_one("#upcalls", ListView).append(ListItem(
            Label("ADD_UPCALL")))

    def action_toggle_dark(self) -> None:
        # self.dark = not self.dark
        self.update_count()
        self.query_one("#upcalls", ListView).append(ListItem(
            Label("TOGGLE_DARK")))


def bpf_callback(ctx, data, size):
    app.bcc_events.append(app.bcc_bpf['buffer'].event(data))


app = UCMon(bpf_callback)
app.run()

print(f"Total events processed: {len(app.bcc_events)}")