Asynchronous logic

[thedimlebowski]

How can I handle asynchronous logic?
For instance, how can I create a button 'sleep' which writes to a div 'sleeping' for a given amount of time, and when the process is finished, writes 'awake'?
I can create a callback launching the asynchronous process with an event on the button, and write 'sleeping' to the div, but how can I change to 'awake' when the process finishes?

[chriddyp]

Great question @thedimlebowski . Here's one way to do it. A few things to note:
1 - The request that runs the long process is blocking. You could modify this to use multiprocessing instead to run it asynchronously or something like celery
2 - When the process starts, it sets a semaphore (a "flag") by writing to the disk to prevent other processes from running. If multiple processes can run at the same time, then you can remove that logic.
3 - An interval poll continuously checks if the process is running or not and updates the RadioItems component with the status.
4 - Since one of the processes is blocking, we're running the code with multiple processes (processes=True) to allow multiple requests at the same time. In practice, just running the app with something like gunicorn in front of it will do this for you.

import dash
from dash.dependencies import Input, Output, Event
import dash_core_components as dcc
import dash_html_components as html

import datetime
import time

class Semaphore:
    def __init__(self, filename='semaphore.txt'):
        self.filename = filename
        with open(self.filename, 'w') as f:
            f.write('done')

    def lock(self):
        with open(self.filename, 'w') as f:
            f.write('working')

    def unlock(self):
        with open(self.filename, 'w') as f:
            f.write('done')

    def is_locked(self):
        return open(self.filename, 'r').read() == 'working'

semaphore = Semaphore()

def long_process():
    if semaphore.is_locked():
        raise Exception('Resource is locked')
    semaphore.lock()
    time.sleep(7)
    semaphore.unlock()
    return datetime.datetime.now()

app = dash.Dash()

def layout():
    return html.Div([
        html.Button('Run Process', id='button'),
        dcc.Interval(id='interval', interval=500),
        dcc.RadioItems(
            id='lock',
            options=[{'label': i, 'value': i} for i in ['Running...', 'Free']]),
        html.Div(id='output')
    ])

app.layout = layout

@app.callback(
    Output('lock', 'value'),
    events=[Event('interval', 'interval')])
def display_status():
    return 'Running...' if semaphore.is_locked() else 'Free'

@app.callback(
    Output('output', 'children'),
    events=[Event('button', 'click')])
def run_process():
    return 'Finished at {}'.format(long_process())

app.scripts.config.serve_locally = True

if __name__ == '__main__':
    app.run_server(debug=True, processes=5)

This example is well suited for using dash as the control panel for hardware, where only one user or session can have access to the system at a single time.

If your long running process is instead for just long-running data analysis code, I'd recommend:

• Adding a caching layer to the function that runs the process so that subsequent requests are fast. Here's the tutorial on caching: https://plot.ly/dash/performance
• Or, precomputing the results of the process and writing them to a file

---

Let me know if that helps!

[thedimlebowski]

Great. Thanks a lot for the detailed example.
I guess I wanted to avoid writing to the disk, but that might be the best option.
It did solve my issue, so many thanks!

[tantrev]

Is there any way to farm out the async callbacks to a cluster of compute nodes?

[tantrev]

Also, is there any way to utilize persistent python processes for the callbacks (similar to how RStudio Connect offers persistent R processes)?

[ned2]

@tantrev, you'd need to invoke a library that enables distributed task scheduling (such as Dask or Celery -- I know that Dask has good support for a range of distributed backends). So long as you manage the issue of the Dash callbacks being blocking, it's up to you how you go about scheduling distributed jobs kicked off by the callbacks.

[chriddyp]

We've been using celery and redis recently with great results. See https://github.com/plotly/dash-redis-celery-periodic-updates/ for one example

[cswarth]

The example above is out of date - here is an updated version of the same example that works with dash version 1.17.0

import dash
from dash.dependencies import Input, Output
import dash_core_components as dcc
import dash_html_components as html
import logging

import datetime
import time

class Semaphore:
    def __init__(self, filename='semaphore.txt'):
        self.filename = filename
        with open(self.filename, 'w') as f:
            f.write('done')

    def lock(self):
        with open(self.filename, 'w') as f:
            f.write('working')

    def unlock(self):
        with open(self.filename, 'w') as f:
            f.write('done')

    def is_locked(self):
        return open(self.filename, 'r').read() == 'working'

semaphore = Semaphore()

def long_process():
    if semaphore.is_locked():
        raise Exception('Resource is locked')
    semaphore.lock()
    time.sleep(7)
    semaphore.unlock()
    return datetime.datetime.now()

app = dash.Dash()
app.logger.setLevel(logging.DEBUG)

def layout():
    return html.Div([
        html.Button('Run Process', id='button'),
        dcc.Interval(id='interval', interval=500),
        dcc.RadioItems(
            id='lock',
            options=[{'label': i, 'value': i} for i in ['Running...', 'Free']]),
        html.Div(id='output')
    ])

app.layout = layout

@app.callback(
    Output('lock', 'value'),
    [Input('interval', 'n_intervals')])
def display_status(interval):
    app.logger.debug("display_status")
    return 'Running...' if semaphore.is_locked() else 'Free'

@app.callback(
    Output('output', 'children'),
    [Input('button', 'n_clicks')])
def run_process(button_input):
    app.logger.debug("run_process")
    return 'Finished at {}'.format(long_process())

app.scripts.config.serve_locally = True

if __name__ == '__main__':
    app.run_server(debug=True)