E/ddlmZddlZddlZddlmZmZmZddlZddl m Z ddl m Z ejeZee j ZGddZGdd eZy) ) OrderedDictN)DictOptionalType)types) exceptionsc*eZdZdZdededefdZdZy)_QuantityReservationz2A (partial) reservation of quantifiable resources.bytes_reserved bytes_neededhas_slotc.||_||_||_y)Nr r r )selfr r r s Clib/third_party/google/cloud/pubsub_v1/publisher/flow_controller.py__init__z_QuantityReservation.__init__"s,(  ct|jd|jd|jd|jdS)Nz(bytes_reserved=z, bytes_needed=z , has_slot=))type__name__r r r )rs r__repr__z_QuantityReservation.__repr__'sNDz""#$"1123 --./ a ) rN)r __module__ __qualname____doc__intboolrrrrr r s$<!s!#!!  rr ceZdZdZdej fdZdeddfdZdeddfdZ dd Z de fd Z dede fd Z dd eed eedefdZy)FlowControllerzA class used to control the flow of messages passing through it. Args: settings: Desired flow control configuration. settingsc||_d|_d|_t|_d|_d|_tj|_ tj|j|_ y)Nr)lock) _settings_message_count _total_bytesr_waiting_reserved_bytes_reserved_slots threadingLock_operational_lock Condition _has_capacity)rr!s rrzFlowController.__init__7sd!  GRm   "+!1'00d6L6LMrmessagereturnNcH|jjtjjk(ry|j 5|j |sL|xjdz c_|xj|jjz c_ dddy|jjtjjk(rk|j|jdz|j|jjz}dj|}tj||jjtjj k(sJ|jj|jj"kDs|jj$dkrQ|jd|jj}dj|}tj|t'j(}|j |r||j*vr5t-d|jjd}||j*|<t.j1d j|j|j2j5t.j1d j|j|j |r|xjdz c_|xj|jjz c_|xj6|j*|j8zc_|xj:dzc_|j*|=dddy#1swYyxYw) aAdd a message to flow control. Adding a message updates the internal load statistics, and an action is taken if these limits are exceeded (depending on the flow control settings). Args: message: The message entering the flow control. Raises: :exception:`~pubsub_v1.publisher.exceptions.FlowControlLimitError`: Raised when the desired action is :attr:`~google.cloud.pubsub_v1.types.LimitExceededBehavior.ERROR` and the message would exceed flow control limits, or when the desired action is :attr:`~google.cloud.pubsub_v1.types.LimitExceededBehavior.BLOCK` and the message would block forever against the flow control limits. N) message_count total_bytesz+Flow control limits would be exceeded - {}.zLTotal flow control limits too low for the message, would block forever - {}.rFrz>Blocking until there is enough free capacity in the flow - {}.z7Woke up from waiting on free capacity in the flow - {}.)r$limit_exceeded_behaviorrLimitExceededBehaviorIGNOREr,_would_overflowr%r&_pbByteSizeERROR _load_infoformatrFlowControlLimitErrorBLOCK byte_limit message_limitr*current_threadr'r _LOGGERdebugr.waitr(r r))rr/ load_info error_msgrB reservations raddzFlowController.addNs$ >> 1 1U5P5P5W5W W   # #''0##q(#!!W[[%9%9%;;! $ #66..445 !OO"&"5"5"9 $ 1 1GKK4H4H4J J, JPP !66yAA66..445 5 $$&)B)BB>>//!3 OO"#1E1E1G, 006y0A!66yAA&557N&&w/!6"6'(%,[[%9%9%;!&#K 5@DMM.1  &!23 ""'')  &!23!&&w/,   1 $    !5!5!7 7   DMM.$A$P$P P  A %  n-O$ # #sANI!N?BNN!c|jjtjjk(ry|j 5|xj dzc_|xj|jjzc_|j dks|jdkrRtjdtdtd|j |_td|j|_|j|jr/t j#d|j$j'dddy#1swYyxYw)zRelease a mesage from flow control. Args: message: The message entering the flow control. Nr2rz=Releasing a message that was never added or already released.)category stacklevelz2Notifying threads waiting to add messages to flow.)r$r5rr6r7r,r%r&r9r:warningswarnRuntimeWarningmax_distribute_available_capacity_ready_to_unblockrCrDr. notify_all)rr/s rreleasezFlowController.releases >> 1 1U5P5P5W5W W   # #   1 $    !5!5!7 7 ""Q&$*;*;a*? S+  '*!T-@-@&A#$'4+<+<$=!  / / 1%%' RS""--/'$ # #s DE  Ec|jj|jz |jz }|jj|j z |j z }|jjD]}|dkr|dkry|dkDr-|js!d|_ |xjdz c_|dz}|dkrG|j|jz }|dkrCdj|j|j}tj|td}t!||}|xj|z c_ |xj |z c_||z}y)zDistribute available capacity among the waiting threads in FIFO order. The method assumes that the caller has obtained ``_operational_lock``. rTr2z Too many bytes reserved: {} / {})rLN)r$rAr%r)r@r&r(r'valuesr r r r=rNrOrPmin)ravailable_slotsavailable_bytesrHbytes_still_neededmsgcan_gives rrRz-FlowController._distribute_available_capacitysR NN ( (4+>+> >AUAU U  NN % %(9(9 9D*>*@%A B !004E4R4RR/%..  rc|jjtj}|r&|j|j k\}|j }nd}d}|j|jz|jjz}||jjkDxr| }| xr3|j|jzdz|jjkD}|xs|S)zDetermine if accepting a message would exceed flow control limits. The method assumes that the caller has obtained ``_operational_lock``. Args: message: The message entering the flow control. Fr2)r'getr*rBr r r r&r(r9r:r$r@r%r)rA)rr/rHenough_reservedr bytes_taken size_overflowmsg_count_overflows rr8zFlowController._would_overflowsmm'' (@(@(BC )88K*>>AUAUAWW #dnn&?&??WDW !)\  4#7#7 7! ;nn** +  2 22rr3r4c | |j}| |j}d|d|jjd|jd|d|jj d|j d S)aReturn the current flow control load information. The caller can optionally adjust some of the values to fit its reporting needs. The method assumes that the caller has obtained ``_operational_lock``. Args: message_count: The value to override the current message count with. total_bytes: The value to override the current total bytes with. z messages: z / z (reserved: z ), bytes: r)r%r&r$rAr)r@r()rr3r4s rr<zFlowController._load_infos   //M  ++Ks4>>+G+G*HI../0!]#dnn&?&?%@A../q 2 r)r0N)NN)rrrrrPublishFlowControlr MessageTyperIrUrRrrSr8rrstrr<rrrr r 0s N!9!9N.\.;\.4\.|0{0t0>&(P4 3{3t38QU %c] @H   rr ) collectionsrloggingr*typingrrrrNgoogle.cloud.pubsub_v1r google.cloud.pubsub_v1.publisherr getLoggerrrC PubsubMessagerjr objectr rrrrts]$''(7 '  H %5&&'   "I VI r