[RFC] New background task API

[masenf]

The next release (0.2.8) will include support for background tasks to allow long running jobs to interact with the state without blocking UI interactions.

A background task is a special event handler which runs concurrently with other event handlers.

To avoid two event handlers modifying the state at the same time, a background task may only set state vars in a context block. Within an async with self context block an exclusive lock is acquired on the state, preventing other event handlers and background tasks from also modifying the state. After exiting the context, any updates are sent to the frontend, similar to using yield in a normal event handler.

class State(rx.State):

    counter: int = 0

    @rx.background
    async def bg_task(self):
        for ix in range(10):
            async with self:
                # state mutation is only allowed inside context block
                self.counter += 1

            # await long operations outside the context to avoid blocking
            await asyncio.sleep(0.5)

Your feedback, suggestions, and comments are welcome 🙏.

Want to test it?

pip install git+https://github.com/reflex-dev/reflex@masenf/oob-state-update

[allComputableThings]

Why not just simply:

class MyState(State):
   async def on_load(self):
      async def myTask()
         while True:
             ... do whatever. Eg. receive an infinite stream of events from a server. Manage reconnects etc....
            async with self.stateUpdateLock:  # Wait for and apply lock. Apply the updates after context exit.
                self.state_var1 = 123 
                self.state_var2 = 456
         
      asyncio.create_task(myTask())    # Self managed

You can try to manage the async task, but why? Why not allow the user to do this - they may start a task, which starts a task, which calls an API, who's response is entered into a queue, which is consumed by .... .... and finally calls something that wants to update the UI state.
Under the above I think your assumption in the that background task that does the work ALSO updates the state. Why require this?
In practice you then for the user to communicate the the background tasks via a queue that the background task must monitor.

If you want the ability for reflex to kill the task:

       self.create_background_task(myTask())      # Reflex can cancel

[allComputableThings]

My understanding the API is that
@rx.background
async def bg_task(self):

Is a singleton task. (How is it started?)

What I'd hope to avoid is having to implement a queue to communicate the results multiple async tasks workers to the UI.
For example: suppose I have a "runTask" button. I quickly click it 3 times. Each time a long running task should be started and the result appended to a list for the UI to see.
In your pattern, I don't see that I can simply create a task and place the result into UI state. I think I'd need to start the worker, run work via create_task and communicate the results to the worker (perhaps through an async queue?).

[masenf]

For example: suppose I have a "runTask" button. I quickly click it 3 times. Each time a long running task should be started

This is actually how it works currently. The tasks all run concurrently, so multiple tasks can be started and executing at once, they are NOT singleton.

The background task is started as an event trigger (on_click, on_submit, etc) or yield/return from another handler. If you wanted to limit the background tasks you'd need to keep track of how many are created and rate limit inside the task itself.

All of the background tasks launched can modify the state inside the context, so if there is a list or dict var in the State, the tasks can all independently append or update keys (just not simultaneously, they have to serialize the mutations).

So you might have a dict of lists keyed on a uuid assigned to each task, and have each task put its results in there and render them live in the frontend.

[hragbalian]

@masenf can you help me troubleshoot this.

I've set up a trigger on a background task where I have the event triggers for that task created within a foreach loop. There appears to be some blocking behavior on the new tasks such that a second trigger won't start until the first finishes, and third won't start until the second finishes, and so on.

Here is a reprex.

import asyncio

import reflex as rx

class Test2State(rx.State):
    processing: dict = dict()
    items: list[str] = ['A', 'B', 'C']
    
    @rx.background
    async def trigger_process(self, item_name: str):
        
        async with self:
            
            self.processing[item_name] = True
            yield
            
            await asyncio.sleep(5)
            
            
def item_display(item) -> rx.Component:
    
    return rx.hstack(

        rx.cond(
            ~Test2State.processing[item],
            rx.button(
                rx.icon(tag="settings"),
                item,
                size="1",
                on_click=Test2State.trigger_process(item)
            ),
            rx.chakra.spinner()
        )
    
    )
    
def interface() -> rx.Component:
    
    return rx.fragment(
        
        rx.foreach(
            Test2State.items,
            lambda item: item_display(item)
        )
    )

There is a five second delay on each of the triggered processes. I would expect that I could click B right after I click A and the process would start immediately rather than wait for A's process to complete.

Is there something I'm missing?

Thanks

[masenf]

async with self holds an exclusive lock on the state.

because the code has await asyncio.sleep(5) inside the async with self block, it's holding the lock while its sleeping, so nothing else can get in there and run.

try

    @rx.background
    async def trigger_process(self, item_name: str):
        
        async with self:
            
            self.processing[item_name] = True
            yield
            
        await asyncio.sleep(5)

any long running operations should be performed outside the async with self block.

[hragbalian]

@masenf Ohhh, yes that was it, thank you!!