Parallel processing of API requests

[lvojeroen]

We have set up three time series A, B and C in Mesh so that C=A+B. I have written a script that sets values on A and B for one day, and then reads back C for that same day. Some performance measurements indicate that writing the values is very fast (~30 ms) and retrieving them back (including the calculation) takes a bit longer with about 800 ms.

From the logs of my script:
2022-04-04T17:34:36.720234 17072 Write time series: elapsed 30 ms
2022-04-04T17:34:36.747233 17072 Write time series: elapsed 26 ms
2022-04-04T17:34:37.588053 17072 Read time series: elapsed 840 ms
2022-04-04T17:34:37.711735 Total duration: 1183 ms / 1 runs / 1183 ms per run

Now I tried to do the same in parallel with 32 threads – so every thread writes to A and B and reads from C (including the calculation) for one day, but every thread uses a different day so there are no conflicts. Now it seems that regardless of the number of threads are started, the total time is linear with the number of threads, meaning that Mesh treats every call consecutively. The “time per run” is always about 900 ms, so two threads take 1800 ms to complete, three threads 2700 ms, and so on.

From the logs of my script with 32 threads (only last line):
2022-04-04T17:42:18.555857 Total duration: 28628 ms / 32 runs / 894 ms per run

My question is: is this expected? Can we modify parameters to make this runs in parallel or will every request always be treated in sequence? Or is this maybe related to the fact this is happing on the same time series and will requests on different time series be treated in parallel?

Below is the script for reference.
This is running on Mesh 2.3.1+23.

import datetime
import random
import threading
import uuid
from time import perf_counter

import pandas
import pyarrow
import pytz
from dateutil.relativedelta import relativedelta
from volue.mesh import Connection, Timeseries, MeshObjectId

pandas.set_option("display.max_rows", 1024)

INPUT_A = uuid.UUID("3af4ba23-a491-4741-8d1d-fb8b78118919")
INPUT_B = uuid.UUID("f71e0259-7b19-423d-84c9-ef73c320a18c")
OUTPUT = uuid.UUID("ccdcef9a-47f0-4ae8-8b21-30996474cb22")

SHOW_TIMESERIES = False
PARALLEL_PROCESSES = 2**5


def log_timestamp():
    return datetime.datetime.now().isoformat()


def elapsed_ms(perf_start):
    return int((perf_counter() - perf_start) * 1000)


def connect_to_mesh():
    address, port, root_pem_certificate = ("<server name>", 50051, "")
    connection = Connection(address, port, root_pem_certificate)
    return connection


def read_from_mesh(start, end, session, uuid_id):
    mesh_object_id = MeshObjectId.with_uuid_id(uuid_id)
    perf_start_read = perf_counter()
    timeseries = session.read_timeseries_points(
        start_time=start, end_time=end, mesh_object_id=mesh_object_id
    )
    print(
        f"{log_timestamp()} {threading.get_ident()} "
        f"Read time series: elapsed {elapsed_ms(perf_start_read)} ms"
    )
    return timeseries


def write_to_mesh(index, flags, values, session, uuid_id):
    new_timeseries = Timeseries(
        table=pyarrow.Table.from_arrays(
            [index, flags, values], schema=Timeseries.schema
        ),
        uuid_id=uuid_id,
    )
    perf_start_write = perf_counter()
    session.write_timeseries_points(new_timeseries)
    print(
        f"{log_timestamp()} {threading.get_ident()} "
        f"Write time series: elapsed {elapsed_ms(perf_start_write)} ms"
    )
    return new_timeseries


def create_random_ts(start, end, freq):
    index = pandas.date_range(start, end, freq=freq, inclusive="left")
    return (
        index,
        [Timeseries.PointFlags.OK.value] * len(index),
        [random.randint(1, 100) / 10] * len(index),
    )


def run_once(id_, start, end):
    print(
        f"{log_timestamp()} {threading.get_ident()} Running {id_} from "
        + start.astimezone(pytz.timezone("Europe/Brussels")).isoformat()
        + " to "
        + end.astimezone(pytz.timezone("Europe/Brussels")).isoformat()
    )
    perf_start_run_once = perf_counter()
    with connect_to_mesh().create_session() as session:
        ts_a = write_to_mesh(*create_random_ts(start, end, "15T"), session, INPUT_A)
        ts_b = write_to_mesh(*create_random_ts(start, end, "1H"), session, INPUT_B)
        out = read_from_mesh(start, end, session, OUTPUT)
        session.commit()
    if SHOW_TIMESERIES:
        print(ts_a.arrow_table.to_pandas())
        print(ts_b.arrow_table.to_pandas())
        print(out.arrow_table.to_pandas())
    print(
        f"{log_timestamp()} {threading.get_ident()} "
        f"Executed {id_} in {elapsed_ms(perf_start_run_once)} ms"
    )


def run():
    perf_start_full = perf_counter()
    start = (
        pytz.timezone("Europe/Brussels")
        .localize(datetime.datetime(2022, 2, 1, 0, 0, 0))
        .astimezone(pytz.utc)
    )
    end = (
        pytz.timezone("Europe/Brussels")
        .localize(datetime.datetime(2022, 2, 2, 0, 0, 0))
        .astimezone(pytz.utc)
    )

    thread_list = []
    for i in range(PARALLEL_PROCESSES):
        thread_list.append(
            threading.Thread(
                target=run_once,
                args=(
                    f"run{str(i).zfill(len(str(PARALLEL_PROCESSES)))}",
                    start + relativedelta(days=i * 2),
                    end + relativedelta(days=i * 2),
                ),
            )
        )

    for thread in thread_list:
        thread.start()

    for thread in thread_list:
        thread.join()

    elapsed = elapsed_ms(perf_start_full)
    print(
        f"{log_timestamp()} Total duration: {elapsed} ms / "
        f"{PARALLEL_PROCESSES} runs / "
        f"{int(elapsed / PARALLEL_PROCESSES)} ms per run"
    )


if __name__ == "__main__":
    run()

[jbbakeng]

My question is: is this expected? Can we modify parameters to make this runs in parallel or will every request always be treated in sequence? Or is this maybe related to the fact this is happing on the same time series and will requests on different time series be treated in parallel?

Yes. This is expected. There is no way to make this run parallelly due to how the Mesh server is implemented and at the moment there is no concrete plans to change this behavior. The result will be the same independent of which time series you write/read to.

[lvojeroen]

Okay thanks, that is clear. Can I expect that only calculations are handled one after the other but read/write requests that don't involve calculations will go in parallel? Or are all requests handled in sequence?

[jbbakeng]

You will have to expect all requests to be handled in sequence. At some point this will probably improve, but there is no concrete plans to work on this area of the Mesh server in the planned future.