[eddlmZddlZddlZddlmZmZddlmZm Z ddl m Z ddl m Z ddlmZmZddlmZdd lmZmZmZmZGd d eeefZdS) ) annotationsN)Callable Collection)ExecutorThreadPoolExecutor)wraps)chain)AnyLiteral)Buffer)KTVTTlockedceZdZUdZded<ded<ded<ded <eejd d d d(fdZd)fd Zd*dZ e d+d,dZ d-d$Z e d.d/d'Z xZS)0 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. Executor | Noneexecutorz int | Nonenthreadszset[asyncio.Future]futureszdict[asyncio.Future, float]evictingN)rrargsr intkwargsreturnNonectj|i|||_|rdn||_|du|_t |_i|_dSN)super__init__rr_internal_executorsetrr)selfrrrr __class__s 1lib/python3.11/site-packages/zict/async_buffer.pyr!zAsyncBuffer.__init__/s[ $)&)))  (6h "*d"2uu  ct|jD]}||j+|j&|jdd|_dSdSdS)NT)wait)r closercancelrrshutdown)r$futurer%s r&r*zAsyncBuffer.close>su  l  F MMOOOO = $)B M " " " - - - DMMM % $)B)Br'funcCallable[..., T]asyncio.Future[T]cN|j$|jsJt|jd|_tj}t j}|j|j|j|g|R}|j || |j j |S)Nzzict.AsyncBuffer offloader)thread_name_prefix) rrrasyncioget_running_loop contextvars copy_contextrun_in_executorrunraddadd_done_callbackremove)r$r.rloopcontextr-s r&_offloadzAsyncBuffer._offloadGs = = . 2NDM'))*,,%%dmW[$NNNN      !4555 r'raisekeysCollection[KT]missingLiteral['raise', 'omit']asyncio.Future[dict[KT, VT]]cxdkrfdDn1dkrD]}|vrt|ntd j}t j}|||S#t$rYnwxYwdfd }|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. omitcg|]}|v| SrH).0keyr$s r& z)AsyncBuffer.async_get..ts777C3$;;C;;;r'r?z%missing: expected raise or omit; got r dict[KT, VT]ci}D]D}jjrtj |||<.#t$r dkrYAwxYw|S)Nr?)fastclosedr3CancelledErrorKeyError)dkr@rBr$s r& _async_getz)AsyncBuffer.async_get.._async_getsA  9#3!02227AaDD'))*) Hs 4AA)rrL)rQ ValueErrorrNget_all_or_nothingr3Future set_resultr>)r$r@rBrJrRfrTs``` r& async_getzAsyncBuffer.async_getXs6 f  77774777DD    ( (d??"3--'# (NWNNOO O  ,,T22A/6n.>.>A LLOOOH     D         }}Z(((s B BBrJr valuercZ||||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) set_noevictasync_evict_until_below_target)r$rJr[s r& __setitem__zAsyncBuffer.__setitem__s2 e$$$ ++-----r'n float | NonecN||j}td|}tt|jjg|j}||krdS||j |}||j|<| |jj dS)zvIf the total weight exceeds n, asynchronously start moving keys from fast to slow in a worker thread. Ng) r`maxminr rN total_weightrvaluesr>evict_until_below_targetr: __delitem__)r$r`weightr-s r&r^z*AsyncBuffer.async_evict_until_below_targets 9A QKKUDI23T]5I5I5K5KLLMM Q;; FtrrZr_r^ __classcell__)r%s@r&rrs84    )))) U6?%)        !!!!!!    " HO>)>)>)>) V>)@.... <<<< V<<<<rxs""""""00000000;;;;;;;;))))))))))))a<a<a<a<a<&R.a<a<a<a<a