U /e@sddlmZddlZddlZddlmZmZddlmZm Z ddl m Z ddl m Z ddlmZmZddlmZdd lmZmZmZmZGd d d eeefZdS) ) annotationsN)Callable Collection)ExecutorThreadPoolExecutor)wraps)chain)AnyLiteral)Buffer)KTVTTlockedcseZdZUdZded<ded<ded<ded <eejd d d d ddd ddfddZddfdd Zdd ddddZ e d+ddddd d!Z d"d#dd$d%d&Z e d,d'dd(d)d*Z ZS)- AsyncBufferaExtension of :class:`~zict.Buffer` that allows offloading all reads and writes from/to slow to a separate worker thread. This requires ``fast`` to be fully thread-safe (e.g. a plain dict). ``slow.__setitem__`` and ``slow.__getitem__`` will be called from the offloaded thread, while all of its other methods (including, notably for the purpose of thread-safety consideration, ``__contains__`` and ``__delitem__``) will be called from the main thread. See Also -------- Buffer Parameters ---------- Same as in Buffer, plus: executor: concurrent.futures.Executor, optional An Executor instance to use for offloading. It must not pickle/unpickle. Defaults to an internal ThreadPoolExecutor. nthreads: int, optional Number of offloaded threads to run in parallel. Defaults to 1. Mutually exclusive with executor parameter. zExecutor | Noneexecutorz int | Nonenthreadszset[asyncio.Future]futureszdict[asyncio.Future, float]evictingN)rrr intNone)argsrrkwargsreturncs>tj||||_|rdn||_|dk|_t|_i|_dS)N)super__init__rrZ_internal_executorsetrr)selfrrrr __class__5/tmp/pip-unpacked-wheel-z3s6s24u/zict/async_buffer.pyr/s  zAsyncBuffer.__init__rcsJt|jD] }|q|jdk rF|jdk rF|jjddd|_dS)NT)wait)rclosercancelrrshutdown)rfuturerr!r"r%>s    zAsyncBuffer.closezCallable[..., T]zasyncio.Future[T])funcrrcGsj|jdkr$|jstt|jdd|_t}t}|j|j|j |f|}|j || |j j |S)Nzzict.AsyncBuffer offloader)Zthread_name_prefix)rrAssertionErrorrasyncioZget_running_loop contextvarsZ copy_contextZrun_in_executorrunraddadd_done_callbackremove)rr)rZloopcontextr(r!r!r"_offloadGs   zAsyncBuffer._offloadraisezCollection[KT]zLiteral[('raise', 'omit')]zasyncio.Future[dict[KT, VT]])keysmissingrcsdkrfddDn2dkr@D]}|kr(t|q(ntdzj}Wntk rrYnXt}|||Sddfdd }|S) ajFetch one or more key/value pairs. If not all keys are available in fast, offload to a worker thread moving keys from slow to fast, as well as possibly moving older keys from fast to slow. Parameters ---------- keys: collection of zero or more keys to get missing: raise or omit, optional raise (default) If any key is missing, raise KeyError. omit If a key is missing, return a dict with less keys than those requested. Notes ----- All keys may be present when you call ``async_get``, but ``__delitem__`` may be called on one of them before the actual data is fetched. ``__setitem__`` also internally calls ``__delitem__`` in a non-atomic way, so you may get ``KeyError`` when updating a value too. Zomitcsg|]}|kr|qSr!r!).0key)rr!r" tsz)AsyncBuffer.async_get..r3z%missing: expected raise or omit; got z dict[KT, VT]r#c sRi}D]D}jjrtz|||<Wqtk rJdkrFYqXq|S)Nr3)fastclosedr+ZCancelledErrorKeyError)dkr4r5rr!r" _async_gets z)AsyncBuffer.async_get.._async_get)r; ValueErrorr9Zget_all_or_nothingr+ZFutureZ set_resultr2)rr4r5r7r<fr?r!r>r" async_getXs   zAsyncBuffer.async_getr r )r7valuercCs||||dS)zImmediately set a key in fast. If this causes the total weight to exceed n, asynchronously start moving keys from fast to slow in a worker thread. N)Z set_noevictasync_evict_until_below_target)rr7rCr!r!r" __setitem__s zAsyncBuffer.__setitem__z float | None)nrcCsh|dkr|j}td|}tt|jjg|j}||kr>dS||j |}||j|<| |jj dS)zvIf the total weight exceeds n, asynchronously start moving keys from fast to slow in a worker thread. Ng) rFmaxminrr9Z total_weightrvaluesr2Zevict_until_below_targetr/ __delitem__)rrFZweightr(r!r!r"rDs  z*AsyncBuffer.async_evict_until_below_target)r3)N)__name__ __module__ __qualname____doc____annotations__rr rr%r2rrBrErD __classcell__r!r!rr"rs"   @r) __future__rr+r,collections.abcrrconcurrent.futuresrr functoolsr itertoolsrtypingr r Z zict.bufferr Z zict.commonr r rrrr!r!r!r"s