o ia@sdZddlZddlZddlZddlmZddlmZdZdZ dZ dZ dZ d Z d Zd Zd Zd ZdZdZdZdZdZdZdZdZe ee ee ee eeeeeeeeeiZddeDZdZdZdZd Z d Z!dZ"dZ#dZ$dZ%dZ&dZ'dZ(dZ)e*e)Z+dZ,Gdd d Z-e-Z. d7d!e/d"e/d#e0d$e1d%e1f d&d'Z2d%e1fd(d)Z3d%e1fd*d+Z4d%e1fd,d-Z5d%e1fd.d/Z6 d8d0e7d1e/d%e1fd2d3Z8d9d0e7d4e7d%e1fd5d6Z9dS):a6 Dirty Worker Binary Protocol Binary message framing over Unix sockets, inspired by OpenBSD msgctl/msgsnd. Replaces JSON protocol for efficient binary data transfer. Header Format (16 bytes): +--------+--------+--------+--------+--------+--------+--------+--------+ | Magic (2B) | Ver(1) | MType | Payload Length (4B) | +--------+--------+--------+--------+--------+--------+--------+--------+ | Request ID (8 bytes) | +--------+--------+--------+--------+--------+--------+--------+--------+ - Magic: 0x47 0x44 ("GD" for Gunicorn Dirty) - Version: 0x01 - MType: Message type (REQUEST, RESPONSE, ERROR, CHUNK, END) - Length: Payload size (big-endian uint32, max 64MB) - Request ID: uint64 (replaces UUID string) Payload is TLV-encoded (see tlv.py). N)DirtyProtocolError) TLVEncodersGDrequestresponseerrorchunkendstashstatusmanagecCsi|]\}}||qSr).0kvrrK/home/ubuntu/.local/lib/python3.10/site-packages/gunicorn/dirty/protocol.py Hsr z>2sBBIQic@seZdZdZeZeZeZeZ e Z e Z eZeZeZeZededededefddZededefd d Ze d7ded ed edededef ddZededefddZededefddZ ededefddZ!ededefddZ"ededefddZ#ed8dedededefdd Z$e d9deded!edefd"d#Z%ededefd$d%Z&ed&e'j(defd'd(Z)ed)e'j*d*edd fd+d,Z+ed-e,j,d.edefd/d0Z-ed-e,j,defd1d2Z.ed-e,j,d*edd fd3d4Z/ed*edefd5d6Z0d S):BinaryProtocolz-Binary message protocol for dirty worker IPC.msg_type request_idpayload_lengthreturncCstttt|||S)a, Encode the 16-byte message header. Args: msg_type: Message type (MSG_TYPE_REQUEST, etc.) request_id: Unique request identifier (uint64) payload_length: Length of the TLV-encoded payload Returns: bytes: 16-byte header )structpack HEADER_FORMATMAGICVERSION)r r!r"rrr encode_headerrs zBinaryProtocol.encode_headerdatacCst|tkrtdt|dt|dtt|dt\}}}}}|tkr7td|dt|ddd|tkrKtd|dt|ddd|tvr]td |d |ddd|t krltd |d t d |||fS)z Decode the 16-byte message header. Args: data: 16 bytes of header data Returns: tuple: (msg_type, request_id, payload_length) Raises: DirtyProtocolError: If header is invalid zHeader too short:  bytes, expected raw_dataNzInvalid magic: z , expected zUnsupported protocol version: zUnknown message type: 0x02xzMessage too large: z bytes (max: )) len HEADER_SIZErr$unpackr&r'r(MSG_TYPE_TO_STRMAX_MESSAGE_SIZE)r*magicversionr lengthr!rrr decode_headers8       zBinaryProtocol.decode_headerNapp_pathactionargskwargscCsB|||rt|ng|p id}t|}tt|t|}||S)al Encode a request message. Args: request_id: Unique request identifier (uint64) app_path: Import path of the dirty app action: Action to call on the app args: Positional arguments kwargs: Keyword arguments Returns: bytes: Complete message (header + payload) )r:r;r<r=)listrencoderr)MSG_TYPE_REQUESTr1)r!r:r;r<r= payload_dictpayloadheaderrrrencode_requests zBinaryProtocol.encode_requestcC,d|i}t|}tt|t|}||S)a Encode a success response message. Args: request_id: Request identifier this responds to result: Result value (must be TLV-serializable) Returns: bytes: Complete message (header + payload) result)rr?rr)MSG_TYPE_RESPONSEr1)r!rFrArBrCrrrencode_response  zBinaryProtocol.encode_responsecCsrddlm}t||r|}nt|tr|}n t|jt|id}d|i}t |}t t |t |}||S)a Encode an error response message. Args: request_id: Request identifier this responds to error: DirtyError instance, dict, or Exception Returns: bytes: Complete message (header + payload) r DirtyError error_typemessagedetailsr)errorsrK isinstanceto_dictdicttype__name__strrr?rr)MSG_TYPE_ERRORr1)r!rrK error_dictrArBrCrrr encode_errors    zBinaryProtocol.encode_errorcCrE)a Encode a chunk message for streaming responses. Args: request_id: Request identifier this chunk belongs to data: Chunk data (must be TLV-serializable) Returns: bytes: Complete message (header + payload) r*)rr?rr)MSG_TYPE_CHUNKr1)r!r*rArBrCrrr encode_chunkrIzBinaryProtocol.encode_chunkcCtt|d}|S)z Encode an end-of-stream message. Args: request_id: Request identifier this ends Returns: bytes: Complete message (header + empty payload) r)rr) MSG_TYPE_ENDr!rCrrr encode_end zBinaryProtocol.encode_endcCr\)z Encode a status query message. Args: request_id: Request identifier Returns: bytes: Complete message (header + empty payload) r)rr)MSG_TYPE_STATUSr^rrr encode_status r`zBinaryProtocol.encode_statusropcountcCs.||d}t|}tt|t|}||S)a4 Encode a worker management message. Args: request_id: Request identifier op: Management operation (MANAGE_OP_ADD or MANAGE_OP_REMOVE) count: Number of workers to add/remove Returns: bytes: Complete message (header + payload) )rcrd)rr?rr)MSG_TYPE_MANAGEr1)r!rcrdrArBrCrrr encode_manage/s zBinaryProtocol.encode_managetablec Cs^||d}|dur ||d<|dur||d<|dur||d<t|}tt|t|}||S)a Encode a stash operation message. Args: request_id: Unique request identifier (uint64) op: Stash operation code (STASH_OP_*) table: Table name key: Optional key for put/get/delete operations value: Optional value for put operation pattern: Optional pattern for keys operation Returns: bytes: Complete message (header + payload) )rcrgNkeyvaluepattern)rr?rr)MSG_TYPE_STASHr1) r!rcrgrhrirjrArBrCrrr encode_stashEs zBinaryProtocol.encode_stashc Cst|\}}}t|t|kr$tdt|dt||ddd|dkr+i}n0|tt|}zt|}Wn tyBtyZ}z td||dddd}~wwt|}|||fS)a Decode a complete message (header + payload). Args: data: Complete message bytes Returns: tuple: (msg_type_str, request_id, payload_dict) msg_type_str is the string name (e.g., "request") payload_dict is the decoded TLV payload as a dict Raises: DirtyProtocolError: If message is malformed zIncomplete message: expected z bytes, got N2r,rFailed to decode TLV payload: ) rr9r1r2rr decode_full Exceptionr4)r*r r!r8rA payload_datae msg_type_strrrrdecode_messagefs0     zBinaryProtocol.decode_messagereaderc s>z |tIdH}Wn%tjy0}zt|jdkrtdt|jdt|jdd}~wwt|\}}}|dkrz ||IdH}Wntjyd}ztdt|jd||jdd}~wwzt |}Wn"tytt y}z td||dddd}~wwi}t |}||d } | || S) ap Read a complete binary message from async stream. Args: reader: asyncio StreamReader Returns: dict: Message dict with 'type', 'id', and payload fields Raises: DirtyProtocolError: If read fails or message is malformed asyncio.IncompleteReadError: If connection closed mid-read NrzIncomplete header: got r+r,zIncomplete payload: got rnrmrTid) readexactlyr2asyncioIncompleteReadErrorr1partialrrr9rrorpr4update) rurCrrr r!r8rqrArsrFrrrread_message_asyncsV     z!BinaryProtocol.read_message_asyncwriterrNcs(t|}|||IdHdS)a[ Write a message to async stream. Accepts dict format for backwards compatibility. Args: writer: asyncio StreamWriter message: Message dict with 'type', 'id', and payload fields Raises: DirtyProtocolError: If encoding fails ConnectionError: If write fails N)r_encode_from_dictwritedrain)r~rNr*rrrwrite_message_asyncs  z"BinaryProtocol.write_message_asyncsockncCsnd}t||kr5||t|}|s+t|dkrtdtdt|d||d||7}t||ks|S)a Receive exactly n bytes from a socket. Args: sock: Socket to read from n: Number of bytes to read Returns: bytes: Received data Raises: DirtyProtocolError: If read fails or connection closed rzConnection closedzConnection closed after r+r,)r1recvr)rrr*rrrr _recv_exactlys    zBinaryProtocol._recv_exactlyc Cst|t}t|\}}}|dkr@t||}zt|}Wn"ty'ty?}z td||dddd}~wwi}t|}||d} | || S)a  Read a complete message from socket (sync). Args: sock: Socket to read from Returns: dict: Message dict with 'type', 'id', and payload fields Raises: DirtyProtocolError: If read fails or message is malformed rrnNrmr,rv) rrr2r9rrorrpr4r|) rrCr r!r8rqrArrrsrFrrr read_messages(      zBinaryProtocol.read_messagecCst|}||dS)a Write a message to socket (sync). Args: sock: Socket to write to message: Message dict with 'type', 'id', and payload fields Raises: DirtyProtocolError: If encoding fails OSError: If write fails N)rrsendall)rrNr*rrr write_message/s zBinaryProtocol.write_messagec Csd|d}|dd}t|trt|d@}t|}|dur&td||tkrAt||dd|d d|d |d S|t krNt ||d S|t kr\t ||d iS|t krit||dS|tkrrt|S|tkrt||d|dd|d|d|dS|tkrt|S|tkrt||d|ddStd|)a Encode a message dict to binary format. Supports the old dict-based API for backwards compatibility. Args: message: Message dict with 'type', 'id', and payload fields Returns: bytes: Complete encoded message rTrwrlNzUnknown message type: r:r;r<r=rFrr*rcrgrhrirjrdrzUnhandled message type: )getrQrVhashMSG_TYPE_FROM_STRrr@rrDrGrHrWrYrZr[r]r_rkrlrarbrerf)rNrsr!r rrrr?sd           z BinaryProtocol._encode_from_dictNNrNNN)1rU __module__ __qualname____doc__r2r5MSG_TYPE_REQUEST_STRr@MSG_TYPE_RESPONSE_STRrGMSG_TYPE_ERROR_STRrWMSG_TYPE_CHUNK_STRrZMSG_TYPE_END_STRr]MSG_TYPE_STASH_STRrkMSG_TYPE_STATUS_STRraMSG_TYPE_MANAGE_STRre staticmethodintbytesr)tupler9rVrSrDrHrYr[r_rbrfrlrtry StreamReaderr} StreamWriterrsocketrrrrrrrrrbsx0   0; 'rr:r;r<r=r#cCs$tj||||r t|ng|pidS)aU Build a request message dict. Args: request_id: Unique request identifier (int or str) app_path: Import path of the dirty app (e.g., 'myapp.ml:MLApp') action: Action to call on the app args: Positional arguments kwargs: Keyword arguments Returns: dict: Request message dict )rTrwr:r;r<r=) DirtyProtocolr@r>)r!r:r;r<r=rrr make_requestsrcCtj||dS)z Build a success response message dict. Args: request_id: Request identifier this responds to result: Result value Returns: dict: Response message dict )rTrwrF)rrG)r!rFrrr make_response rcCsTddlm}t||r|}nt|tr|}n t|jt|id}tj ||dS)z Build an error response message dict. Args: request_id: Request identifier this responds to error: DirtyError instance or dict with error info Returns: dict: Error response message dict rrJrL)rTrwr) rPrKrQrRrSrTrUrVrrW)r!rrKrXrrrmake_error_responses   rcCr)z Build a chunk message dict for streaming responses. Args: request_id: Request identifier this chunk belongs to data: Chunk data Returns: dict: Chunk message dict )rTrwr*)rrZ)r!r*rrrmake_chunk_messagerrcCs tj|dS)z Build an end-of-stream message dict. Args: request_id: Request identifier this ends Returns: dict: End message dict rv)rr])r!rrrmake_end_messages rrcrgcCsDtj|||d}|dur||d<|dur||d<|dur ||d<|S)a Build a stash operation message dict. Args: request_id: Unique request identifier (int or str) op: Stash operation code (STASH_OP_*) table: Table name key: Optional key for put/get/delete operations value: Optional value for put operation pattern: Optional pattern for keys operation Returns: dict: Stash message dict )rTrwrcrgNrhrirj)rrk)r!rcrgrhrirjmsgrrrmake_stash_messagesrrdcCstj|||dS)a Build a worker management message dict. Args: request_id: Unique request identifier (int or str) op: Management operation (MANAGE_OP_ADD or MANAGE_OP_REMOVE) count: Number of workers to add/remove Returns: dict: Manage message dict )rTrwrcrd)rre)r!rcrdrrrmake_manage_messages  rrrr):rryrr$rPrtlvrr'r(r@rGrWrZr]rkrarerrrrrrrrr4itemsr STASH_OP_PUT STASH_OP_GETSTASH_OP_DELETE STASH_OP_KEYSSTASH_OP_CLEAR STASH_OP_INFOSTASH_OP_ENSURESTASH_OP_DELETE_TABLESTASH_OP_TABLESSTASH_OP_EXISTS MANAGE_OP_ADDMANAGE_OP_REMOVEr&calcsizer2r5rrrVrrSrrrrrrrrrrrrs    -