Load Management for client requests

[wachsylon]

Hi,

if a user wants to retrieve a bigger part of a dataset which includes multiple chunks, they are requested all at once. Depending on the processing on the server side, this leads to high memory consumption which probably blocks all further requests. Is it possible to limit the requests per user by adapting the configuration on the server side?

Best and thanks a lot,
Fabi

[mpiannucci]

This is a good but also hard question to pin down. Can you give specifics on

1. Your configuration and how you are spinning up xpublish
2. An example workflow: are you just opening the server dataset as zarr?

Thanks!

[abkfenris]

Right now there isn't anything built into Xpublish itself for rate limiting. Since it's FastAPI under the hood, it's possible to use it's libraries like fastapi-limiter to do rate limiting by injecting a limiter into route dependencies.

rest = xpublish.Rest(...)
app = rest.app

@app.on_event("startup")
async def startup():
    redis = redis.from_url("redis://localhost", encoding="utf-8", decode_responses=True)
    await FastAPILimiter.init(redis)

for route in app.routes:
    if route.path.startswith("/datasets"):  # or another check to see if a route is one that should be limited
        route.dependencies.append(Depends(RateLimiter(times=2, seconds=5)))

I'd like to continue expanding the plugin system, so that we could have plugins modify other routes to support things like limiting, but I wanted to see what uses folks would have before taking a swing at it.

[wachsylon]

I tried this one but I could not connect to the url:

INFO:     Started server process [513403]
INFO:     Waiting for application startup.
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/starlette/routing.py", line 677, in lifespan
    async with self.lifespan_context(app) as maybe_state:
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/starlette/routing.py", line 566, in __aenter__
    await self._router.startup()
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/starlette/routing.py", line 654, in startup
    await handler()
  File "/home/k/k204210/host_dynamic_dsets.py", line 99, in startup
    await FastAPILimiter.init(redisurl)
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/fastapi_limiter/__init__.py", line 85, in init
    cls.lua_sha = await redis.script_load(cls.lua_script)
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/redis/asyncio/client.py", line 518, in execute_command
    return await conn.retry.call_with_retry(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/redis/asyncio/retry.py", line 59, in call_with_retry
    return await do()
           ^^^^^^^^^^
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/redis/asyncio/client.py", line 492, in _send_command_parse_response
    return await self.parse_response(conn, command_name, **options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/redis/asyncio/client.py", line 539, in parse_response
    response = await connection.read_response()
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/redis/asyncio/connection.py", line 782, in read_response
    response = await self._parser.read_response(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/redis/asyncio/connection.py", line 262, in read_response
    response = await self._read_response(disable_decoding=disable_decoding)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/volume/envs/eerie_io/lib/python3.11/site-packages/redis/asyncio/connection.py", line 308, in _read_response
    raise InvalidResponse(f"Protocol Error: {raw!r}")
redis.exceptions.InvalidResponse: Protocol Error: b'HTTP/1.1 400 Bad Request'

Is this redis something I have to set up first?

[wachsylon]

I managed it to work after starting a redis service. However, it seems like this only reduces the accesses per route. What we need is a limit for requests per user, not per route. As each chunk is a different route, one user can request 100 different routes at once which could not be prevented by limiting the route requests, right?

[wachsylon]

ok it is obviously per user too, but it still looks like each chunk is a unique route preventing this rate-limit from working.

[abkfenris]

Hmm, while it looks like to can configure how the library identifies uniqueness, I missed that it currently doesn't work with sync endpoints, which most of Xpublish's endpoints are.

Using a middleware like asgi-ratelimit (example) or doing the limiting at the Nginx reverse proxy level probably will work better for global per-user limits.

[abkfenris]

Rate limiting is also something that is often well addressed by your reverse-proxy server, which will save Xpublish/FastAPI from needing to make any decisions.

For instance if you're deploying with Kubernetes and using ingress-nginx it has a lot of fine grained controls for limiting.

[wachsylon]

thank you for your quick answers. In production, xpublish runs behind a nginx reverse-proxy. I am a beginner in server deployment.
I want to use it to host a live-view of existing kerchunk files. As discussed in 75, I therefore run

collection = xp.Rest([], cache_kws=dict(available_bytes=0))

collection.register_plugin(DynamicKerchunk())

and the plugin does a lot of glob.glob for given name templates.

So regarding the memory management: Do I understand correctly that when I open a file with xarray on the server side, the chunks are decompressed first independently of how I send them to the clients? Even if the compressor do not change (not my use case)?

I assume that it is most performant to send compressed chunks because this reduces traffic. Would you agree?

[wachsylon]

1. Ratelimit:

I decided to use the nginx rate limiter. After experimenting a bit with the limit_req directive in nginx, I find that the number of requests are a problem for a necessary queue which is created with the burst parameter. If I terminate the loading of a dataset (>100 chunks), I cannot access the server for a long time. I guess the requests are somehow still in the queue and it takes too long to terminate them.

So I decided to set limit_conn addr 4; which should allow each users to have 4 concurrent connections to the server. This requires that users do not send all gets at once. All requests sent after the 4th will get an error as response. Unfortunately, a to_zarr command does not recognize these errors but instead has wrong chunks in the output.
I think that a dask cluster with 4 threads should be able to fulfill this requirement. At least it worked in my tests.

2. Chunk-memory-disk relations:

On the server I use dask and target at chunks with size 100MB. In my test case, I open a huge netcdf dataset on the server, uncompressed and kerchunked. The client (with 4 threads) directly writes a subset of the data to a disk.

• the decompressed chunks had 30 MB allocated in memory
• xpublish needed 16GB memory (50x)
• With a default xpublish compression, but adapted to BITSHUFFLE, the output chunks after a to_zarr have about 7.5MB.

The speed is 3MB/s compressed (12MB/s uncompressed).

However, there are open questions from my previous post which influences this performance.