Fix slow receive when using native WebSocket

[jprochazk]

What

• Closes Receiving over websocket is slow when large number of chunks-per-sec #7739
• Depends on Fix blocking receiver sleeping after every read ewebsock#48

Tested using https://github.com/slyfox3/RerunDrawLinesPerf/tree/main adapted to not use click:

Testing script (expand)

import subprocess
import time

import rerun as rr

from rerun.datatypes import RotationAxisAngle


def one_strip(number_points: int):
    return rr.LineStrips3D(
        [[0.3 if i % 2 == 0 else -0.3, i, 0] for i in range(number_points)],
    )


def rotate(theta: float):
    return rr.Transform3D(
        rotation_axis_angle=RotationAxisAngle(axis=[0, 0, 1], degrees=theta),
        axis_length=0,
    )


def draw_lines(lines_count: int, pts_per_line: int):
    theta_delta = 360 / lines_count
    zig_zag = one_strip(pts_per_line)
    for i in range(lines_count):
        rr.log(
            f"line_{i}",
            rotate(theta_delta * i),
            static=True,
        )
        rr.log(
            f"line_{i}/strip",
            zig_zag,
            static=True,
        )


def main(ws_port: int = 9877, count: int = 500, pts_per_line: int = 10):
    # Prep Rerun
    rr.init("Many Lines", spawn=ws_port < 0)
    if ws_port > 0:
        rr.serve(
            open_browser=True,
            ws_port=ws_port,
        )
    else:
        rr.connect()

    # Now, render
    draw_lines(count, pts_per_line)

    print("Done drawing lines.")
    print("Press Ctrl+C to exit.")
    while True:
        time.sleep(0.5)


if __name__ == "__main__":
    main()

I don't have concrete numbers, but the difference is so significant that it's easier to just show it in videos:

c63bbea:

2024-10-23.13-15-11.webm

This PR:

2024-10-23.13-14-21.webm

Checklist

• I have read and agree to Contributor Guide and the Code of Conduct
• I've included a screenshot or gif (if applicable)
• I have tested the web demo (if applicable):
  • Using examples from latest main build: rerun.io/viewer
  • Using full set of examples from nightly build: rerun.io/viewer
• The PR title and labels are set such as to maximize their usefulness for the next release's CHANGELOG
• If applicable, add a new check to the release checklist!
• If have noted any breaking changes to the log API in CHANGELOG.md and the migration guide

• PR Build Summary
• Recent benchmark results
• Wasm size tracking

To run all checks from main, comment on the PR with @rerun-bot full-check.

[slyfox3]

Thanks for the quick fix! It'd be nice if we include some perf check in the automated test.

[Famok]

@jprochazk I'm currently experimenting with multiple processes sending to one rerun --serve using the code below.
This works fine but it seems like it will only add data from one process at a time if I'm feeding from more than one.

Is this related to this fix or is it something different?

import rerun as rr
import numpy as np


rec = rr.new_recording("my_app", recording_id="rec1",
                       make_default=True,
            make_thread_default=True,
            default_enabled=True,
            spawn=False,)
rr.spawn(connect=False, recording=rec)
rec.connect()

# generate some data for testing
times = np.linspace(0, 1, 10000)
y = np.random.rand(len(times))


# Log the data iteratively
for i,t in enumerate(times):
    rr.set_time_seconds('sensor_time', t)
    rr.log('anotherval', rr.Scalar(y[i]))

# Or log all data points at once
rr.send_columns(
    f"myval",
    times=[rr.TimeSecondsColumn("sensor_time", times)],
    components=[rr.components.ScalarBatch(y), rr.components.ScalarBatch(y)],
)


#################################################################################
# Images
#################################################################################

times = np.linspace(0, 1, 10000)
imgs = np.random.randint(0, 255, size=(len(times), 100, 100))

# Log the data iteratively
for t, img in zip(times, imgs):
    rr.set_time_seconds("sensor_time", t)
    rr.log("image", rr.Image(img))

# Logging all Images at once is little more complicated


def log_image_batch(entity_path, times, image_batch):
    # Log the ImageFormat and indicator once, as static.
    color_model = "RGB" if len(image_batch.shape) > 3 else "L"
    format_static = rr.components.ImageFormat(
        width=image_batch.shape[2],
        height=image_batch.shape[1],
        color_model=color_model,
        channel_datatype="U8",
    )
    rr.log(entity_path, [format_static, rr.Image.indicator()], static=True)

    # Reshape the images so `ImageBufferBatch` can tell that this is several blobs.
    rr.send_columns(
        entity_path,
        times=[rr.TimeSecondsColumn("sensor_time", times)],
        components=[
            rr.components.ImageBufferBatch(image_batch.reshape(len(image_batch), -1))
        ],
    )


log_image_batch("images", times, imgs)