diff --git a/CLAUDE.md b/CLAUDE.md index a9ed035..1dfba38 100644 --- a/CLAUDE.md +++ b/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`. diff --git a/docs/BACKENDS.md b/docs/BACKENDS.md index 0e93e44..67dc45c 100644 --- a/docs/BACKENDS.md +++ b/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 diff --git a/fastapi_cachex/backends/base.py b/fastapi_cachex/backends/base.py index 3af3132..83c187c 100644 --- a/fastapi_cachex/backends/base.py +++ b/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: diff --git a/fastapi_cachex/backends/memcached.py b/fastapi_cachex/backends/memcached.py index 64b205e..f3dbf3c 100644 --- a/fastapi_cachex/backends/memcached.py +++ b/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. diff --git a/i18n/zh-TW/docs/BACKENDS.md b/i18n/zh-TW/docs/BACKENDS.md index 31a1c2f..3e6c72e 100644 --- a/i18n/zh-TW/docs/BACKENDS.md +++ b/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。設定多台伺服器時，失敗的那台會立即移出輪替，它的鍵會改由其餘伺服器處理，直到它恢復回應。 diff --git a/tests/backends/test_memcached.py b/tests/backends/test_memcached.py index c2c6f3c..01d2f5b 100644 --- a/tests/backends/test_memcached.py +++ b/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