o bi@sddlZddlmZmZmZmZmZmZddlm Z ddl m Z m Z ddl mZeeZedZedZeeee e gee feee e e gee fedee ffZeGd d d ZeGd d d eZGd ddeZdeeefdefddZdS)N)AnyCallableIterableOptionalTypeVarUnion) TaskContext)BlockUserDefinedFunction) DeveloperAPITU.c@s eZdZdS)ComputeStrategyN)__name__ __module__ __qualname__rrN/home/ubuntu/.local/lib/python3.10/site-packages/ray/data/_internal/compute.pyrsrc@sBeZdZ d deefddZdedefddZde fd d Z dS) TaskPoolStrategyNsizecCs$|dur |dkr td|||_dS)zConstruct TaskPoolStrategy for a Dataset transform. Args: size: Specify the maximum size of the task pool. Nz`size` must be >= 1) ValueErrorr)selfrrrr__init__!s  zTaskPoolStrategy.__init__otherreturncCs(t|tr |j|jkp|dko|jduS)Ntasks) isinstancerrrrrrr__eq__/szTaskPoolStrategy.__eq__cCsd|jdS)NzTaskPoolStrategy(size=))rrrrr__repr__4szTaskPoolStrategy.__repr__N) rrrrintrrboolrstrr"rrrrrs rc @sfeZdZdZddddddeedeedeedeefdd Zd ed efd d Z d e fddZ dS)ActorPoolStrategyaSpecify the compute strategy for a Dataset transform. ActorPoolStrategy specifies that an autoscaling pool of actors should be used for a given Dataset transform. This is useful for stateful setup of callable classes. For a fixed-sized pool of size ``n``, specify ``compute=ActorPoolStrategy(size=n)``. To autoscale from ``m`` to ``n`` actors, specify ``ActorPoolStrategy(min_size=m, max_size=n)``. To increase opportunities for pipelining task dependency prefetching with computation and avoiding actor startup delays, set max_tasks_in_flight_per_actor to 2 or greater; to try to decrease the delay due to queueing of tasks on the worker actors, set max_tasks_in_flight_per_actor to 1. N)rmin_sizemax_sizemax_tasks_in_flight_per_actorrr(r)r*cCs|dur|dkr td||dus|durtd|}|}|dur*|dkr*td||dur>|dur4d}||kr>td|||durK|dkrKtd||pNd|_|pUtd|_||_d |_d |_dS) aConstruct ActorPoolStrategy for a Dataset transform. Args: size: Specify a fixed size actor pool of this size. It is an error to specify both `size` and `min_size` or `max_size`. min_size: The minimum size of the actor pool. max_size: The maximum size of the actor pool. max_tasks_in_flight_per_actor: The maximum number of tasks to concurrently send to a single actor worker. Increasing this will increase opportunities for pipelining task dependency prefetching with computation and avoiding actor startup delays, but will also increase queueing delay. Nrzsize must be >= 1z>min_size and max_size cannot be set at the same time as `size`zmin_size must be >= 1zmin_size must be <= max_sizez1max_tasks_in_flight_per_actor must be >= 1, got: infrg?)rr(floatr)r* num_workersready_to_total_workers_ratio)rrr(r)r*rrrrIs6     zActorPoolStrategy.__init__rrcCs.t|to|j|jko|j|jko|j|jkSr#)rr'r(r)r*rrrrr|s   zActorPoolStrategy.__eq__c Cs.d|jd|jd|jd|jd|jd S)NzActorPoolStrategy(min_size=z , max_size=z , max_tasks_in_flight_per_actor=z )num_workers=z, ready_to_total_workers_ratio=r )r(r)r*r-r.r!rrrr"s zActorPoolStrategy.__repr__) rrr__doc__rr$rrr%rr&r"rrrrr'8s" 3r' compute_specrcCsTt|ttfstd|d|r|dkrtS|dkrtSt|tr&|Std)NzXIn Ray 2.5, the compute spec must be either TaskPoolStrategy or ActorPoolStrategy, was: .ractorsz;compute must be one of [`tasks`, `actors`, ComputeStrategy])rrr'rr)r0rrr get_computes  r3)loggingtypingrrrrrr'ray.data._internal.execution.interfacesrray.data.blockr r ray.util.annotationsr getLoggerrloggerr r BlockTransformrrr'r&r3rrrrs(     V