o ;i> @sddlZddlZddlZddlZddlZddlmZddlmZddlmZm Z m Z m Z ddl m Z ddlZddlmZddlmZmZmZmZmZmZmZmZmZmZmZmZmZddl m!Z!dd l"m#Z#dd l$m%Z%m&Z&m'Z'm(Z(dd l)m*Z*m+Z+m,Z,dd l-m.Z.dd l/m0Z0ddl1m2Z2ddl3m4Z4ej5rddl6Ze*e j7gZ8dZ9dZ:GdddZ;ee;ZdZ?dZ@ej5rddlAZGdddZBGdddZCGdddeCZDGdd d eCZEd!d"d#e;d$d%d&eFeGeHffd'd(ZId!d"d#e;d$d%d)eJd*eJd+eJd,e e eHeHgdfd-d.fd/d0ZKd!d"d#e;d$d%d)eJd*eJd+eJd,e e eHeHgdfd&ejLedffd1d2ZMid3d4d3fd5d6d7ejLedfd)eJd*eJd+eJd&ejLedff d8d9ZNd:eGd*eJd+eJfd;d<ZOd&eJfd=d>ZPed?d@id3d4d3dAd5d6dBej ejQeejRefd)eJd*eJd+eJd&ejLedff dCdDZSedEd@id3d4d3dAdFej ejQejTeejRejTefd)eJd*eJd+eJd&ejRef dGdHZUid4dIdJeJd&dfdKdLZVedMd@id3d4d3dAdBejQed)eJd*eJd+eJd&ef dNdOZWidPdidRdSZXif 6 & QdjdTdUZYidPdidVdWZZidPdkdXdYZ[idPdkdZd[Z\id4dIdJeJfd\d]Z]ed^d@id3d4d3dAdFejQejTed)eJd*eJd+eJd&ef d_d`Z^Gdadbdbej_Z`Gdcddddej_ZaGdedfdfZbGdgdhdhZcdS)lN)FIRST_COMPLETED) dataclass)AnyCallableOptionalUnion)Status)current_input_id) AsyncOrSyncIterableTimestampPriorityQueueaclosing async_mapasync_map_ordered async_merge async_zipqueue_batch_iterator%run_coroutine_in_temporary_event_loopsync_or_async_itersynchronize_api synchronizer!warn_if_generator_is_not_consumed)BLOB_MAX_PARALLELISMdeprecation_warning)ATTEMPT_TIMEOUT_GRACE_PERIODOUTPUTS_TIMEOUT _create_input_process_result)RETRYABLE_GRPC_STATUS_CODESRetryRetryWarningMessage) DecodedJwt)logger) RetryManager)api_pb2c@s4eZdZdZddZejddZejddZdS) _SynchronizedQueuez mdmd:hiddencst|_dSN)asyncioQueueqselfr.F/home/ubuntu/.local/lib/python3.10/site-packages/modal/parallel_map.pyinit9sz_SynchronizedQueue.initcs|j|IdHdSr()r+putr-itemr.r.r/r1>sz_SynchronizedQueue.putcs|jIdHSr()r+getr,r.r.r/r4Bsz_SynchronizedQueue.getN) __name__ __module__ __qualname____doc__r0rno_io_translationr1r4r.r.r.r/r'5s r'c@seZdZUeed<dS) _OutputValuevalueN)r5r6r7r__annotations__r.r.r.r/r:Js r:1ic@s^eZdZdZdddedejdddeegd fd egd ff d d Z d dZ ddZ ddZ d S)InputPreprocessorzx Constructs FunctionPutInputsItem objects from the raw-input queue, and puts them in the processed-input queue. clientmodal.client._Clientraw_input_queueprocessed_input_queuefunctionmodal.functions._Functioncreated_callbackN done_callbackcCs.||_||_d|_||_||_||_||_dSNr)r@rDinputs_createdrBrCrFrG)r-r@rBrCrDrFrGr.r.r/__init__`  zInputPreprocessor.__init__cCs( |jIdH}|durdS|Vqr()rBr4)r- raw_inputr.r.r/ input_iterrszInputPreprocessor.input_itercsfdd}|S)NcsHj}jd7_j|\}}t||jj|jdIdHS)NidxrD)rIrFrr@stubrD) argskwargsrPargskwargsr,r.r/ create_inputzs  z.create_inputr.)r-rUr.r,r/create_input_factoryys  z&InputPreprocessor.create_input_factoryc Cstt||td4IdH}|2z3dHW}|j|IdHq6WdIdHn 1IdHs6wY|jdIdH|dVdS)N concurrency)r rrMrVrrCr1rG)r-streamerr3r.r.r/drain_input_generators ( z'InputPreprocessor.drain_input_generator) r5r6r7r8r'r)r*rintrJrMrVrZr.r.r.r/r?[s$    r?c@sXeZdZdZdddddejddd ed ed ed f d dZ ddZ e de fddZ dS) InputPumperz\ Reads inputs from a queue of FunctionPutInputsItems, and sends them to the server. N)map_items_managerr@rA input_queuerDrEfunction_call_idmax_batch_sizer]_MapItemsManagercCs.||_||_||_||_d|_||_||_dSrH)r@rDr]r^ inputs_sentr_r`)r-r@r^rDr_r`r]r.r.r/rJrKzInputPumper.__init__cCs|jjsJt|j|jd2zc3dHW}|jdur#|j|IdHtj|j j ||j d}t dt|d|jd|jjj||jdIdH}|jt|7_|jdura|j|jt dt|d|jdq6dVdS) Nr`) function_idinputsr_zPushing z6 inputs to server. Num queued inputs awaiting push is z. retryzSuccessfully pushed .)r@rQrr^r`r] add_itemsr$FunctionPutInputsRequestrD object_idr_r"debuglenqsizeFunctionPutInputs_function_inputs_retryrbhandle_put_inputs_responsere)r-itemsrequestrespr.r.r/ pump_inputss4      zInputPumper.pump_inputsreturncCs2td|jjddtjgd}tdttjg|dS)Nz#Warning: map progress for function zZ is limited. Common bottlenecks include slow iteration over results, or function backlogs.r%)messagewarning_intervalerrors_to_warn_for) max_retries max_delayadditional_status_codeswarning_message)r rD_function_namerRESOURCE_EXHAUSTEDrPUMP_INPUTS_MAX_RETRY_DELAY)r-retry_warning_messager.r.r/rpsz"InputPumper._function_inputs_retry)r5r6r7r8r)r*strr[rrJrupropertyrrpr.r.r.r/r\s&  r\csDeZdZdddejdedddeded d ffd d Zd dZZ S)SyncInputPumperr@rAr^ retry_queuerDrEfunction_call_jwtr_r]racs.tj||||t|d||_d|_||_dS)N)r^rDr_r`r]r)superrJMAP_INVOCATION_CHUNK_SIZErinputs_retriedr)r-r@r^rrDrr_r] __class__r.r/rJs  zSyncInputPumper.__init__cCst|j|jd2z?3dHW}|j|IdH}tj|j|d}|jj j ||j dIdH}|j |j tdt|d|jt|7_q 6dVdS)Nrc)rrerfzSuccessfully pushed retry for z to server.)rrr`r]prepare_items_for_retryr$FunctionRetryInputsRequestrr@rQFunctionRetryInputsrphandle_retry_response input_jwtsr"rlrmr)r-retriable_idxsrersrtr.r.r/ retry_inputss  zSyncInputPumper.retry_inputs) r5r6r7r)r*r rrJr __classcell__r.r.rr/rs" rcs<eZdZdddejdddeffdd Zfd d ZZS) AsyncInputPumperr@rAr^rDrEr_cstj||||tddS)N)r^rDr_r`)rrJSPAWN_MAP_INVOCATION_CHUNK_SIZE)r-r@r^rDr_rr.r/rJs zAsyncInputPumper.__init__cs\t2z3dHW}q6tj|jj|j|jd}|jj j |t dddIdHdVdS)N)rdr_ num_inputs)rzrf) rrur$FunctionFinishInputsRequestrDrkr_rbr@rQFunctionFinishInputsr)r-_rsrr.r/rus zAsyncInputPumper.pump_inputs) r5r6r7r)r*rrJrurr.r.rr/rsrrDrErBr@rArvc s|jsJtj|jtpdtjtjd}|j|IdH}|j}ddfdd}fdd}t }t ||||||d } t ||||d fd d fd d} dd} t | } t | | | IdH| | IdH|fS)N)rdparent_input_idfunction_call_typefunction_call_invocation_typeFrcs|dus |ks J|dSr(r.set_inputs_created)rIr.r/r3sz1_spawn_map_invocation..set_inputs_createdcsddS)NTr.r.)have_all_inputsr.r/set_have_all_inputs8sz2_spawn_map_invocation..set_have_all_inputsr@rBrCrDrFrG)r@r^rDr_c s$tdddjddS)Nzhave_all_inputs= inputs_created=z inputs_sent= )r"rlrbr.)r input_pumperrIr.r/ log_statsMsz(_spawn_map_invocation..log_statscs> z tdIdHWntjyYdSwq)NT r)sleepCancelledErrorr.rr.r/log_taskRsz'_spawn_map_invocation..log_taskcs|2z3dHW}q6dSr(r.)genrr.r.r/consume_generator\sz0_spawn_map_invocation..consume_generator)rQr$FunctionMapRequestrkr FUNCTION_CALL_TYPE_MAP#FUNCTION_CALL_INVOCATION_TYPE_ASYNC FunctionMapr_r)r*r?r create_taskgatherrZrucancel) rDrBr@rsresponser_rrr^input_preprocessorrrlog_debug_stats_taskr.)rrrIrr/_spawn_map_invocation#sP          r order_outputsreturn_exceptionswrap_returned_exceptionscount_update_callbackr,api_pb2.FunctionCallInvocationType.ValueTypecs2jsJtj|jtpdtj|d}j|IdH} | j| j} | j } | j | j p.t } d t  d ddddddddttt  t| || t| |fddfddd} t || d  d fd d  fd d fdddtjdtttfffdd fdd} fdd}t |}tt|   |4IdH}|2z3dHW} | dur| jVq6WdIdHn 1IdHs wY| |IdHdS)Nr)rdrrrrFrcs |dS)Nrr.)x update_stater.r/ z!_map_invocation..cs ddS)NTrr.r.rr.r/rrr)r@r^rrDr]rr_cs|dusJ|dus|ksJ|dus|ksJ|dur ||dur&||dur,|dur5rAkrCdSdSdS)NFset)rrset_outputs_completedrrrImap_done_eventoutputs_completedr.r/rs    z%_map_invocation..update_statec sjsJd}std ddddD}tjt|dt|d}t jj |t d tt d d }t }z6t j||gtd IdH\}}||vrb||}nshJW||dSW||n ||w|j}tt}|jD]d} d 7 | |IdH} | tjkr d 7 n1| tjkrd 7n'| tjkrd 7n| tjkrɈ d 7 n| tjkrӈd 7n | tjkr܈ d 7 | tjks| tjkr| j d d| Vqr dSdS)N0-0zRequesting outputs. Have z outputs, z inputs.cSsg|]}|qSr.r.).0 input_jwtr.r.r/ sz<_map_invocation..get_all_outputs..F)r_timeout last_entry_idclear_on_success requested_atrrzattempt_timeoutrf return_whenrN)r)rQis_setr"rl!get_input_jwts_waiting_for_outputr$FunctionGetOutputsRequestrtimer)rFunctionGetOutputsrrwaitrrresultrr[outputshandle_get_outputs_response _OutputTypeSUCCESSFUL_COMPLETIONFAILED_COMPLETIONNO_CONTEXT_DUPLICATESTALE_RETRY_DUPLICATEALREADY_COMPLETE_DUPLICATERETRYINGaddinput_id) rrrsget_response_task map_done_taskdonependingr now_secondsr3 output_type)already_complete_duplicatesr@completed_outputsfailed_completionsr_rIrr]no_context_duplicatesroutputs_receivedretried_outputsstale_retry_duplicatessuccessful_completionsrr.r/get_all_outputssz                      z(_map_invocation..get_all_outputsc sjsJzjt4IdH}|2z 3dHW}|Vq6WdIdHn*1IdHs.wYWtjdddtd}j|IdHIdHdSWtjdddtd}j|IdHIdHdStjdddtd}j|IdHIdHw)NrrT)r_rrrr)rQr r$rrrclose) output_itemsr3rs)r@r_rrr.r/get_all_outputs_and_clean_up sH *z5_map_invocation..get_all_outputs_and_clean_upr3rvc spzt|j|jjIdH}Wn"ty2}zr&r#tj|}n|}n|WYd}~nd}~ww|j|fSr() rr data_formatrQ Exceptionmodal exceptionUserCodeExceptionrPr3outpute)r@rrr.r/ fetch_outputs z%_map_invocation..fetch_outputc si}d}tttd4IdH8}|2z)3dHW\}}s&t|Vq|||< ||vr0n||}t|V|d7}q+q6WdIdHn 1IdHsQwYt|dks^JdS)NrrWTrNr rrr:poprmreceived_outputs output_idxrYrPrrrrr.r/ poll_outputs/,    (z%_map_invocation..poll_outputscsd fdd} |z tdIdHWntjy0|YdSwq)Nc!sxtd dddjdjdd dd d d d d d dtdS)Nz'Map stats: sync_client_retries_enabled=z have_all_inputs=rz input_sent=z inputs_retried=z outputs_received=z successful_completions= failed_completions= no_context_duplicates=z old_retry_duplicates= already_complete_duplicates= retried_outputs= input_queue_size=z retry_queue_size=z map_items_manager=)r"rlrbrrnrmr.rrrrr^rIr]rrrrrrsync_client_retries_enabledr.r/rLs>    z;_map_invocation..log_debug_stats..log_statsTrrrrr.r/log_debug_statsKs&z(_map_invocation..log_debug_stats)NNN)!rQr$rrkr rrr_r retry_policyrmax_inputs_outstandingMAX_INPUTS_OUTSTANDING_DEFAULTr)Eventr rr*rar?rFunctionGetOutputsItemtupler[rrr rrZrurr;r)rDrBr@rrrrrrsrrrrrr rrrYr.)rr@rrrrr_rrrrr^rIrr]rrrrrrrrrrrrr/_map_invocationjs       (M$&  * rcsLjsJdjIdH jsJdd td dddddddd ddttttjddddd }t |tj d td d  d'd t d t dt t dff fdd  fddfdd  fdd} fdd}  fdd}  fddfdddtjd tt tff fd!d" fd#d$}  f d%d&} t| } tt|| | | 4IdH}|2z3dHW}|dur|jVq6WdIdHn 1IdHswY| dS)(a<Input-plane implementation of a function map invocation. This is analogous to `_map_invocation`, but instead of the control-plane `FunctionMap` / `FunctionPutInputs` / `FunctionGetOutputs` RPCs it speaks the input-plane protocol consisting of `MapStartOrContinue`, `MapAwait`, and `MapCheckInputs`. zO_map_invocation_inputplane should only be used for input-plane backed functionsNzBClient must be hydrated with a stub for _map_invocation_inputplaneFrrr=g?)retriesinitial_delay_ms max_delay_msbackoff_coefficientT)rrrrris_input_plane_instance created_deltacompleted_deltarcsV|r|7|r |7|dur|durr'kr)dSdSdSr(r)r!r"rrr.r/update_counterss   z3_map_invocation_inputplane..update_counterscsBd}dd|\}}t||j|dIdH}tj|dS)NrN)r!rO)input)rrQr$MapStartOrContinueItem)rRrPrSrTput_item)r@rDrIr#r.r/rUs  z0_map_invocation_inputplane..create_inputcs& IdH}|durdS|Vqr()r4)rL)rBr.r/rMsz._map_invocation_inputplane..input_iterc stttd4IdH }|2z3dHW}t|IdHq6WdIdHn 1IdHs5wYdddVdS)NrWTr)r rrr1r)rYq_item)rUrMqueuer#r.r/rZs (  z9_map_invocation_inputplane..drain_input_generatorcsttd2zj3dHW}dd|DIdHtjjtp%dd}jIdH}j |t t j gt dd|dIdH}fddt|jD}|durq|j|jpdt|j|jq6dVdS) NrccSsg|] }tj|j|jdqS)r$ attempt_token)r$r%r$r*)rqir.r.r/rszC_map_invocation_inputplane..pump_inputs..r)rd map_tokenrrr)r|r{rzrgmetadatacs g|] \}}|jj|fqSr.r$rP)rrPr* request_itemsr.r/r s)rradd_items_inputplaner$MapStartOrContinueRequestrkr get_input_plane_metadata_input_plane_regionMapStartOrContinuerrrr enumerateattempt_tokenshandle_put_continue_responser,rrrset_retry_policyrupdate_items_retry_policy)batchrsr.rresponse_items_idx_tuple)r@rDinput_plane_stubr]r,map_token_receivedrr(r0r/rusF      /z/_map_invocation_inputplane..pump_inputscszwssdur IdHqttd}t}tj||gtdIdH\}}||vr4n?ddD}tjd|d} j IdH}j ||dIdHfddt j D}|IdHrdVWdStjyYdSw) NrNrcSsg|]\}}|qSr.r.)rrr*r.r.r/r0zI_map_invocation_inputplane..check_lost_inputs..r)rrr8)r.cs&g|]\}}|dj|fqS)r)lost)rresp_idxr check_inputsrr.r/r;s)rrr)rrr!get_input_idxs_waiting_for_outputr$MapCheckInputsRequestr4r5MapCheckInputsr7rAhandle_check_inputs_responser) sleep_taskrrrr8rsr.check_inputs_response)r@rDr>rrr]r,r?rCr/check_lost_inputs s>  z5_map_invocation_inputplane..check_lost_inputsc ssdur IdHqtjttd}jIdH}t j |t dtt d|d}t }z6tj||gt dIdH\}}||vrY||}ns_JW||dSW||n ||w|j|jD]d}|ttIdH}|tjkr d7 n9|tjkrd7n/|tjkr d7 n%|tjkr d7 n|tjkrĈ d7 n|tjkrΈd7ntd||tjks|tjkrdd|VqrdSdS) N)r,rrrrrr-rrNzUnknown output type: )r")rrr$MapAwaitRequestrrr4r5r)rMapAwaitrrrrrrrrr[rrrrrrrr) rsr.rrrrr output_itemr)rr@rrDr>rrr]r,r?rrrrr#r.r/rDst                     z3_map_invocation_inputplane..get_all_outputsc sz@t4IdH}|2z 3dHW}|Vq6WdIdHn1IdHs)wYWIdHdSWIdHdSIdHwr()r r)streamr3)rr(r.r/rs*z@_map_invocation_inputplane..get_all_outputs_and_clean_upr3rvc snzt|j|jIdH}Wn"ty1}zr%r"tj|}n|}n|WYd}~nd}~ww|j|fSr()rrrrrrrrPr)r@r>rrr.r/rs z0_map_invocation_inputplane..fetch_outputc si}d}tttd4IdH8}|2z)3dHW\}}s&t|Vq|||< ||vr0n||}t|V|d7}q+q6WdIdHn 1IdHsQwYt|dks^JdS)NrNrWTrrrr r.r/r r z0_map_invocation_inputplane..poll_outputsc s\ f dd} |z tdIdHWntjy,|YdSwq)NcsNtd dddddddd td dS) Nz"Map stats: successful_completions=r rz stale_retry_duplicates=rrz map_token=z max_inputs_outstanding=z map_items_manager_size=r)r"rlrmr. rrinput_queue_sizer]r,rrrrrr.r/rs&zF_map_invocation_inputplane..log_debug_stats..log_statsTrrrrPr.r/rs z3_map_invocation_inputplane..log_debug_stats)rrN)_input_plane_urlget_stubrQr)rr rr$FunctionRetryPolicyra"FUNCTION_CALL_INVOCATION_TYPE_SYNCr[rboolrrrrr rr;r)rDrBr@rrrrrrZrurKr rrmerged maybe_outputr.)rr@rrUrrrDrrrrMr>rQrIrrr]r,r?rrrrr(rBrrrrr#rr/_map_invocation_inputplanets    3($E&    * rYTFr-modal.functions.Functionasync_input_genc stjIdHfdd}tt|j||||4IdH}|2z 3dHW}|Vq*6WdIdHdS1IdHsFwYdS)aFCore implementation that supports `_map_async()`, `_starmap_async()` and `_for_each_async()`. Runs in an event loop on the main thread. Concurrently feeds new input to the input queue and yields available outputs to the caller. Note that since the iterator(s) can block, it's a bit opaque how often the event loop decides to get a new input vs how often it will emit a new output. We could make this explicit as an improvement or even let users decide what they prefer: throughput (prioritize queueing inputs) or latency (prioritize yielding results) Nc s~t4IdH }|2z3dHW}j|fIdHq 6WdIdHn 1IdHs/wYjdIdHdSr(r r1aiorYrSr[rTrBr.r/ feed_queues(z_map_helper..feed_queue)SynchronizedQueuer0r]r r_map) r-r[rTrrrr`map_output_streamrr.r_r/ _map_helpers  .rd func_namecCs,|r|rtdd|d|ddSdSdS)N)iz Function.a@ currently leaks an internal exception wrapping type (modal.exceptions.UserCodeException) when `return_exceptions=True` is set. In the future, this will change, and the underlying exception will be returned directly. To opt into the future behavior and silence this warning, add `wrap_returned_exceptions=False`: f.z=(..., return_exceptions=True, wrap_returned_exceptions=False)r)rerrr.r.r/_maybe_warn_about_exceptions srhcCs4zt}|jjjj}|dkWStyYdSw)zBCheck whether the calling function was called from a sync wrapper.asendF)inspect currentframef_backf_codeco_namer)framecaller_function_namer.r.r/_invoked_from_sync_wrappers   rqzFunction.map.aio) function_namerTrrrinput_iteratorscGsTts td||tdd|D}t||||||d2z 3dHW}|Vq6dS)Nzmap.aiocSg|]}t|qSr.rritr.r.r/r1r@z_map_async..rs)rqrhrrd)r-rTrrrrtr[rr.r.r/ _map_async$s ryzFunction.starmap.aioinput_iteratorcCsFts td||t|t|||||d2z 3dHW}|Vq6dS)Nz starmap.aiors)rqrhrdr)r-rzrTrrrrr.r.r/_starmap_async=s r{rTignore_exceptionsr}cs<tdd|D}t|||d|dd2z3dHW}q6dS)NcSrur.rvrwr.r.r/rWr@z#_for_each_async..Frs)rrd)r-rTr}rtr[rr.r.r/_for_each_asyncTsr~z Function.mapcGs0td||tt|g|R||||dddS)aParallel map over a set of inputs. Takes one iterator argument per argument in the function being mapped over. Example: ```python @app.function() def my_func(a): return a ** 2 @app.local_entrypoint() def main(): assert list(my_func.map([1, 2, 3, 4])) == [1, 4, 9, 16] ``` If applied to a `app.function`, `map()` returns one result per input and the output order is guaranteed to be the same as the input order. Set `order_outputs=False` to return results in the order that they are completed instead. `return_exceptions` can be used to treat exceptions as successful results: ```python @app.function() def my_func(a): if a == 2: raise Exception("ohno") return a ** 2 @app.local_entrypoint() def main(): # [0, 1, UserCodeException(Exception('ohno'))] print(list(my_func.map(range(3), return_exceptions=True))) ``` maprszgYou can't iter(Function.map()) from an async function. Use async for ... in Function.map.aio() instead.nested_async_message)rhr ry)r-rTrrrrtr.r.r/ _map_synccs - r)rTmodal.functions._FunctionCallcs&tdd|D}t|||IdHS)NcSrur.rvrwr.r.r/rr@z1_experimental_spawn_map_async..)r_spawn_map_helper)r-rTrtr[r.r.r/_experimental_spawn_map_asyncsrcsLtjIdHfdd}t|j|IdH\}}|S)Nc s~t4IdH }|2z3dHW}j|fIdHq 6WdIdHn 1IdHs/wYjdIdHdSr(r\r^r_r.r/r`s(z%_spawn_map_helper..feed_queue)rar0r]r)r _spawn_map)r-r[rTr`fcrr.r_r/rs  rcGtt|g|Rd|idS)aQmdmd:hidden Spawn parallel execution over a set of inputs, returning as soon as the inputs are created. Unlike `modal.Function.map`, this method does not block on completion of the remote execution but returns a `modal.FunctionCall` object that can be used to poll status and retrieve results later. Takes one iterator argument per argument in the function being mapped over. Example: ```python @app.function() def my_func(a, b): return a ** b @app.local_entrypoint() def main(): fc = my_func.spawn_map([1, 2], [3, 4]) ``` rT`You can't run Function.spawn_map() from an async function. Use Function.spawn_map.aio() instead.)rrr-rTrtr.r.r/_experimental_spawn_map_syncsrcsDfdd}tdd|D}t||dd2z3dHW}q6dS)zThis runs in an event loop on the main thread. It consumes inputs from the input iterators and creates async function calls for each. csjj|iS)a Returns co-routine that invokes a function with the given arguments. On RESOURCE_EXHAUSTED, it will retry indefinitely with exponential backoff up to 30 seconds. Every 10 retriable errors, log a warning that the function call is waiting to be created. )_spawn_map_innerr])rSrTr-r.r/_call_with_argssz)_spawn_map_async.._call_with_argscSrur.rvrwr.r.r/rr@z$_spawn_map_async..rWN)rr )r-rTrtr input_genrr.rr/_spawn_map_asyncs  rcGr)aSpawn parallel execution over a set of inputs, exiting as soon as the inputs are created (without waiting for the map to complete). Takes one iterator argument per argument in the function being mapped over. Example: ```python @app.function() def my_func(a): return a ** 2 @app.local_entrypoint() def main(): my_func.spawn_map([1, 2, 3, 4]) ``` Programmatic retrieval of results will be supported in a future update. rTr)rrrr.r.r/_spawn_map_syncsrcGs tt|g|R||dddS)aExecute function for all inputs, ignoring outputs. Waits for completion of the inputs. Convenient alias for `.map()` in cases where the function just needs to be called. as the caller doesn't have to consume the generator to process the inputs. r|zhYou can't run `Function.for_each()` from an async function. Use `await Function.for_each.aio()` instead.r)rr~)r-rTr}rtr.r.r/_for_each_syncsrzFunction.starmapc Cs(td||tt||||||dddS)aRLike `map`, but spreads arguments over multiple function arguments. Assumes every input is a sequence (e.g. a tuple). Example: ```python @app.function() def my_func(a, b): return a + b @app.local_entrypoint() def main(): assert list(my_func.starmap([(1, 2), (3, 4)])) == [3, 7] ``` starmaprszsYou can't `iter(Function.starmap())` from an async function. Use `async for ... in Function.starmap.aio()` instead.r)rhr r{)r-rzrTrrrr.r.r/ _starmap_syncs  rc@s eZdZdZdZdZdZdZdS) _MapItemStaterNN)r5r6r7SENDINGWAITING_FOR_OUTPUTWAITING_TO_RETRYrCOMPLETEr.r.r.r/r>s rc@s$eZdZdZdZdZdZdZdZdS)rrNrrrrrfN) r5r6r7rrrrrrr.r.r.r/rKsrc @seZdZUeed<ejed<eed<eed<e j ed<e j ed<e e ed<e j ed< d%dejdeded efd d Zd e fddZdejfddZdejdedddedef ddZdejfddZdejfddZde fdd Zd!edejfd"d#Zd$S)&_MapItemContextstater$ retry_managerrrrprevious_input_jwt _event_loopFr cCsFtj|_||_||_||_t|_|j |_ |j |_ ||_ dSr() rrrr$rrr)get_event_loopr create_futurerr_is_input_plane_instance)r-r$rrr r.r.r/rJas    z_MapItemContext.__init__r*cCsJ|js |j|n t|_|j||jtjkr#tj|_dSdSr() rr set_resultr)Futurerrrr)r-r*r.r.r/%handle_map_start_or_continue_responsevs     z5_MapItemContext.handle_map_start_or_continue_responser3cCs8|j|j|j|j|jtjkrtj|_dSdSr()rrrrrrrr2r.r.r/rqs   z*_MapItemContext.handle_put_inputs_responserrrrrvc s`|jtjkrtd|jd|jd|jtj S|j|j jkr;td|jd|jd|jd|j jtj S|j j tjjksK|tjksK|js]tj|_|j j tjjkrZtjStjS|j }|j j tjjkrld}|dusx|j j tjjkrtj|_tjStj|_|jr||jIdH}|||d|IdHtjS|||d|jIdHtjS) z Processes the output, and determines if it is complete or needs to be retried. Return True if input state was changed to COMPLETE, otherwise False. zRReceived output for input marked as complete. Must be duplicate, so ignoring. idx= input_id= retry_count=z9Received output with stale retry_count, so ignoring. idx=z expected_retry_count=rNr=)rrrr"rlrPr retry_countrrrrrstatusr$ GenericResultGENERIC_STATUS_SUCCESSrUrrr get_delay_msGENERIC_STATUS_INTERNAL_FAILUREGENERIC_STATUS_TERMINATEDrr!create_map_start_or_continue_itemr1r)r-r3rrrdelay_ms retry_itemr.r.r/rs\   z+_MapItemContext.handle_get_outputs_responsecs8tj|_|jIdH}|j|_tj||j|j j dS)N)rr$r) rrrrrrr$FunctionRetryInputsItemr$rrr-rr.r.r/prepare_item_for_retrys  z&_MapItemContext.prepare_item_for_retryrcCst||_dSr()r#rr-rr.r.r/r:sz _MapItemContext.set_retry_policycCs|j|tj|_dSr()rrrrrrr.r.r/rs  z%_MapItemContext.handle_retry_responserPcs(|jIdH}tjtj|j|d|dS)Nr/r))rr$r%FunctionPutInputsItemr$)r-rPr*r.r.r/rs z1_MapItemContext.create_map_start_or_continue_itemNF)r5r6r7rr<r$ FunctionInputr#rVr)rrrAbstractEventLooprJrFunctionPutInputsResponseItemrqrr[r rrrrrTr:rr%rr.r.r.r/rTsH         I rc @s^eZdZ d2dejdddedededef d d Zdejfd d Z d e ej fddZ d e ej fddZde ede ejfddZdejfddZde efddZde eeeffddZdefddZdedefdd Zd e eeeffd!d"Zd e ejfd#d$Zd%e efd&d'Zd(e eeeffd)d*Zd+ejd,edefd-d.Z d/d0Z!d1S)3raFrrrrrrr cCs4||_||_||_t||_i|_||_||_dSr() _retry_policyr _retry_queuer)BoundedSemaphore_inputs_outstanding _item_context_sync_client_retries_enabledr)r-rrrrrr r.r.r/rJs   z_MapItemsManager.__init__cCs ||_dSr()rrr.r.r/r: z!_MapItemsManager.set_retry_policyrrcs@|D]}|jIdHt|jt|j|jd|j|j<qdS)N)r$rr) racquirerr$r#rrrrPr-rrr3r.r.r/ri sz_MapItemsManager.add_itemscsf|D]-}|jdkrtj|j|jj_q|jIdHt |jjt |j |j |j d|j|jj<qdS)Nr)r$rrr )r*rrrr$rPrrrrr#rrrrr.r.r/r2s z%_MapItemsManager.add_items_inputplanerrvcsfdd|DIdHS)Ncs"g|] }j|IdHqSr()rr)rrPr,r.r/r$s z<_MapItemsManager.prepare_items_for_retry..r.)r-rr.r,r/r#sz(_MapItemsManager.prepare_items_for_retrycCs|jD]}||qdSr()rvaluesr:)r-rctxr.r.r/r;&s z*_MapItemsManager.update_items_retry_policycCdd|jDS)zV Returns a list of input_jwts for inputs that are waiting for output. cSs,g|]}|jtjkr|jr|jqSr.rrrrrr)rrr.r.r/r/s zF_MapItemsManager.get_input_jwts_waiting_for_output..)rrr,r.r.r/r*z2_MapItemsManager.get_input_jwts_waiting_for_outputcCr)zV Returns a list of input_idxs for inputs that are waiting for output. cSs4g|]\}}|jtjkr|jr||jfqSr.r)rrPrr.r.r/r:s  zF_MapItemsManager.get_input_idxs_waiting_for_output..)rrrr,r.r.r/rE5rz2_MapItemsManager.get_input_idxs_waiting_for_outputitem_idxcCs|j|=|jdSr()rrreleaser-rr.r.r/ _remove_item@sz_MapItemsManager._remove_itemcCs |j|Sr()rr4rr.r.r/get_item_contextDs z!_MapItemsManager.get_item_contextcCs2|D]\}}|j|d}|dur||qdSr()rr4r)r-rrindexr3rr.r.r/r9Gs  z-_MapItemsManager.handle_put_continue_responsecCs0|D]}|j|jd}|dur||qdSr()rr4rPrq)r-rrr3rr.r.r/rqSs  z+_MapItemsManager.handle_put_inputs_responsercCs>|D]}t|}|j|jdd}|dur||qdS)NrP)r!decode_without_verificationrr4payloadr)r-rr decoded_jwtrr.r.r/r\s  z&_MapItemsManager.handle_retry_responsercsh|D].\}}|j|d}|dur1|r1tj|_||IdH}|j}|j t |IdHqdSr() rr4rrrrrrrr1r)r-rrPrArrrr.r.r/rHfs  z-_MapItemsManager.handle_check_inputs_responser3rc s|j|jd}|dur"td|jd|jd|jdtjS| |||j |j IdH}|tj ks9|tj kr?||j|S)NzOReceived output that does not have entry in item_context map, so ignoring. idx=rrr)rr4rPr"rlrrrrrrrrrr)r-r3rrrr.r.r/rps&   z,_MapItemsManager.handle_get_outputs_responsecCs t|jSr()rmrr,r.r.r/__len__rz_MapItemsManager.__len__Nr)"r5r6r7r$rTr rVr[rJr:listrrir%r2rrr;rrrrErrrr9rrqrrHrrrrr.r.r.r/ras@         ra)rvr)r-rZrvr)rvN)dr)enumrjrtypingr dataclassesrrrrrgrpclibrmodal.exceptionr modal._runtime.execution_contextr modal._utils.async_utilsr r r r rrrrrrrrrmodal._utils.blob_utilsrmodal._utils.deprecationrmodal._utils.function_utilsrrrrmodal._utils.grpc_utilsrrr modal._utils.jwt_utilsr! modal.configr" modal.retriesr# modal_protor$ TYPE_CHECKING modal.clientr'PUMP_INPUTS_RETRYABLE_GRPC_STATUS_CODESPUMP_INPUTS_MAX_RETRIESrr'rar:rrrmodal.functionsr?r\rrrrr[rrVrAsyncGeneratorrYrdrhrqIterable AsyncIterablerySequencer{r~rrrrrrrrEnumrrrrar.r.r.r/sz    <       <C,  G    n   ,  $  =  *