Overfull multi-processing pipe() causing delays between socket client and dash client

Viewed 125

I am communicating the last return from my server through the client to a dashboard. Both the client and the dashboard are running as a separate process, using multiprocessing. Currently, I am communicating the received package through a pipe(). When displaying the plotly dash-dashboard, I am using @app.callback with a 1 second interval.

This results in the pipe being cramped and my values being updated with a huge delay.

Dashboard code

def dashboard():
    app = dash.Dash()
    app.layout = html.Div([
        dcc.Interval(
            id="interval-components",
            interval=1*1000,
            n_intervals=0
        ),

        # dcc.Graph(figure=fig)
        daq.Gauge(
            id='gauge-chart',
            color={"gradient": True, "ranges": {
                "green": [0, 10], "yellow": [10, 30], "red": [30, 60]}},
            value=0,
            max=60,
            min=0,
            units="M/S",
        )
    ], className='row', style={'textAlign': 'center'})
    @app.callback(
        output=[Output('gauge-chart', 'value')],
        inputs=[Input('interval-components', 'n_intervals')]
    )
    def update_gauge(n):
        value = receive_packet(child_conn)[0][2]
        return [value]

    app.run_server(port=8047,debug=True)

def receive_packet(conn):
    packet = conn.recv()
    packet = list(map(int, packet))
    return [packet]

Multiprocessing

    parent_conn, child_conn = multiprocessing.Pipe()
    p1 = multiprocessing.Process(target=get_data, args=(parent_conn, address,))
    p2 = multiprocessing.Process(target=dashboard, args=())
    p3 = multiprocessing.Process(target=receive_packet, args=(child_conn, ))

    p1.start()
    p2.start()
    p3.start()

    p1.join()
    p2.join()
    p3.join()

get_data sending into the pipe

#SOME CODE TO RECEIVE THE PACKAGES FROM THE SERVER
while True:
   conn.send([can_id, an1, anVar, connected, time_since_valid])

Is there a better way to share data between processes, without blocking the stream? Maybe shared states?

FAILED FIX 1

I already tried to fix it, by letting the get_data process time.sleep(1), to match the interval. This doesn't seem to work.

FAILED FIX 2

I tried to add some I am ready messages, but this does not solve the problem:

def receive_packet(conn = child_conn):
    conn.send("GIMMI DATA")
    packet = conn.recv()
    conn.send("0")
    packet = list(map(int, packet))
    return [packet]


def get_data(conn = parent_conn):
    #################
    ####SOME CODE####
    #################
    packet = conn.recv()
    if packet == "GIMMI DATA":
        conn.send([can_id, an1, anVar, connected, time_since_valid])
    else:
        pass
1 Answers

SUCCESFUL FIX 1

Instead of running multiprocessing.Pipe() I fell back to using multiprocessing.Value(), which seems to be way more stable for constant datastreams. Suggestions for more efficient solutions are always welcome.

Related