o 3/i Y@s~ddlmZddlmZddlmZddlmZmZm Z ddl m Z m Z m Z ddlmZmZmZddlmZmZddlmZmZdd lmZdd lmZdd lmZdd lmZdd lmZddlm Z!ddlm"Z#ddZ$dededefddZ%ddZ&GdddeZ'dede(fddZ)Gdd d Z*ed!d"Gd#d$d$eZe$eee_+ed!d"Gd%d&d&e*eZed!d"Gd'd(d(e*eZed!d"Gd)d*d*e*eZe$eee_+ed!d"Gd+d,d,e*eZe$eee_+ed!d"Gd-d.d.e*e!Z e$e!e e _+ed!d"Gd/d0d0e*e#Z"e$e#e"e"_+ed!d"Gd1d2d2e*eZ,e$ee,e,_+Gd3d4d4eZ-ed!d"Gd5d6d6eZ.dS)7)Optional) dataclassN)BooleanInt32 const_expr) if_generateand_ dsl_user_op) MbarrierArrayCooperativeGroup PipelineOp) PipelineStatePipelineUserType)Agent agent_sync) NamedBarrier) PipelineAsync)PipelineCpAsync)PipelineTmaAsync)PipelineTmaUmma)PipelineUmmaAsync)PipelineAsyncUmmacstfdd}|S)zPCreate a static factory that constructs parent_cls then re-classes to child_cls.cs"j|i|}t|d|S)N __class__)createobject __setattr__)argskwargsobj child_cls parent_cls[/lsinfo/ai/hellotax_ai/llm_service/venv_vllm/lib/python3.10/site-packages/quack/pipeline.pyrsz _override_create..create) staticmethod)r!r rr"rr#_override_createsr%indexphasereturncCstdtd||dS)zPConstruct a PipelineState from index and phase (count/stages unused by callers).r)stagescountr&r')r r)r&r'r"r"r# _make_state%sr+cCsnt|r-t|r tjtj|||||dWddS1s&wYdS|||||ddS)zGOptionally wrap a parent pipeline method call in sync_warp + elect_one.locipN)rcutearch sync_warp elect_one) parent_methodselfstater2syncwarpr-r.r"r"r#_call_with_elect_one*s  "r7c@s.eZdZeddddefddZddZdS)PipelineStateWAdvanceNr,num_iterationscCsH|jt|7_|jt|}||j}|j|N_||j|_dSN)_countr_indexr)_phase)r4r9r-r. new_index num_crossingsr"r"r# advance_iters9s  z#PipelineStateWAdvance.advance_iterscCs(t|jt|dt|dt|dS)Nr)r8r)rr4valuesr"r"r#__new_from_mlir_values__Cs"z.PipelineStateWAdvance.__new_from_mlir_values__)__name__ __module__ __qualname__r rr@rEr"r"r"r#r88s r8typer)cCsP|tjurt|tdtdtdS|tjur$t|tdtdtdSJd)zz Creates a pipeline state. Producers are assumed to start with an empty buffer and have a flipped phase bit of 1. rrAFzBError: invalid PipelineUserType specified for make_pipeline_state.)rProducerr8rConsumer)rIr)r"r"r#make_pipeline_stateIs  rLc @seZdZdZe dddddededeefddZeddddefd d Z e dddddeded eefd d Z eddddefddZ dS)_PipelineIndexPhaseMixinz_Mixin providing _w_index_phase / _w_index methods that delegate to PipelineState-based parents.Nr,r&r'try_acquire_tokencC t||}|j||||ddSNr,)r+producer_acquire)r4r&r'rNr-r.r5r"r"r#producer_acquire_w_index_phase[ z7_PipelineIndexPhaseMixin.producer_acquire_w_index_phasecC"t|td}|j|||ddSNrr,)r+rproducer_commitr4r&r-r.r5r"r"r#producer_commit_w_indexhz0_PipelineIndexPhaseMixin.producer_commit_w_indextry_wait_tokencCrOrP)r+ consumer_wait)r4r&r'rZr-r.r5r"r"r#consumer_wait_w_index_phasemrSz4_PipelineIndexPhaseMixin.consumer_wait_w_index_phasecCrTrU)r+rconsumer_releaserWr"r"r#consumer_release_w_indexzrYz1_PipelineIndexPhaseMixin.consumer_release_w_indexr:) rFrGrH__doc__r rrrrRrXr\r^r"r"r"r#rMXs<    rMT)frozenc@sReZdZeedZeddddeddfddZeddddeddfddZ dS) rNr,r&r(cCs tjj|j||j||ddS)z The aligned flavor of arrive is used when all threads in the CTA will execute the same instruction. See PTX documentation.  barrier_idnumber_of_threadsr-r.N)r/r0barrier_arriverb num_threadsr4r&r-r.r"r"r#arrive_w_indexs  zNamedBarrier.arrive_w_indexcCs tjj|j||j||ddS)Nra)r/r0barrierrbrerfr"r"r#arrive_and_wait_w_indexs  z$NamedBarrier.arrive_and_wait_w_index) rFrGrHr%NamedBarrierOgrr rrgrir"r"r"r#rs   rc @seZdZUdZdZeed<dZeed<dZeed<dZ eed<e dddddd ed ed ed efd dZ e dddde fddZe dddde fddZdS)ra PipelineAsync with optional elect_one for producer_commit and consumer_release. When elect_one_*=True (set at create time), only one elected thread per warp signals the barrier arrive. This is useful when the mask count is set to 1 per warp. Args (to create): elect_one_commit: If True, only elected thread signals producer_commit. syncwarp_before_commit: If True (default), issue syncwarp before elect_one. elect_one_release: If True, only elected thread signals consumer_release. syncwarp_before_release: If True (default), issue syncwarp before elect_one. Set syncwarp to False when threads are already converged (e.g. after wgmma wait_group). F_elect_one_commitT_syncwarp_before_commit_elect_one_release_syncwarp_before_release)elect_one_commitsyncwarp_before_commitelect_one_releasesyncwarp_before_releaserorprqrrcOsZtj|i|}t|dtt|d|t|d|t|d|t|d||S)Nrrkrlrmrn)PipelineAsyncOgrrrr)rorprqrrrrrr"r"r#rs zPipelineAsync.createNr,r5cCttj|||j|j||dSr:)r7rsrVrkrlr4r5r-r.r"r"r#rVzPipelineAsync.producer_commitcCrtr:)r7rsr]rmrnrur"r"r#r]rvzPipelineAsync.consumer_release)rFrGrHr_rkbool__annotations__rlrmrnr$rr r rVr]r"r"r"r#rs0       rc@s^eZdZUdZeed<dZeed<eddddedefdd Ze d d d d e fd dZ d S)rFrmTrn)rqrrrqrrcOs>tj|i|}t|dtt|d|t|d||S)Nrrmrn)PipelineCpAsyncOgrrrr)rqrrrrrr"r"r#rs zPipelineCpAsync.createNr,r5cCrtr:)r7ryr]rmrnrur"r"r#r]rvz PipelineCpAsync.consumer_release) rFrGrHrmrwrxrnr$rr r r]r"r"r"r#rs    rc @<eZdZdZe  d ddddedeedefdd Z dS) r>Override producer_acquire to take in extra_tx_count parameter.Nrr,r5rNextra_tx_countcsxt|dup|dkfdddt|dkr)jjjjddSjj|}jjj|ddS) TMA producer commit conditionally waits on buffer empty and sets the transaction barrier for leader threadblocks. NrcjjjjdSrPsync_object_emptywaitr&r'r"r.r-r4r5r"r# z3PipelineTmaAsync.producer_acquire..r,)rrsync_object_fullarriver& producer_masktx_countarrive_and_expect_tx)r4r5rNr|r-r.rr"rr#rQs   z!PipelineTmaAsync.producer_acquireNr rFrGrHr_r r rrintrQr"r"r"r#r rc @rz) rr{Nrr,r5rNr|cst|dup|dkfdddt|dkr-tjfddddSjj|tjfddddS)r}Nrcr~rPrr"rr"r#rDrz2PipelineTmaUmma.producer_acquire..r,cjjjjdSrPrrr&rr"rr"r#rKs csjjjdSrP)rrr&r"r.r-r4r5rr"r#rUs )rr is_leader_ctarr)r4r5rNr|r-r.r"rr#rQ5s(    z PipelineTmaUmma.producer_acquirerrr"r"r"r#r1rrc@ eZdZdS)rNrFrGrHr"r"r"r#rcrc@r)rNrr"r"r"r#rnrrc @sZeZdZdZe  d ddddedeedeefdd Zeddddefd d Z dS) PipelineTmaCpAsyncz PipelineTmaCpAsync is used for CpAsync + TMA producers and AsyncThread consumers. Compared to PipelineTmaAsync, producer_acquire gates the full-barrier arrive on is_tma_warp. NTr,r5rN is_tma_warpcsLt|dup|dkfdddt|fddddS)Nrcr~rPrr"rr"r#rrz5PipelineTmaCpAsync.producer_acquire..r,crrPrr"rr"r#rr)rr4r5rNrr-r.r"rr#rQs  z#PipelineTmaCpAsync.producer_acquirecC"tjj|j|||d||ddS)z9We need the mbarrier to track the completion of cp.async.r,Nr/r0cp_async_mbarrier_arrive_noincproducer_get_barrierrur"r"r#producer_cpasync_commits z*PipelineTmaCpAsync.producer_cpasync_commitNT) rFrGrHr_r r rrrQrr"r"r"r#rys" rc@s^eZdZe  dddddejdedeee fdede e d df d d Z d d Z ddZdS)MbarrierArrayWDropCountrNr,barrier_storage num_stagesagentr drop_countr(cCs||_||_||_|\|_|_|jj|_||_|jdkr td|jdkr)td|jt j ur8|jdkr8tdt |durD|j||_|j|_ |j ||ddS)Nrz3Error: Mbarrier stage count must be greater than 0.z4Error: Mbarrier arrive count must be greater than 0.z=Error: Mbarrier tx count must not be less than 0 for TMA ops.r,)rrrop_typecgsize arrive_countr ValueErrorr TmaLoadr mbarrier_base mbarrier_init)r4rrrrrr-r.r"r"r#__init__s        z MbarrierArrayWDropCount.__init__cCs |j|jgSr:)rr)r4r"r"r#__extract_mlir_values__s z/MbarrierArrayWDropCount.__extract_mlir_values__cCs$t|d|j|j|jf|j|dS)NrrA)rrrrrrCr"r"r#rEsz0MbarrierArrayWDropCount.__new_from_mlir_values__)rN)rFrGrHr r/Pointerrtupler r rrrrrEr"r"r"r#rs,  " rc@seZdZdZeedddddddddededed ed ej d e ej d e eefd e de efddZe  dddddede ede efddZeddddefddZdS)PipelineTmaCpAsyncUmmazr PipelineTmaCpAsync is used for CpAsync + TMA producers and UMMA consumers (e.g. Blackwell mainloops) N)rArAF)rcta_layout_vmnk mcast_mode_mn defer_syncproducer_drop_countr-r.rproducer_groupconsumer_grouprrrrrrc  CsTt|tjstdt|tj} tj} | |f} | |f}t|j dd|| ||| | d}t j |j dd|||| | d}|dusJtj || | ddkrOd}d}nt j ||| | d}t j|| | d}|dusptj |d g| | d dkrvtjjjjntjjjj}|}|stj|dustj || | ddkrttjnttjdd t|||||||S) aCreates and initializes a new PipelineTmaUmma instance. :param num_stages: Number of buffer stages for this pipeline :type num_stages: int :param producer_group: CooperativeGroup for the producer agent :type producer_group: CooperativeGroup :param consumer_group: CooperativeGroup for the consumer agent :type consumer_group: CooperativeGroup :param tx_count: Number of bytes expected to be written to the transaction barrier for one stage :type tx_count: int :param barrier_storage: Pointer to the shared memory address for this pipeline's mbarriers :type barrier_storage: cute.Pointer, optional :param cta_layout_vmnk: Layout of the cluster shape :type cta_layout_vmnk: cute.Layout, optional :param mcast_mode_mn: Tuple specifying multicast modes for m and n dimensions (each 0 or 1) :type mcast_mode_mn: tuple[int, int], optional :raises ValueError: If barrier_storage is not a cute.Pointer instance :return: A new PipelineTmaUmma instance configured with the provided parameters :rtype: PipelineTmaUmma z7Expected barrier_storage to be a cute.Pointer, but got ) min_align)rr-r.r,NrATr)moder-r.) is_relaxed) isinstancer/r TypeErrorrIr r TCGen05MmaralignPipelineTmaUmmaOg_make_sync_objectr_compute_mcast_arrival_mask_compute_is_leader_ctanvgputcgen05CtaGroupONETWOr0mbarrier_init_fencerr ThreadBlockThreadBlockClusterr)rrrrrrrrrr-r. producer_type consumer_typeproducerconsumerrrrr cta_group consumer_maskr"r"r#rsh $       zPipelineTmaCpAsyncUmma.createTr,r5rNrcsTt|dup|dkfdddttj|fddddS)z TMA producer commit conditionally waits on buffer empty and sets the transaction barrier for leader threadblocks. Nrcr~rPrr"rr"r#rTrz9PipelineTmaCpAsyncUmma.producer_acquire..r,crrPrr"rr"r#r\r)rrrrr"rr#rQDs  z'PipelineTmaCpAsyncUmma.producer_acquirecCr)zJ We need the mbarrier to track the completion of cp.async r,Nrrur"r"r#ras z.PipelineTmaCpAsyncUmma.producer_cpasync_commitr)rFrGrHr_r r$rr r/rrLayoutrrwrrr rrQrr"r"r"r#rsZ   d r)/typingr dataclassesr cutlass.cuter/cutlassrrrcutlass.cutlass_dslrrr cutlass.pipeliner r r r rrrrrjrrsrryrPipelineTmaAsyncOgrrrPipelineUmmaAsyncOgrPipelineAsyncUmmaOgr%r+r7r8rrLrMrrrrr"r"r"r#s\           + D#  +   '0