o i@sddlmZddlmZddlZddlmZddlmZm Z m Z ddl m Z m Z mZmZddl mZmZGdddZd ed efd d Zed dGddde ZdS))Optional) dataclassN)BooleanInt32 if_generate) PipelineAsync PipelineStateCooperativeGrouppipeline_init_wait)PipelineUserType PipelineOpc@szeZdZdZdedefddZdddZedefd d Z edefd d Z edefd dZ ddZ ddZ ddZdS)PipelineStateSimplea# Pipeline state contains an index and phase bit corresponding to the current position in the circular buffer. Use a single Int32 to store both the index and phase bit, then we use divmod to get the index and phase. If stages is a power of 2, divmod turns into bit twiddling. stages phase_indexcCs||_||_dSN)_stages _phase_index)selfrrrN/home/ubuntu/vllm_env/lib/python3.10/site-packages/flash_attn/cute/pipeline.py__init__s zPipelineStateSimple.__init__returncCst|j|jSr)r rrrrrrcloneszPipelineStateSimple.clonecCs|jSr)rrrrrrszPipelineStateSimple.stagescCs |j|jSrrrrrrrindex$s zPipelineStateSimple.indexcCs |j|jSrrrrrrphase*s zPipelineStateSimple.phasecCs|jd7_dS)N)rrrrradvance3szPipelineStateSimple.advancecCs|j}|gSr)rir_value)rrrrr__extract_mlir_values__Es z+PipelineStateSimple.__extract_mlir_values__cCst|jt|dS)Nr)r rr)rvaluesrrr__new_from_mlir_values__Isz,PipelineStateSimple.__new_from_mlir_values__N)rr )__name__ __module__ __qualname____doc__intrrrpropertyrrrrr r"rrrrr s  r typercCs8|tjur t|t|S|tjurt|tdSJd)zz Creates a pipeline state. Producers are assumed to start with an empty buffer and have a flipped phase bit of 1. rFzBError: invalid PipelineUserType specified for make_pipeline_state.)r Producerr rConsumer)r)rrrrmake_pipeline_stateMs  r,T)frozenc@sveZdZdZe ddejdededede de j e f d d Z dd ed eefddZd efddZd efddZd S)PipelineTmaAsyncNoClustera If size(ClusterShape) == 1, PipelineTmaAsync has all threads signaling the barrier during consumer_release. This causes a perf regression in FA3 forward pass (especially hdim 128 causal). We instead implement a version of PipelineTmaAsync where only 1 out of 128 threads signals the barrier. Assumptions: (1) num_consumers % NumThreadsPerWarpGroup == 0 (2) all 128 threads in the warp group are sync'ed right before calling consumer_release Tbarrier_storage num_stagesproducer_groupconsumer_grouptx_count init_waitcCsvtj}tj}||f}||f} t|jdd|||} t|jdd||| } d} d} t|r3tt | | || | S)a This helper function computes any necessary attributes and returns an instance of PipelineTmaAsync. :param barrier_storage: Pointer to the smem address for this pipeline's mbarriers :type barrier_storage: cute.Pointer :param num_stages: Number of buffer stages for this pipeline :type num_stages: Int32 :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 ) min_alignN) r TmaLoad AsyncThreadr_make_sync_objectaligncutlass const_exprr r.)r/r0r1r2r3r4 producer_type consumer_typeproducerconsumersync_object_fullsync_object_emptydst_rank producer_maskrrrcreategs* z PipelineTmaAsyncNoCluster.createNstatetry_acquire_tokencs6t|dup|dkfddjjjdS)z TMA producer commit conditionally waits on buffer empty and sets the transaction barrier for leader threadblocks. NrcsjjjSr)rBwaitrrrrrFrrz.)rrAarriverrD)rrFrGrrIrproducer_acquires  z*PipelineTmaAsyncNoCluster.producer_acquirecCsdS)zv TMA producer commit is a NOP. The transaction barrier signals the commit upon completion of the TMA. NrrIrrrproducer_commitsz)PipelineTmaAsyncNoCluster.producer_commitcs*ttjdddkfdddS)z^ TMA consumer release conditionally signals the empty buffer to the producer. rcsjjjSr)rBrLr consumer_maskrrIrrrJrKz.N)rcutearch thread_idxrIrrIrconsumer_releases z*PipelineTmaAsyncNoCluster.consumer_release)Tr)r#r$r%r& staticmethodrQPointerrr r'r; ConstexprboolrErrrrMrNrTrrrrr.Zs(  + r.)typingr dataclassesrr; cutlass.cuterQcutlass.cutlass_dslrrrcutlass.pipelinerrr r r r r r'r,r.rrrrs   ?