&`iP$ddlZddlmZmZmZmZmZmZddlm Z ddl m Z m Z ddl mZmZejeZedZedZeeee e gee feee e e gee fedee ffZeGd d ZeGd d eZeGd deZdeeefdefdZdS)N)AnyCallableIterableOptionalTypeVarUnion) TaskContext)BlockUserDefinedFunction) DeveloperAPI PublicAPITU.ceZdZdS)ComputeStrategyN)__name__ __module__ __qualname__n/home/jaya/work/projects/VOICE-AGENT/VIET/agent-env/lib/python3.11/site-packages/ray/data/_internal/compute.pyrrsDrrcJeZdZdZ d deefdZdedefdZ de fdZ dS) TaskPoolStrategya;Specify the task-based compute strategy for a Dataset transform. TaskPoolStrategy executes dataset transformations using Ray tasks that are scheduled through a pool. Provide ``size`` to cap the number of concurrent tasks; leave it unset to allow Ray Data to scale the task count automatically. NsizecD||dkrtd|||_dS)zConstruct TaskPoolStrategy for a Dataset transform. Args: size: Specify the maximum size of the task pool. Nz`size` must be >= 1) ValueErrorr)selfrs r__init__zTaskPoolStrategy.__init__)s.  q2D99 9 rotherreturncjt|tr|j|jkp|dko|jduS)Ntasks) isinstancerrrr s r__eq__zTaskPoolStrategy.__eq__7s<5"233O UZ8O W  2d!2 rcd|jdS)NzTaskPoolStrategy(size=))rrs r__repr__zTaskPoolStrategy.__repr__<s4 4444rN) rrr__doc__rintrrboolr&strr*rrrrrs#  sm     C D    5#555555rrc eZdZdZddddddddeedeedeedeed eed ef d Zd ed efdZ d e fdZ dS)ActorPoolStrategyae Specify the actor-based 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``, use ``ActorPoolStrategy(size=n)``. To autoscale from ``m`` to ``n`` actors, use ``ActorPoolStrategy(min_size=m, max_size=n)``. To autoscale from ``m`` to ``n`` actors, with an initial size of ``initial``, use ``ActorPoolStrategy(min_size=m, max_size=n, initial_size=initial)``. 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. The `enable_true_multi_threading` argument primarily exists to prevent GPU OOM issues with multi-threaded actors. The life cycle of an actor task involves 3 main steps: 1. Batching Inputs 2. Running actor UDF 3. Batching Outputs The `enable_true_multi_threading` flag affects step 2. If set to `True`, then the UDF can be run concurrently. By default, it is set to `False`, so at most 1 actor UDF is running at a time per actor. The `max_concurrency` flag on `ray.remote` affects steps 1 and 3. Below is a matrix summary: - [`enable_true_multi_threading=False or True`, `max_concurrency=1`] = 1 actor task running per actor. So at most 1 of steps 1, 2, or 3 is running at any point in time. - [`enable_true_multi_threading=False`, `max_concurrency>1`] = multiple tasks running per actor (respecting GIL) but UDF runs 1 at a time. This is useful for doing CPU and GPU work, where you want to use a large batch size but want to hide the overhead of *batching* the inputs. In this case, CPU *batching* is done concurrently, while GPU *inference* is done 1 at a time. Concretely, steps 1 and 3 can have multiple threads, while step 2 is done serially. - [`enable_true_multi_threading=True`, `max_concurrency>1`] = multiple tasks running per actor. Unlike bullet #3 ^, the UDF runs concurrently (respecting GIL). No restrictions on steps 1, 2, or 3 NOTE: `enable_true_multi_threading` does not apply to async actors NF)rmin_sizemax_size initial_sizemax_tasks_in_flight_per_actorenable_true_multi_threadingrr2r3r4r5r6cd|1|dkrtd||||td|}|}|}||dkrtd|||d}||krtd||||dkrtd||pd|_|ptd|_|d||jkrtd |d |jd |jtdkr&||jkrtd |d |jd |p|j|_||_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. initial_size: The initial number of actors to start with. If not specified, defaults to min_size. Must be between min_size and max_size. 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. enable_true_multi_threading: If enable_true_multi_threading=True, no more than 1 actor task runs per actor. Otherwise, respects the `max_concurrency` argument. Nrzsize must be >= 1zMmin_size, max_size, and initial_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: infzinitial_size (z) must be >= min_size (r(z) must be <= max_size (rg?) rr2floatr3r4r5 num_workersready_to_total_workers_ratior6)rrr2r3r4r5r6s rrzActorPoolStrategy.__init__ms6  axx !4d;;;#x';|?W cHHL  HqLL4h?? ?  ("" !?8TTT ) 5-11C-  ! A  0E%LL   #dm++ Z\ZZ$-ZZZ}e ,, 1M1M Z\ZZ$-ZZZ)9DM-J*,/)+F(((rr r!ct|toO|j|jko?|j|jko/|j|jko|j|jko|j|jkSr+)r$r1r2r3r4r6r5r%s rr&zActorPoolStrategy.__eq__sw%!233 MU^ + 3 / 3!U%77 30E4UU 3223  rcxd|jd|jd|jd|jd|jd|jd|jdS) NzActorPoolStrategy(min_size=z , max_size=z, initial_size=z , max_tasks_in_flight_per_actor=z )num_workers=z, enable_true_multi_threading=z, ready_to_total_workers_ratio=r()r2r3r4r5r:r6r;r)s rr*zActorPoolStrategy.__repr__s Q$- Q Q  Q Q - Q Q.2-O Q Q +  Q Q ,0+K  Q Q -1,M  Q Q Q r) rrrr,rr-r.rrr&r/r*rrrr1r1@s))\#"&"&&*7;,1GGGGGGsmGG3- GG 3- GG sm GG(0}GG&*GGGGGGGGR C D      #       rr1 compute_specr!ct|ttfstd|d|r|dkrtS|dkrtSt|tr|Std)NzXIn Ray 2.5, the compute spec must be either TaskPoolStrategy or ActorPoolStrategy, was: .r#actorsz;compute must be one of [`tasks`, `actors`, ComputeStrategy])r$rr1rr)r>s r get_computerBs l%57H$I J J X K;G K K K   X\W44!!!  ! ! """ L/ 2 2XVWWWr)loggingtypingrrrrrr'ray.data._internal.execution.interfacesr ray.data.blockr r ray.util.annotationsr r getLoggerrloggerrrBlockTransformrrr1r/rBrrrrKsDDDDDDDDDDDDDDDD??????5555555588888888  8 $ $ GCLL GCLL  huo{ +Xe_ <= huo{,? @(5/ QR S(5/ !"$          5555555 5@ H H H H H H H  H V XeC$89 Xo X X X X X Xr