U /e=m@sUddlmZddlZddlZddlZddlmZddlmZddlm Z ddl m Z ddl m Z mZddlZddlZddlmZdd lmZdd lmZdd lmZmZdd lmZmZmZmZmZdd l m!Z!m"Z"ej#$dZ%ej#$dZ&iZ'de(d<dhddZ)ddZ*diddZ+ddZ,ddZ-ddZ.ddZ/d d!Z0d"d#Z1d$d%Z2iZ3d&d'Z4e4d(e)e*e4d)e+e,e4d*e0e1e4d+de2d,d-Z5djd/d0d1d2d3d4d5Z6dkd6d7Z7dld8d9Z8dmd:d;Z9Gdd?d?ZdDdEZ?dFdGZ@dHdIZAdJdKZBdLdMZCdNdOZDe ejEe<dPdQZFe%EeGdRdSZHe%EeIdTdUZJe&EeGdVdWZKe&EeIdXdYZLe%EedZd[ZMe&Eed\d]ZNe%EeOd^d_ZPe&EeOd`daZQdbdcZRGdddedeZSeSd(ZTe&EeUeTj7d(e%e&fdfdgZVdS)n) annotationsN)array)Enum)partial) ModuleType)AnyLiteral)normalize_token)typename)pickle) decompressmaybe_compress)frame_split_sizemerge_memoryviews msgpack_optspack_frames_prelude unpack_frames)ensure_memoryview has_keyworddask_serializedask_deserializezdict[str, ModuleType]_cached_allowed_modulescCstt|}ztt|}Wntk r:t|YnXt|drX|||d\}}n ||\}}||tt|dd}||fS)z/Serialize object using the class-based registrycontextrdask) sub-headertypetype-serialized serializer) r rrdispatch TypeErrorNotImplementedErrorrr dumps)xr type_namer"Z sub_headerframesheaderr'B/tmp/pip-unpacked-wheel-g426oqom/distributed/protocol/serialize.py dask_dumps#s    r)cCs&t|d}t|}||d|S)Nrr)r loadsrr)r&r%typr*r'r'r( dask_loads8s r,csRdggfdd}tj|||r0|ddnddd<dtd}|fS)Ncs$t|}||j dSN) memoryviewappendreadonly)fr% writeabler'r(buffer_callbackBs z%pickle_dumps..buffer_callbackzpickle-protocol)r4protocolrr )rr3)r r"gettuple)r#rr4r&r'r2r( pickle_dumps>s r8cCsX|d|dd}}|d}|s0t|d}ddt|tt|D}tj||dS)Nrr3r-cSs6g|].\}}||jkr.t|r$t|nt|n|qSr')r0r. bytearraybytes).0wmvr'r'r( [sz pickle_loads..)buffers)r6lenzipmaprr r*)r&r%r#r@r3r'r'r( pickle_loadsTs  rDcCsp|tkrt|S|d}|ddd}|rX|tjdkrXt|t|<t|St dt |ddS)Nascii.r9rz%distributed.scheduler.allowed-importsz Importing z is not allowed, please add it to the list of allowed modules the scheduler can import via the distributed.scheduler.allowed-imports configuration setting.) rencodedecodesplitrconfigr6 importlib import_module RuntimeErrorrepr)namerootr'r'r(import_allowed_modulecsrQcCs\d|kr0t|d}t||d}t||dSd|krDt|dSd|krXt|dS|S) , Custom packer/unpacker for msgpack __Enum__ __module____name__rO__Set__as-list__Serialized__data)rQgetattrset Serialized)objmodr+r'r'r(msgpack_decode_defaultxs   r_cCsXt|trdt|jdSt|tr.)rmr )r&r%msgr'r'r(serialization_error_loadssrucCs|||ot|dft|<dS)Nr)rfamilies)rOr"r*r'r'r(register_serialization_familysrwrr rgerrorcCst|tttfkr*t|r*ttt|St|tkrVt|rVttt| dSzt t|WdSt k r~YnXdS)Nr9TF) rrcr[r7rAcheck_dask_serializablenextiterdictitemsrrr r#r'r'r(rysrymessageobjectzLiteral['message' | 'raise']z bool | Nonez/tuple[dict[str, Any], list[bytes | memoryview]])r#on_erroriterate_collectionreturnc s@dkr dt|tr"|j|jfSt|tr@t|jddS|dkrt|tt t t fkrt|tkrdkrdkp d dk}|st |}t|t kr|rztt|Wntk rd}YnXd}t|tt t fkr|st|t kr$|r$|r$t|t r^g}|D]4\}}t|d\} } || d <|| | fq&n,t|tt t fsttfd d |D}g} g} g} |D]F\} } | | t| }| || | d pdgt| qd d |Dd| t|jd}tdd| Dr| |d <|| fSd}D]}t|\}}}z4|rR||dn||\}} ||d<|| fWStk rYq,Yn&tk rt}YqYnXq,dt|j}dkr |g}|r||dddd |D} ddi| fSdkr,t|t|ddnt dddS)a! Convert object to a header and list of bytestrings This takes in an arbitrary Python object and returns a msgpack serializable header and a list of bytes or memoryview objects. The serialization protocols to use are configurable: a list of names define the set of serializers to use, in order. These names are keys in the ``serializer_registry`` dict (e.g., 'pickle', 'msgpack'), which maps to the de/serialize functions. The name 'dask' is special, and will use the per-class serialization methods. ``None`` gives the default list ``['dask', 'pickle']``. Notes on the ``iterate_collection`` argument (only relevant when ``x`` is a collection): - ``iterate_collection=True``: Serialize collection elements separately. - ``iterate_collection=False``: Serialize collection elements together. - ``iterate_collection=None`` (default): Infer the best setting. Examples -------- >>> serialize(1) ({}, [b'\x80\x04\x95\x03\x00\x00\x00\x00\x00\x00\x00K\x01.']) >>> serialize(b'123') # some special types get custom treatment ({'type': 'builtins.bytes'}, [b'123']) >>> deserialize(*serialize(1)) 1 Returns ------- header: dictionary containing any msgpack-serializable metadata frames: list of bytes or memoryviews, commonly of length one See Also -------- deserialize : Convert header and frames back to object to_serialize : Mark that data in a message should be serialized register_serialization : Register custom serialization functions N)rr T) serializersrrrrgr Frrrkeycsg|]}t|dqS)r)rbr<r]rrrr'r(r?<szserialize.. compressioncSsg|] }|dqS)rr'rr'r'r(r?Ms) sub-headers is-collection frame-lengthsrcss|]}|dk VqdSr-r')r<rr'r'r( Rszserialize..rrz#Could not serialize object of type ricSsg|] }|qSr')rGrsr'r'r(r?jsrxraisei'z on_error=z; expected 'message' or 'raise')!r`r\r&r%rarbrYrrcr[r7r|indexryrgr"keysrhr}r/AssertionErrorextendrAr6rUanyrvr! traceback format_excr str ValueError)r#rrrrZ dict_safeZheaders_frameskv_headerZ_framesr%lengths compressionslengthheaderstbrOr"_ wants_contextr&rtZ txt_framesr'rr(rbs0           $     rbc Csd|kr|d}|d}ttttd|d}d}|tkri}t||D]8\}} |d} t||||| |d|| <|| 7}qH|Sg} t||D]0\}} | t||||| |d|| 7}q|| S|d } |d k r| |krt d | t t|ft | \} }}|||S) a Convert serialized header and list of bytestrings back to a Python object Parameters ---------- header : dict frames : list of bytes deserializers : dict[str, tuple[Callable, Callable, bool]] | None An optional dict mapping a name to a (de)serializer. See `dask_serialize` and `dask_deserialize` for more. See Also -------- serialize rrr)r7rcr[r|rrr deserializersrNzAData serialized with %s but only able to deserialize data with %s) r7rcr[r|rBpop deserializer/r6r rrv)r&r%rrrclsstartdr_lengthrlstrOr"r*rr'r'r(rssL      rcCst||||\}}g}g}g} g} t||dp>> msg = {'op': 'update', 'data': to_serialize(123)} >>> msg # doctest: +SKIP {'op': 'update', 'data': } See also -------- distributed.protocol.dumps cCs ||_dSr-rYselfrYr'r'r(__init__szSerialize.__init__cCsd|jdS)Nz rrr'r'r(__repr__szSerialize.__repr__cCst|to|j|jkSr-)r`rarYrotherr'r'r(__eq__szSerialize.__eq__cCs ||k Sr-r'rr'r'r(__ne__ szSerialize.__ne__cCs t|jSr-hashrYrr'r'r(__hash__szSerialize.__hash__N rUrT __qualname____doc__rrrrrr'r'r'r(ras  rac@s(eZdZdZddZddZddZdS) r\aAn object that is already serialized into header and frames Normal serialization operations pass these objects through. This is typically used within the scheduler which accepts messages that contain data without actually unpacking that data. cCs||_||_dSr-rnrr&r%r'r'r(rszSerialized.__init__cCs"t|to |j|jko |j|jkSr-)r`r\r&r%rr'r'r(r!s    zSerialized.__eq__cCs ||k Sr-r'rr'r'r(r(szSerialized.__ne__NrUrTrrrrrr'r'r'r(r\sr\c@s8eZdZdZddZddZddZdd Zd d Zd S) ToPickleaQMark an object that should be pickled Both the scheduler and workers with automatically unpickle this object on arrival. Notice, this requires that the scheduler is allowed to use pickle. If the configuration option "distributed.scheduler.pickle" is set to False, the scheduler will raise an exception instead. cCs ||_dSr-rrr'r'r(r7szToPickle.__init__cCsdt|jS)Nz)rrYrr'r'r(r:szToPickle.__repr__cCst|t|o|j|jkSr-)r`rrYrr'r'r(r=szToPickle.__eq__cCs ||k Sr-r'rr'r'r(r@szToPickle.__ne__cCs t|jSr-rrr'r'r(rCszToPickle.__hash__Nrr'r'r'r(r,s  rc@s(eZdZdZddZddZddZdS) PickledzyAn object that is already pickled into header and frames Normal pickled objects are unpickled by the scheduler. cCs||_||_dSr-rnrr'r'r(rMszPickled.__init__cCs&t|t|o$|j|jko$|j|jkSr-)r`rr&r%rr'r'r(rQs   zPickled.__eq__cCs ||k Sr-r'rr'r'r(rXszPickled.__ne__Nrr'r'r'r(rGsrcsfdd|S)z Replace all Serialize and Serialized values nested in *x* with the original values. Returns a copy of *x*. >>> msg = {'op': 'update', 'data': to_serialize(123)} >>> nested_deserialize(msg) {'op': 'update', 'data': 123} cst|tkr||}|D]\\}}t|}|tks<|tkrJ|||<q|tkr^|j||<q|tkrt|j |j ||<qnzt|tkrt|}t |D]\\}}t|}|tks|tkrƈ|||<q|tkr|j||<q|tkrt|j |j ||<q|Sr-) rr|copyr}rcrarYr\rr&r% enumerate)r#rrr+ replace_innerr'r(rfs*    z)nested_deserialize..replace_innerr'r~r'rr(nested_deserialize\s rcKslt|f|\}}|r(ttt|\}}ng}||d<t||d<tj|dd}|f|}|dt||S)NrcountTrer) rrBrCr rArgr"insertr)r#kwargsr&r%rZframes2r'r'r(serialize_bytelists  rcKst|f|}d|S)Nrk)rrm)r#rLr'r'r(serialize_bytess rcCsLt|}|d|dd}}|r4tj|ddd}ni}t||}t||S)Nrr9F)rawrl)rrgr*r r)br%r&r'r'r(deserialize_bytess rcCs2t|trtdt||t||dS)aRegister a new class for dask-custom serialization Parameters ---------- cls : type serialize : callable(cls) -> Tuple[Dict, List[bytes]] deserialize : callable(header: Dict, frames: List[bytes]) -> cls Examples -------- >>> class Human: ... def __init__(self, name): ... self.name = name >>> def serialize(human): ... header = {} ... frames = [human.name.encode()] ... return header, frames >>> def deserialize(header, frames): ... return Human(frames[0].decode()) >>> register_serialization(Human, serialize, deserialize) >>> serialize(Human('Alice')) ({}, [b'Alice']) See Also -------- serialize deserialize z^Strings are no longer accepted for type registration. Use dask_serialize.register_lazy insteadN)r`rr rregisterr)rrbrr'r'r(register_serializations rcCs tddS)z[Register a registration function to be called if *toplevel* module is ever loaded. z9Serialization registration has changed. See documentationN)rh)toplevelfuncr'r'r(register_serialization_lazysrcCs|jg|jSr-rn)or'r'r(normalize_SerializedsrcCsi}|g}||fSr-r'r]r&r%r'r'r(_serialize_bytessrcCsi}|g}||fSr-r'rr'r'r(_serialize_bytearraysrcCs0t|dkr"t|dtr"|dSd|SdS)Nr9rrk)rAr`r;rmrnr'r'r(_deserialize_bytessrcCs2t|dkr"t|dtr"|dSt|SdS)Nr9r)rAr`r:rmrnr'r'r(_deserialize_bytearraysrcCs|jdd}t|g}||fS)Nr-)typecoder3)rr.rr'r'r(_serialize_arrays  rcCsRt|d}t|}|dkr0|t|dn|dkrN|dtt||S)Nrr9rrk)rrAZ frombytesrrmrC)r&r%aZnframesr'r'r(_deserialize_arrays rcCsD|jdkrtd|s(|jdkr(td|j|jd}|g}||fS)NOz7Cannot serialize `memoryview` containing Python objectsr9z+Cannot serialize empty non-1-D `memoryview`)formatshape)rrndimrrr'r'r(_serialize_memoryview s rcCsft|dkrt|d}ntd|}|rB||d|d}n ||d}|j|dksbt|S)Nr9rrkrr)rArr.rmcastrr)r&r%outr'r'r(_deserialize_memoryviews rcCst|}|dkp|tkp|tkp|tkp|tkp|tkpt|trjtt t | rjtdd| Dpt|t tfott t |S)Ncss|]}t|tkVqdSr-)rr)r<r#r'r'r(r9sz+_is_msgpack_serializable..)rrboolr;intfloatr`r|allrC_is_msgpack_serializablevaluesrrcr7)rr+r'r'r(r.s(   rc@s$eZdZddZddZddZdS)ObjectDictSerializercCs ||_dSr-)r)rrr'r'r(r@szObjectDictSerializer.__init__c Cs|jtt|iid}g}t|tr.|}n|j}|D]\}}t|rZ||d|<q>> import sklearn.base >>> from distributed.protocol import register_generic >>> register_generic(sklearn.base.BaseEstimator) See Also -------- dask_serialize dask_deserialize N)rrrbr)rZserializer_nameZserialize_funcZdeserialize_funcZobject_with_dict_serializerr'r'r(register_generic|s%r)N)N)NrNN)N)NrNN)N)W __future__rrrrKrrenumr functoolsrtypesrtypingrrrgrZ dask.baser Z dask.utilsr Zdistributed.protocolr Z distributed.protocol.compressionr r Zdistributed.protocol.utilsrrrrrZdistributed.utilsrrutilsZDispatchrrr__annotations__r)r,r8rDrQr_rdrjrorurvrwryrbrrrraZ to_serializer\rrrrrrrrrrr;rr:rrrrrr.rrrrZ dask_object_with_dict_serializerr|rr'r'r'r(s                  " ; , &)          8