o
    `j                     @   s   d Z ddlZddlZddlZddlZddlmZ ddlmZ ddl	m
Z
 ddlmZ dZdZd	Zd
d Zdd Zdd Zdd Zdd Zdd Zdd ZedkrUe  dS dS )a  Benchmarks for the producer encode hot path.

Measures the cost of finalizing a record batch and encoding it into a
ProduceRequest - the pipeline that runs on every send to the broker.

To compare two implementations (e.g. before/after a change) run this
script twice and diff the output:

    # baseline
    git stash
    python -m kafka.benchmarks.producer_encode_path -o baseline.json
    git stash pop

    # new
    python -m kafka.benchmarks.producer_encode_path -o new.json

    pyperf compare_to baseline.json new.json

For an allocation count (tracemalloc) run with --allocations, which is
a separate mode (not a timing benchmark).
    N)ProducerBatch)MemoryRecordsBuilder)TopicPartition)ProduceRequestd      i   c                 C   s`   t dd}tddtd}t||}d| }t| D ]}||d |g }|d us-J d|  q|S )Nbenchr      magiccompression_type
batch_size   xzbatch too small for %d records)r   r   DEFAULT_BATCH_SIZEr   range
try_append)num_records
value_sizetprecordsbatchvalueifuture r   c/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/benchmarks/producer_encode_path.py_build_unclosed_batch(   s   

r   c                 C   s   t | |}|j  |S N)r   r   close)r   r   r   r   r   r   _build_closed_batch4   s   

r   c           
      C   st   d| }t dd}t }t| D ]"}tddtd}t||}t|D ]
}	||	d|g  q#|j	  qt | S )a  Time build + close - isolates the batch-finalization path.

    close() is fast enough that pyperf would need many thousands of loops
    to measure it directly, so we include the build cost and rely on the
    per-run diff to expose the close() delta.
    r   r   r   r	   r
   N)
r   pyperfperf_counterr   r   r   r   r   r   r   )
loopsr   r   r   r   t0_r   r   r   r   r   r   bench_build_and_close:   s   

r%   c           	      C   sv   t ||}|j }dd|fgfg}t }t| D ]}td ddd|d}|jddd |jd	d	d
 qt | S )z|Time ProduceRequest.encode() with a pre-closed batch.

    Isolates the protocol encoding cost from batch finalization.
    r   r      N   0u  transactional_idacks
timeout_ms
topic_datacorrelation_id	client_idTframedheader)	r   r   bufferr    r!   r   r   with_headerencode)	r"   r   r   r   bufr-   r#   r$   reqr   r   r   bench_encodeO   s   

r9   c              	   C   s   d| }t dd}t }t| D ]C}tddtd}t||}t|D ]
}	||	d|g  q#|j	  t
d ddd	dd|j fgfgd
}
|
jddd |
jddd qt | S )zTime the full producer hot path: append -> close -> encode.

    Mirrors what the Sender thread does for each drained batch.
    r   r   r   r	   r
   Nr&   r'   r(   r)   r.   Tr1   )r   r    r!   r   r   r   r   r   r   r   r   r4   r5   r6   )r"   r   r   r   r   r#   r$   r   r   r   r8   r   r   r   bench_full_pipelinee   s(   


r:   c                 C   sZ  t | |}|j }dd|fgfg}td ddd|d}|jddd |jd	d	d
 t  t }|jd	d	d
}t }t	  |
|d}	tdd |	D }
tdd |	D }td| |f  tdt|  td|
|f  t  td t|	dd d	ddd D ]%}|jdkrq|jd }d|jdd |jf }td|j|j|f  qdS )z@Use tracemalloc to count bytes allocated during a single encode.r   r   r&   Nr'   r(   r)   r.   Tr1   linenoc                 s       | ]	}t d |jV  qdS r   N)max	size_diff.0sr   r   r   	<genexpr>       z%report_allocations.<locals>.<genexpr>c                 s   r<   r=   )r>   
count_diffr@   r   r   r   rC      rD   z%Config        : %d records x %d byteszWire bytes    : %dz.Allocated     : %d bytes across %d allocationszTop allocation sites:c                 S   s   | j S r   )r?   )rB   r   r   r   <lambda>   s    z$report_allocations.<locals>.<lambda>)keyreverse
   z%s:%dz/workspace/z  %8d bytes / %3d allocs  %s)r   r   r4   r   r5   r6   tracemallocstarttake_snapshotstop
compare_tosumprintlensortedr?   	tracebackfilenamesplitr;   rE   )r   r   r   r7   r-   r8   snap_beforedata
snap_afterstatstotal_bytestotal_allocsrB   framelocr   r   r   report_allocations   s>   



r_   c                  C   s   t jtt jd} | jdttdd | jdttdd | jddd	d
 |  \}}|j	r4t
|j|j d S tjd g| t_t }|j|jd< |j|jd< |dt|j|j |dt|j|j |dt|j|j d S )N)descriptionformatter_classz	--recordsz(records per batch (default: %(default)d))typedefaulthelpz--value-sizez1record value size in bytes (default: %(default)d)z--allocations
store_truez@report tracemalloc allocations for one encode (not a timing run))actionrd   r   records_per_batchvalue_size_bytesbuild_and_closer6   full_pipeline)argparseArgumentParser__doc__RawDescriptionHelpFormatteradd_argumentintDEFAULT_RECORDS_PER_BATCHDEFAULT_VALUE_SIZEparse_known_argsallocationsr_   r   r   sysargvr    Runnermetadatabench_time_funcr%   r9   r:   )pargspyperf_argsrunnerr   r   r   main   s:   

r~   __main__)rm   rk   ru   rK   r    kafka.producer.producer_batchr   kafka.record.memory_recordsr   kafka.structsr   kafka.protocol.producerr   rq   rr   r   r   r   r%   r9   r:   r_   r~   __name__r   r   r   r   <module>   s,   )
