perf(memcached): one worker call per operation; delete_many counts removals

CLAUDE.md

@@ -96,7 +96,7 @@ Four non-abstract atomic primitives live on the base class with non-atomic fallb
 
 `validate_ttl` (in `backends/base.py`) accepts `None` or an `int` from 1 to `MAX_TTL` (2**31 - 1) and raises `TypeError` for floats/bools; `validate_delta` requires an `int` in signed 64-bit range. Both run before any I/O. Memcached's `_expiry` also rejects expiries after 2038-01-19.
 
-`delete_many(keys) -> int` is the fifth non-abstract base method: a per-key loop by default, one batched operation on Redis (`DEL`) and Memory (single lock).
+`delete_many(keys) -> int` is the fifth non-abstract base method: a per-key loop by default, one batched operation on Redis (`DEL`) and Memory (single lock). Memcached sends one acknowledged `DELETE` per key inside a single worker call and counts the ones that existed (pymemcache's `delete_many` returns `True` regardless). Every Memcached multi-step op (`increment`, `get_and_delete`, `*_if_equals`) also runs as one sync helper in one `asyncio.to_thread` call.
 
 `backends/codec.py` holds the JSON `CacheEntry` codec shared by Redis and Memcached; `decode_entry` maps a bare integer to a counter entry and every malformed value to `None`.
 

docs/BACKENDS.md

@@ -153,7 +153,12 @@ BackendProxy.set(backend)
 The synchronous pymemcache client runs in worker threads and is connection-pooled,
 so concurrent requests never share a socket. Writes wait for the server's
 acknowledgement (`default_noreply=False`), which keeps a value readable from
-any pooled connection as soon as `set()` returns.
+any pooled connection as soon as `set()` returns. Each call takes a single trip
+to a worker thread, including the multi-step
+[atomic operations](#atomic-backend-primitives).
+`delete_many()` sends one `DELETE` per key within that call and returns how many
+of the keys existed; before 0.3.8 it took one thread trip per key and returned
+how many keys it was given.
 
 When a server cannot be reached, every call that would go to it raises, and the
 server is tried again after one second. Before 0.3.8, calls in the second after a

fastapi_cachex/backends/base.py

@@ -113,9 +113,8 @@ async def delete_many(self, keys: Iterable[str]) -> int:
 
         The base implementation deletes one key at a time and reports how
         many were attempted, since ``delete`` does not say whether the key
-        existed. The memory and Redis backends override it with a single
-        batched operation that counts what was actually removed; Memcached
-        keeps this per-key loop.
+        existed. The built-in backends override it and count what was
+        actually removed.
         """
         count = 0
         for key in keys:

fastapi_cachex/backends/memcached.py

@@ -5,6 +5,7 @@
 import logging
 import time
 import warnings
+from collections.abc import Iterable
 
 from fastapi_cachex.backends.codec import decode_entry
 from fastapi_cachex.backends.codec import encode_entry
@@ -192,14 +193,18 @@ async def get_and_delete(self, key: str) -> CacheEntry | None:
             CacheXError: If concurrent writes cause the CAS deletion to exceed
                 the maximum retry attempts.
         """
+        return await asyncio.to_thread(self._get_and_delete, key)
+
+    def _get_and_delete(self, key: str) -> CacheEntry | None:
+        """Run ``get_and_delete``'s GETS/CAS loop in one worker thread."""
         prefixed_key = self._make_key(key)
         for _ in range(_CAS_MAX_RETRIES):
-            raw, cas_token = await asyncio.to_thread(self.client.gets, prefixed_key)
+            raw, cas_token = self.client.gets(prefixed_key)
             if raw is None:
                 logger.debug("Memcached GET_AND_DELETE MISS; key=%s", key)
                 return None
-            cas_result = await asyncio.to_thread(
-                self.client.cas, prefixed_key, b"", cas_token, -1, noreply=False
+            cas_result = self.client.cas(
+                prefixed_key, b"", cas_token, -1, noreply=False
             )
             if cas_result is True:
                 logger.debug("Memcached GET_AND_DELETE HIT; key=%s", key)
@@ -245,14 +250,16 @@ async def delete_if_equals(self, key: str, expected: CacheEntry) -> bool:
         immediately": it succeeds only if nothing wrote the key since ``GETS``
         read the value that was compared.
         """
+        return await asyncio.to_thread(self._delete_if_equals, key, expected)
+
+    def _delete_if_equals(self, key: str, expected: CacheEntry) -> bool:
+        """Run ``delete_if_equals``'s GETS, compare and CAS in one worker thread."""
         prefixed_key = self._make_key(key)
-        raw, cas_token = await asyncio.to_thread(self.client.gets, prefixed_key)
+        raw, cas_token = self.client.gets(prefixed_key)
         if raw is None or decode_entry(raw) != expected:
             logger.debug("Memcached DELETE_IF_EQUALS MISMATCH; key=%s", key)
             return False
-        deleted = await asyncio.to_thread(
-            self.client.cas, prefixed_key, b"", cas_token, -1, noreply=False
-        )
+        deleted = self.client.cas(prefixed_key, b"", cas_token, -1, noreply=False)
         logger.debug(
             "Memcached DELETE_IF_EQUALS %s; key=%s",
             "HIT" if deleted else "LOST RACE",
@@ -267,14 +274,22 @@ async def expire_if_equals(self, key: str, expected: CacheEntry, ttl: int) -> bo
         CAS write with the same bytes read by GETS and the new exptime.
         """
         validate_ttl(ttl)
+        # Converted up front so a ttl Memcached cannot store fails before I/O.
+        exptime = _expiry(ttl)
+        return await asyncio.to_thread(
+            self._expire_if_equals, key, expected, ttl, exptime
+        )
+
+    def _expire_if_equals(
+        self, key: str, expected: CacheEntry, ttl: int, exptime: int
+    ) -> bool:
+        """Run ``expire_if_equals``'s GETS, compare and CAS in one worker thread."""
         prefixed_key = self._make_key(key)
-        raw, cas_token = await asyncio.to_thread(self.client.gets, prefixed_key)
+        raw, cas_token = self.client.gets(prefixed_key)
         if raw is None or decode_entry(raw) != expected:
             logger.debug("Memcached EXPIRE_IF_EQUALS MISMATCH; key=%s", key)
             return False
-        updated = await asyncio.to_thread(
-            self.client.cas, prefixed_key, raw, cas_token, _expiry(ttl), noreply=False
-        )
+        updated = self.client.cas(prefixed_key, raw, cas_token, exptime, noreply=False)
         logger.debug(
             "Memcached EXPIRE_IF_EQUALS %s; key=%s ttl=%s",
             "HIT" if updated else "LOST RACE",
@@ -291,6 +306,16 @@ def _add_delta(self, prefixed_key: str, delta: int) -> int | None:
             result = self.client.incr(prefixed_key, delta, noreply=False)
         return None if result is None else int(result)
 
+    def _increment(self, prefixed_key: str, delta: int, exptime: int) -> int | None:
+        """Run ``increment``'s INCR, ADD and retried INCR in one worker thread."""
+        value = self._add_delta(prefixed_key, delta)
+        if value is None:
+            # No counter yet: ADD is atomic and a no-op when a concurrent
+            # call created it first, so the retry always finds a counter.
+            self.client.add(prefixed_key, b"0", exptime, noreply=False)
+            value = self._add_delta(prefixed_key, delta)
+        return value
+
     async def increment(self, key: str, delta: int = 1, ttl: int | None = None) -> int:
         """Atomically add ``delta`` to the counter at ``key`` (see base class).
 
@@ -305,14 +330,9 @@ async def increment(self, key: str, delta: int = 1, ttl: int | None = None) -> i
         # Converted up front so a ttl Memcached cannot store fails before I/O.
         exptime = _expiry(ttl)
         try:
-            value = await asyncio.to_thread(self._add_delta, prefixed_key, delta)
-            if value is None:
-                # No counter yet: ADD is atomic and a no-op when a concurrent
-                # call created it first, so the retry always finds a counter.
-                await asyncio.to_thread(
-                    self.client.add, prefixed_key, b"0", exptime, noreply=False
-                )
-                value = await asyncio.to_thread(self._add_delta, prefixed_key, delta)
+            value = await asyncio.to_thread(
+                self._increment, prefixed_key, delta, exptime
+            )
         except MemcacheClientError as e:
             if "non-numeric" not in str(e):
                 raise
@@ -334,6 +354,30 @@ async def delete(self, key: str) -> None:
         await asyncio.to_thread(self.client.delete, prefixed)
         logger.debug("Memcached DELETE; key=%s", key)
 
+    async def delete_many(self, keys: Iterable[str]) -> int:
+        """Remove every key in ``keys`` in one worker call; returns how many existed.
+
+        pymemcache's own ``delete_many`` always returns ``True`` and, on
+        ``HashClient``, still sends one DELETE per key, so it cannot say what
+        was removed. Each DELETE here waits for its reply instead, which costs
+        the same round trips and counts only keys that were present. A
+        duplicate key is sent once.
+        """
+        prefixed_keys = list(dict.fromkeys(self._make_key(key) for key in keys))
+        removed = await asyncio.to_thread(self._delete_each, prefixed_keys)
+        logger.debug(
+            "Memcached DELETE_MANY; requested=%s removed=%s",
+            len(prefixed_keys),
+            removed,
+        )
+        return removed
+
+    def _delete_each(self, prefixed_keys: list[str]) -> int:
+        """Delete ``prefixed_keys`` one by one; returns how many were present."""
+        return sum(
+            bool(self.client.delete(key, noreply=False)) for key in prefixed_keys
+        )
+
     async def clear(self) -> None:
         """Clear all values from cache.
 

i18n/zh-TW/docs/BACKENDS.md

@@ -111,7 +111,7 @@ BackendProxy.set(backend)
 - 超過伺服器項目大小上限（預設 1 MB，可用 `memcached -I` 調整）的值會被拒絕並拋出錯誤。`@cache` 會記錄該錯誤，並照常送出不儲存的回應（見[後端發生錯誤時](HTTP_CACHING.md#when-the-backend-fails)）；其他呼叫端則會收到該錯誤
 - 若需要依模式清除快取，請考慮使用 Redis 後端
 
-同步的 pymemcache 用戶端在工作執行緒中執行，並使用連線池，因此並行的請求絕不會共用同一個 socket。寫入會等待伺服器確認（`default_noreply=False`），因此只要 `set()` 返回，就能從連線池中的任何連線讀到該值。
+同步的 pymemcache 用戶端在工作執行緒中執行，並使用連線池，因此並行的請求絕不會共用同一個 socket。寫入會等待伺服器確認（`default_noreply=False`），因此只要 `set()` 返回，就能從連線池中的任何連線讀到該值。每次呼叫只切換到工作執行緒一次，多步驟的[原子操作](#atomic-backend-primitives)也是如此。`delete_many()` 在這一次呼叫中逐一對每個鍵送出 `DELETE`，並回傳實際存在的鍵數；0.3.8 之前它每個鍵都要切換一次執行緒，回傳的是傳入的鍵數。
 
 伺服器無法連線時，所有要送往它的呼叫都會拋出錯誤，一秒後會再嘗試連線。0.3.8 以前，失敗後一秒內的呼叫會回傳虛構的結果：`get()` 當成未命中、`set()` 沒有任何反應（寫入遺失）、`increment()` 當成新的計數器並回傳 0。設定多台伺服器時，失敗的那台會立即移出輪替，它的鍵會改由其餘伺服器處理，直到它恢復回應。
 

tests/backends/test_memcached.py

@@ -927,7 +927,9 @@ def test_expiry_up_to_2038_is_sent_as_a_timestamp(monkeypatch) -> None:
         _expiry(2**31 - 2_000_000_000)
 
 
-@pytest.mark.parametrize("operation", ["set", "set_if_absent", "increment"])
+@pytest.mark.parametrize(
+    "operation", ["set", "set_if_absent", "increment", "expire_if_equals"]
+)
 @pytest.mark.asyncio
 async def test_ttl_past_2038_is_rejected_before_io(operation: str) -> None:
     """Such a write used to succeed while Memcached dropped the item at once (#229)."""
@@ -939,3 +941,96 @@ async def test_ttl_past_2038_is_rejected_before_io(operation: str) -> None:
         await getattr(backend, operation)(*args, ttl=2**31 - 1)
     assert isinstance(backend.client, MagicMock)
     assert backend.client.method_calls == []
+
+
+@requires_memcached
+@pytest.mark.asyncio
+async def test_memcached_delete_many_counts_only_existing_keys(
+    memcached_backend: MemcachedBackend,
+) -> None:
+    """It used to report how many keys it was given, not how many it removed (#174)."""
+    entry = CacheEntry(fingerprint="e", content=b"v")
+    long_key = "k" * 300  # hashed by _make_key
+    for key in ("dm-a", "dm-b", long_key, "dm-keep"):
+        await memcached_backend.set(key, entry, 60)
+
+    removed = await memcached_backend.delete_many(
+        ["dm-a", "dm-b", long_key, "dm-a", "dm-missing"]
+    )
+
+    assert removed == 3
+    for key in ("dm-a", "dm-b", long_key):
+        assert await memcached_backend.get(key) is None
+    assert await memcached_backend.get("dm-keep") == entry
+
+
+@pytest.mark.asyncio
+async def test_memcached_delete_many_sends_each_namespaced_key_once() -> None:
+    backend = stubbed_backend()
+    backend.client.delete.side_effect = [True, False, True]
+    long_key = "k" * 300
+
+    assert await backend.delete_many(["a", "missing", "a", long_key]) == 2
+
+    sent = [call.args[0] for call in backend.client.delete.call_args_list]
+    assert sent == [
+        backend._make_key("a"),
+        backend._make_key("missing"),
+        backend._make_key(long_key),
+    ]
+    assert all(
+        call.kwargs == {"noreply": False}
+        for call in backend.client.delete.call_args_list
+    )
+
+
+@pytest.mark.asyncio
+async def test_memcached_delete_many_with_no_keys_does_no_io() -> None:
+    backend = stubbed_backend()
+
+    assert await backend.delete_many([]) == 0
+    assert backend.client.method_calls == []
+
+
+def _stub_for_single_hop(backend: MemcachedBackend) -> None:
+    """Script the stub so each operation takes its longest path."""
+    entry = encode_entry(_ENTRY)
+    # increment: INCR misses, ADD creates, INCR succeeds.
+    backend.client.incr.side_effect = [None, 1]
+    # get_and_delete: the first CAS loses to a writer, the second wins.
+    # The *_if_equals calls read a match and CAS once.
+    backend.client.gets.return_value = (entry, b"1")
+    backend.client.cas.side_effect = [False, True]
+
+
+@pytest.mark.parametrize(
+    ("operation", "args"),
+    [
+        ("increment", ("k",)),
+        ("get_and_delete", ("k",)),
+        ("delete_if_equals", ("k", _ENTRY)),
+        ("expire_if_equals", ("k", _ENTRY, 5)),
+        ("delete_many", (["a", "b", "c"],)),
+    ],
+)
+@pytest.mark.asyncio
+async def test_memcached_multi_step_operations_take_one_thread_hop(
+    operation: str, args: tuple[object, ...], monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Each round trip used to be its own asyncio.to_thread call (#176)."""
+    backend = stubbed_backend()
+    _stub_for_single_hop(backend)
+    hops = 0
+    real_to_thread = asyncio.to_thread
+
+    async def counting_to_thread(func, /, *a, **kw):
+        nonlocal hops
+        hops += 1
+        return await real_to_thread(func, *a, **kw)
+
+    monkeypatch.setattr(asyncio, "to_thread", counting_to_thread)
+
+    await getattr(backend, operation)(*args)
+
+    assert hops == 1
+    assert len(backend.client.method_calls) > 1