Async

loon/mqtt/handlers/world.py

@@ -22,7 +22,7 @@ async def world_handler(client, userdata, msg, data):
 
     if settings.redis.enabled:
         try:
-            set_chunk_cache(x, z, data)
+            await set_chunk_cache(x, z, data)
         except redis.RedisError:
             pass
 

loon/redis/__init__.py

@@ -1,4 +1,4 @@
-import redis
+import redis.asyncio as redis
 from redis.backoff import ExponentialBackoff
 from redis.exceptions import ConnectionError, TimeoutError
 from redis.retry import Retry

loon/redis/chunk.py

@@ -4,16 +4,23 @@ from config import settings
 from loon.redis import r
 
 
-def set_chunk_cache(x: int, z: int, chunk_payload):
+async def set_chunk_cache(x: int, z: int, chunk_payload):
-    r.setex(f'{x}:{z}', settings.redis.expire_time, str(json.dumps(chunk_payload)))
+    await r.setex(f'{x}:{z}', settings.redis.expire_time, str(json.dumps(chunk_payload)))
 
-def get_chunk_cache(x: int, z: int):
+async def get_chunk_cache(x: int, z: int):
-    chunk = r.get(f'{x}:{z}')
+    chunk = await r.get(f'{x}:{z}')
 
     if chunk:
         return json.loads(chunk)
 
     return None
 
-def reset_chunk_cache(x: int, z: int):
+async def get_chunks_cache(coords):
-    return r.delete(f'{x}:{z}')
+    if not coords:
+        return {}
+
+    vals = await r.mget([f'{x}:{z}' for x, z in coords])
+    return {c: json.loads(v) for c, v in zip(coords, vals) if v}
+
+async def reset_chunk_cache(x: int, z: int):
+    return await r.delete(f'{x}:{z}')

loon/web/world/router.py

@@ -5,7 +5,7 @@ from fastapi import APIRouter, Request, HTTPException
 from starlette import status
 
 from config import settings
-from loon.redis.chunk import get_chunk_cache, reset_chunk_cache
+from loon.redis.chunk import get_chunks_cache, reset_chunk_cache
 from loon.web import get_mqtt_manager
 from loon.web.auth.middleware import authenticated
 from loon.web.users.state import user_threads, user_world_requests
@@ -45,29 +45,35 @@ async def request_world(
     misses = []
     thread = user_threads.get(uuid)
 
-    try:
+    all_coords = [
-        for x in range(x_start, x_end + 1):
+        (x, z)
-            for z in range(z_start, z_end + 1):
+        for x in range(x_start, x_end + 1)
-                cached_chunk = get_chunk_cache(x, z)
+        for z in range(z_start, z_end + 1)
+    ]
 
-                if cached_chunk is not None:
+    try:
-                    if thread:
+        hits = await get_chunks_cache(all_coords)
-                        await thread.put(
+    except redis.exceptions.RedisError:
-                            json.dumps({"topic": f"world/chunk/{x}/{z}", "payload": cached_chunk})
-                        )
-                else:
-                    if wanted is not None:
-                        wanted.add((x, z))
-                    misses.append((x, z))
-    except redis.RedisError:
         if wanted is not None:
-            for x in range(x_start, x_end + 1):
+            wanted.update(all_coords)
-                for z in range(z_start, z_end + 1):
-                    wanted.add((x, z))
 
         get_mqtt_manager().publish(f"loon/world/chunks/{x_start}:{x_end}/{z_start}:{z_end}/request")
         return 200
 
+    for coord in all_coords:
+        cached_chunk = hits.get(coord)
+
+        if cached_chunk is not None:
+            if thread:
+                x, z = coord
+                await thread.put(
+                    json.dumps({"topic": f"world/chunk/{x}/{z}", "payload": cached_chunk})
+                )
+        else:
+            if wanted is not None:
+                wanted.add(coord)
+            misses.append(coord)
+
     for xa, xb, za, zb in miss_row_rects(misses):
         get_mqtt_manager().publish(f"loon/world/chunks/{xa}:{xb}/{za}:{zb}/request")