....................................../////.===Hehe-Here===./////................................................ > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < > < ------------------------------------------------------------------------------------------------------------------- /////////////////////////////////////////////////////////////////////////////////////////////////////////////////// RIFF¤ WEBPVP8 ˜ ðÑ *ôô>‘HŸK¥¤"§£±¨àð enü¹%½_F‘åè¿2ºQú³íªú`N¿­3ÿƒügµJžaÿ¯ÿ°~¼ÎùnúîÞÖô•òíôÁÉß®Sm¥Ü/ ‡ó˜f£Ùà<˜„xëJ¢Ù€SO3x<ªÔ©4¿+ç¶A`q@Ì“Úñè™ÍÿJÌ´ª-˜ÆtÊÛL]Ïq*‘Ý”ì#ŸÌÏãY]@ê`¿ /ªfkØB4·®£ó z—Üw¥Pxù–ÞLШKÇN¾AkÙTf½è'‰g gÆv›Øuh~ a˜Z— ïj*á¥t d£“uÒ ¨`K˜¹ßþ]b>˜]_ÏÔ6W—è2r4x•íÖ…"ƒÖNîä!¦å Ú}ýxGøÌ —@ ;ÆÚŠ=ɾ1ý8lªË¥ô ^yf®Œ¢u&2©nÙÇ›ñÂñŒ³ aPo['½»øFùà­+4ê“$!lövlüÞ=;N®3ð‚õ›DÉKòÞ>ÄÍ ¥ˆuߤ#ˆ$6ù™¥îŠ‡y’ÍB¼ çxÛ;X"WL£R÷͝*ó-¶Zu}º.s¸sšXqù–DþÿvªhüïwyŸ ¯é³lÀ:KCûÄ£Ëá\…­ ~—ýóî ¼ûûÜTÓüÇy…ŽÆvc»¾×U ñ¸žþоP÷¦ó:Ò¨¨5;Ð#&#ÖúñläÿÁœ GxÉ­/ñ‡áQðìYÉtÒwŽ¼GÔ´zàÒò ð*ëzƒ•4~H]Ø‹f ñÓÈñ`NåWçs'ÆÏW^ø¹!XžµmQ5ÃËoLœÎ: ÞËÍ¥J ù…î èo£ßPÎñ¶ž8.Œ]ʵ~5›ÙË-ù*8ÙÖß±~ ©¹rÓê‚j¶d¸{^Q'˜±Crß ÚH—#¥¥QlÀ×ëã‡DÜ«èî þ&Çæžî;ŽÏºò6ÒLÃXy&ZŒ'j‚¢Ù€IßÚù+–MGi‰*jE€‘JcÜ ÓÌ EÏÚj]o˜ Þr <¾U ûŪæÍ/šÝH¥˜b”¼ ÁñßX GP›ï2›4WŠÏà×£…íÓk†¦H·ÅíMh–*nó÷à]ÁjCº€b7<ب‹¨5車bp2:Á[UªM„QŒçiNMa#<5›áËó¸HýÊ"…×Éw¹¦ì2º–x<›»a±¸3Weü®FÝ⑱ö–î–³|LPÈ~çð~Çå‡|º kD¢µÏàÆAI %1À% ¹Ò – ”ϝS¦‰4&¶£°à Öý”û_Ò Áw°A«Å€?mÇÛgHÉ/8)á¾ÛìáöŽP í¨PŸNÙµº¦‡§Ùš"ÿ«>+ªÕ`Ê÷‡‚ß Õû˜þãÇ-PÍ.¾XV‘€ dÜ"þ4¹ ±Oú‘©t¥¦FªÄÃÄ•b‚znýu½—#cDs˜ÃiÑOˆñ×QO=*IAÊ,¶ŽZƒ;‡wøXè%EÐk:F±Ú” .Ѽ+Áu&Ç`."pÈÉw o&¿dE6‘’EqTuK@Ì¥ã™À(Êk(h‰,H}RÀIXÛš3µ1©_OqÚÒJAñ$ÊÙÜ;D3çŒ[þùœh¬Ã³™ö6ç†NY".Ú‰ï[ªŸŒ '²Ð öø_¨ÂÉ9u鶳ÒŠõTàîMØ#û¯gN‡bÙ놚X„ö …ÉeüÌ^J ‹€.œ$Æ)βÄeæW#óüßĺŸ€ ÀzwV 9oä»f4V*uB «Ë†¹ì¯žR霓æHXa=&“I4K;¯ç‹h×·"UŠ~<•â•ªVêª&ÍSÃÆÅ?ÔqÎ*mTM ˜›µwêd#[C¡©§‘D<©àb†–ÁœøvH/,í:¯( ²£|4-„Æövv„Yͼ™^Á$ˆ„¢Û[6yB.åH*V¨æ?$=˜Ñ€•î¦ñ·­(VlŸ‘ nÀt8W÷´Bûba?q9ú¶Xƒl«ÿ\ù¶’þòUÐj/õ¢Ìµ³g$ƒÎR!¸»|Oߍë’BhîÚÑ¢ñåŒJ„®„£2Ð3•ô02Nt…!£Í]Ïc½Qÿ?ˆ<&ÃA¾Ú,JˆijÌ#5yz„‰Î|ÊŽ5QÏ:‹ÐaóVÔxW—CpeÏzÐïíçôÿÅ_[hãsÐ_/ŽTÝ?BîˆííV$<¿i>²F¬_Eß¿ †bÊŒº­ÿ®Z H“C}”¬,Mp ý/Bá£w>˜YV°aƒúh+cŠ- r/[%|üUMHäQ°X»|û/@|°¥Ð !BÔ Ç¢Ä©š+Õì D«7ìN¶ŽðÔ " ƶ’ÖçtA‰Û×}{tþz­¾GÍ›k¹OEJR$ Â׃ «ëÁ"oÉôž$oUK(Ä)Ãz³Ê-‹êN[Ò3Œñbï8P 4ƒ×q¢bo|?<ÛX¬òÄÍ°L–±›(™ûG?ýË©ÚÄ–ÂDØÐ_Ç¡ô ¾–ÄÏø ×e8Ë©$ÄF¹Å‹ì[©óìl:F¾f´‹‹Xì²ï®\¬ôùƒ ÿat¥óèÒùHß0äe‚;ü×h:ÆWðHž=Ã8骣"kœ'Y?³}Tûè€>?0l›e1Lòñ„aæKÆw…hÖŠùW…ÈÆÄ0ši·›[pcwËþñiêíY/~-Á5˜!¿†A›™Mÿþ(±“t@â“ö2­´TG5yé]çå僳 .·ÍïçÝ7UÚ±Ð/Nè»,_Ï ùdj7\ï Wì4›„»c¸àešg#ÒÊ⥭áØo5‘?ÌdÝô¯ ¹kzsƒ=´#ëÉK›Ø´±-¥eW?‡çßtòTã…$Ý+qÿ±ƒ÷_3Ô¥í÷:æ–ž<·Ö‡‰Å¢ š‡%Ô—utÌÈìðžgÖÀz²À—ï÷Óîäõ{K'´È÷³yaÏÁjƒô}ž§®æÊydÕÈë5¯èˆõvÕ©ã*çD„ “z„Ó‡^^xÂ3M§A´JG‚öï 3W'ˆ.OvXè¡ÊÕª?5º7†˜(˜Ç¶#çê’¶!ÌdZK§æ 0fãaN]òY³RV ™î$®K2R¨`W!1Ôó\;Ý ýB%qæK•&ÓÈe9È0êI±žeŸß -ú@žQr¦ ö4»M¼Áè¹µmw 9 EÆE_°2ó„ŸXKWÁ×Hóì^´²GѝF©óäR†¦‰ç"V»eØ<3ùd3ÿÚ¤Žú“Gi" —‘_ÙËÎ~Üö¯¥½Î»üŸEÚŽåmÞþí ;ÞólËΦMzA"Âf(´òá;Éï(/7½ûñÌ­cïÕçлþÝz¾-ÍvÑ“pH­–ðÓj$¸Äû¤‚‘ãUBË-n“2åPkS5&‹Â|+g^œ®Ì͆d!OïäîU«c;{Û!ÅŽ«ëZ9Ókóˆ]¯ƒ›né `ÇÒ+tÆš (ØKá¾—=3œ®•vuMñg²\ï Ec€ 05±d™‡×iÇ×›UúvÌ¢£Èþ¡ÕØô¶ßÎA"ß±#Ö²ˆÊŸ¦*Ä~ij|àø.-¼'»Ú¥£h ofº¦‡VsR=N½„Î v˜Z*SÌ{=jÑB‹tê…;’HžH¯8–îDù8ñ¢|Q•bÛçš–‹m³“ê¨ åÏ^m¬Žãþ©ïêO‡½6] µÆ„Ooòü ²x}N¦Ë3ïé¿»€›HA˜m%çÞ/¿í7Fø“‹léUk)É°Œµ8Q8›:ÀŠeT*šõ~ôڝG6 ¢}`ùH­–”¡k ‰P1>š†®9z11!X wKfmÁ¦xÑ,N1Q”–æB¶M…ÒÃv6SMˆhU¬ÊPŽï‘öj=·CŒ¯u¹ƒVIЃsx4’ömÛýc塶7ßŠß 57^\wÒÐÆ k§h,Œý î«q^R½3]J¸ÇðN ‚çU¬ôº^Áì} ³f©Õœ§ˆã:FÄÈ‚é(€™?àýÓüè1Gô£¼éj‚OÅñ  #>×—ßtà 0G¥Åa뀐kßhc™À_ÉñÞ#±)GD" YîäË-ÿÙ̪ ¹™a¯´¢E\ÝÒö‚;™„ë]_ p8‰o¡ñ+^÷ 3‘'dT4œŽ ðVë½° :¬víÑ«£tßÚS-3¶“þ2 †üüʨòrš¹M{É_¤`Û¨0ìjœøJ‡:÷ÃáZ˜†@GP&œÑDGÏs¡þ¦þDGú‘1Yá9Ôþ¼ ûø…§÷8&–ÜÑnÄ_m®^üÆ`;ÉVÁJ£?â€-ßê}suÍ2sõA NÌúA磸‘îÿÚ»ƒìö·á¿±tÑÐ"Tÿü˜[@/äj¬€uüªìù¥Ý˜á8Ý´sõj 8@rˆð äþZÇD®ÿUÏ2ùôõrBzÆÏÞž>Ì™xœ“ wiÎ×7_… ¸ \#€MɁV¶¥üÕÿPÔ9Z‡ø§É8#H:ƒ5ÀÝå9ÍIŒ5åKÙŠ÷qÄ>1AÈøžj"µÂд/ªnÀ qªã}"iŸBå˜ÓÛŽ¦…&ݧ;G@—³b¯“•"´4í¨ôM¨åñC‹ïùÉó¯ÓsSH2Ý@ßáM‡ˆKÀªÛUeø/4\gnm¥‹ŸŒ qÄ b9ÞwÒNÏ_4Ég³ú=܆‚´ •â¥õeíþkjz>éÚyU«Íӝ݃6"8/ø{=Ô¢»G¥ äUw°W«,ô—¿ãㆅү¢³xŠUû™yŒ (øSópÐ 9\åTâ»—*oG$/×ÍT†Y¿1¤Þ¢_‡ ¼ „±ÍçèSaÓ 3ÛMÁBkxs‰’R/¡¤ˆÙçª(*õ„üXÌ´ƒ E§´¬EF"Ù”R/ÐNyÆÂ^°?™6¡œïJ·±$§?º>ÖüœcNÌù¯G ‹ñ2ŠBB„^·úìaz¨k:#¨Æ¨8LÎõލ£^§S&cŒÐU€ü(‡F±Š¼&P>8ÙÁ ‰ p5?0ÊƃZl¸aô š¼¡}gÿ¶zÆC²¹¬ÎÖG*HB¡O<º2#ñŒAƒ–¡B˜´É$¥›É:FÀÔx¾u?XÜÏÓvN©RS{2ʈãk9rmP¼Qq̳ è¼ÐFׄ^¡Öì fE“F4A…!ì/…¦Lƒ… … $%´¾yã@CI¬ á—3PþBÏNÿ<ý°4Ü ËÃ#ØÍ~âW«rEñw‹eùMMHß²`¬Öó½íf³:‹k˜¯÷}Z!ã¿<¥,\#öµÀ¯aÒNÆIé,Ћ–lŽ#Àæ9ÀÒS·I’½-Ïp Äz¤Š Â* ­íÄ9­< h>׍3ZkËU¹§˜ŒŠ±f­’¤º³Q ÏB?‹#µíÃ¥®@(Gs«†vI¥Mµ‹Á©e~2ú³ÁP4ìÕi‚²Ê^ö@-DþÓàlÜOÍ]n"µã:žpsŽ¢:! Aõ.ç~ÓBûH÷JCÌ]õVƒd «ú´QÙEA–¯¯Œ!.ˆˆëQ±ù œ·Ì!Õâ )ùL„ÅÀlÚè5@B…o´Æ¸XÓ&Û…O«˜”_#‡ƒ„ûÈt!¤ÁÏ›ÎÝŠ?c9 â\>lÓÁVÄÑ™£eØY]:fÝ–—ù+p{™ðè û³”g±OƒÚSù£áÁÊ„ä,ï7š²G ÕÌBk)~ÑiCµ|h#u¤¶îK¨² #²vݯGãeÖ϶ú…¾múÀ¶þÔñ‚Š9'^($¤§ò “š½{éúp÷J›ušS¹áªCÂubÃH9™D™/ZöØÁ‡¦ÝÙŸ·kð*_”.C‹{áXó€‡c¡c€§/šò/&éš÷,àéJþ‰X›fµ“C¨œ®r¬"kL‰Â_q…Z–.ÉL~O µ›zn‚¹À¦Öª7\àHµšÖ %»ÇníV[¥*Õ;ƒ#½¾HK-ÖIÊdÏEÚ#=o÷Óò³´Š: Ç?{¾+9›–‘OEáU·S€˜j"ÄaÜ ŒÛWt› á–c#a»pÔZÞdŽtWê=9éöÊ¢µ~ ë ;Öe‡Œ®:bî3±ýê¢wà¼îpêñ¹¾4 zc¾ðÖÿzdêŒÑÒŝÀ‰s6¤í³ÎÙB¿OZ”+F¤á‡3@Ñëäg©·Ž ˆèª<ù@É{&S„œÕúÀA)‰h:YÀ5^ÂÓŒ°õäU\ ùËÍû#²?Xe¬tu‰^zÒÔãë¼ÛWtEtû …‚g¶Úüâî*moGè¨7%u!]PhÏd™Ý%Îx: VÒ¦ôÊD3ÀŽKÛËãvÆî…N¯ä>Eró–ð`5 Œ%u5XkñÌ*NU%¶áœÊ:Qÿú»“úzyÏ6å-၇¾ ´ ÒÊ]y žO‘w2Äøæ…H’²f±ÎÇ.ª|¥'gîV•Ü .̘¯€šòü¤U~Ù†*¢!?ò wý,}´°ÔÞnïoKq5µb!áÓ3"vAßH¡³¡·G(ÐÎ0Îò¼MG!/ài®@—¬04*`…«é8ªøøló“ˆÊ”èù¤…ßÊoÿé'ËuÌÖ5×È¡§ˆˆfŽë9}hìâ_!!¯  B&Ëö¶‰ÀAÙNVŸ Wh›¸®XÑJì¨ú“¿÷3uj²˜¨ÍÎìë±aúŠÝå¯ð*Ó¨ôJ“yºØ)m°WýOè68†ŸÏ2—‰Ïüꪫٚ¥‹l1 ø ÏÄFjêµvÌbü¦èÝx:X±¢H=MÐß—,ˆÉÇ´(9ú¾^ÅÚ4¿m‡$âX‘å%(AlZo@½¨UOÌÕ”1ø¸jÎÀÃÃ_ µ‘Ü.œº¦Ut: Æï’!=¯uwû#,“pþÇúŒø(é@?³ü¥‘Mo §—s@Œ#)§ŒùkL}NOÆêA›¸~r½¼ÙA—HJ«eˆÖ´*¡ÓpÌŸö.m<-"³ûÈ$¬_6­åf£ïÚâj1y§ÕJ½@dÞÁr&Í\Z%D£Íñ·AZ Û³øüd/ªAi†/Й~  ‡âĮҮÏh§°b—›Û«mJžòG'[ÈYýŒ¦9psl ýÁ ®±f¦x,‰½tN ‚Xª9 ÙÖH.«Lo0×?͹m¡å†Ѽ+›2ƒF ±Ê8 7Hցϓ²Æ–m9…òŸï]Â1äN†VLâCˆU .ÿ‰Ts +ÅÎx(%¦u]6AF Š ØF鈄‘ |¢¶c±soŒ/t[a¾–û:s·`i햍ê›ËchÈ…8ßÀUÜewŒðNOƒõD%q#éû\9¤x¹&UE×G¥ Í—™$ð E6-‡¼!ýpãÔM˜ Âsìe¯ñµK¢Ç¡ùôléœ4Ö£”À Š®Ðc ^¨À}ÙËŸ§›ºê{ÊuÉC ×Sr€¤’fÉ*j!úÓ’Gsùìoîßîn%ò· àc Wp÷$¨˜)û»H ×8ŽÒ€Zj¤3ÀÙºY'Ql¦py{-6íÔCeiØp‘‡XÊîÆUߢ܂ž£Xé¼Y8þ©ëgñß}é.ÎógÒ„ÃØËø¯»™§Xýy M%@NŠ À(~áÐvu7&•,Ù˜ó€uP‡^^®=_E„jt’ 403WebShell
403Webshell
Server IP : 159.198.67.129  /  Your IP : 216.73.216.244
Web Server : LiteSpeed
System : Linux server166.web-hosting.com 4.18.0-513.18.1.lve.el8.x86_64 #1 SMP Thu Feb 22 12:55:50 UTC 2024 x86_64
User : trooaisr ( 4033)
PHP Version : 7.4.33
Disable Function : NONE
MySQL : OFF  |  cURL : ON  |  WGET : ON  |  Perl : ON  |  Python : ON  |  Sudo : OFF  |  Pkexec : OFF
Directory :  /opt/cloudlinux/venv/lib64/python3.11/site-packages/lvestats/plugins/generic/burster/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Command :

[ Back ]     

Current File : /opt/cloudlinux/venv/lib64/python3.11/site-packages/lvestats/plugins/generic/burster/lve_sm.py
# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2023 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

# pylint: disable=not-an-iterable

import inspect
import contextlib
from dataclasses import dataclass
from datetime import timedelta
from typing import TypeAlias, NamedTuple, AbstractSet, Self

from lvestats.orm import BurstingEventType

from .utils import bootstrap_gen
from ._logs import logger
from .common import (
    BurstingMultipliers,
    LveId,
    Timestamp,
    LveLimits,
    ApplyLveSettings,
    LveStats,
    InvalidStateError,
    LveUsage,
    LveState,
    calc_bursted_limits,
    infer_lve_state,
)
from .history import LveHistory
from .notify import Emitter, Signal


class LveStateManager:
    @dataclass(frozen=True, slots=True)
    class Disappered(RuntimeError):
        now: Timestamp

    @dataclass(frozen=True, slots=True)
    class Burst:
        now: Timestamp

    @dataclass(frozen=True, slots=True)
    class Unburst:
        now: Timestamp

    # NOTE(vlebedev): Seems that namedtuples are ~20% faster to create compared to
    #                 classes - dataclasses or usual ones.
    class UpdateReadings(NamedTuple):
        now: Timestamp
        normal_limits: LveLimits | None
        stats: LveStats | None
        usage: LveUsage | None

    Request: TypeAlias = UpdateReadings | Disappered | Burst | Unburst

    @dataclass(frozen=True, slots=True)
    class _StateSwitched(RuntimeError):
        pass

    def __init__(
        self,
        now: Timestamp,
        lve_id: LveId,
        bursting_multipliers: BurstingMultipliers,
        initial_normal_limits: LveLimits,
        initial_stats: LveStats,
        initial_usage: LveUsage,
        quota: timedelta,
        quota_window: timedelta,
        apply_lve_settings: ApplyLveSettings,
        initial_history: LveHistory = LveHistory(),
        fail_fast: bool = True,
    ):
        if quota > quota_window:
            raise ValueError('Bursting quota must not exceed bursting quota window!')

        # NOTE(vlebedev): Constants start here.
        self._lve_id = lve_id
        self._bursting_multipliers = bursting_multipliers
        self._quota_sec = quota.total_seconds()
        self._quota_window_sec = quota_window.total_seconds()
        self._fail_fast = fail_fast
        # NOTE(vlebedev): Constants end here.

        # NOTE(vlebedev): "Input" variables start here.
        #                 "Input" variables are those that are read from external source (Request) and must not
        #                  be altered by internal logic.
        self._now = now
        self._normal_limits = initial_normal_limits
        self._stats = initial_stats
        self._usage = initial_usage
        self._state = infer_lve_state(
            initial_stats,
            initial_normal_limits,
            initial_usage,
        )
        # NOTE(vlebedev): "Input" variables end here.

        # NOTE(vlebedev): "Output" variables start here.
        #                 "Output" variables are a channel that communicates internal state to external world.
        #                 Internal logic is responsible for mutating these vairables.
        self._history = initial_history
        # NOTE(vlebedev): "Output" variables end here.

        # NOTE(vlebedev): Utility stuff starts here.
        #                 These variables are not supposed to rewritten by internal logic.
        def apply_lve_settings_wrapper(cmd) -> None:
            if isinstance(cmd, self.Burst):
                limits = calc_bursted_limits(self._normal_limits, bursting_multipliers)
            elif isinstance(cmd, self.Unburst):
                limits = self._normal_limits
            else:
                raise ValueError(f'Unexpected command: {cmd}')

            apply_lve_settings(self._lve_id, limits)
        self._apply_lve_settings = apply_lve_settings_wrapper

        root_handler_gen = self._create_root_handler_gen()

        def step(request):
            response = root_handler_gen.send(request)
            if isinstance(response, Exception):
                raise response
            return response
        self.step = step

        self._on_state_changed = Emitter()

    @property
    def lve_id(self) -> LveId:
        return self._lve_id

    @property
    def now(self) -> Timestamp:
        return self._now

    @property
    def state(self) -> LveState:
        return self._state

    @property
    def is_bursted(self) -> bool:
        return self._state in {LveState.BURSTED, LveState.OVERUSING}

    @property
    def is_unbursted(self) -> bool:
        return self._state == LveState.UNBURSTED

    @property
    def is_overusing(self) -> bool:
        return self._state == LveState.OVERUSING

    @property
    def history_contains_overusing(self) -> bool:
        return self._history.contains_overusing

    @property
    def history(self) -> LveHistory:
        return self._history

    def check_quota_exceeded(self, now: float) -> bool:
        return self._history.get_overusing_duration(now) >= self._quota_sec

    @property
    def on_state_changed(self) -> Signal:
        return self._on_state_changed

    def trim_history(self, now: Timestamp) -> None:
        cutoff = Timestamp(int(now - self._quota_window_sec))
        self._history = self._history.trim(cutoff)

    @bootstrap_gen
    def _create_root_handler_gen(self):
        # NOTE(vlebedev): This coroutine is solely responsible for modifying "input" variables.
        #                 No mutations of "input" variables must happen outside of it.

        state_to_handler_factory = {
            LveState.EXISTED: self._create_existed_state_handler_gen,
            LveState.UNBURSTED: self._create_unbursted_state_handler_gen,
            LveState.BURSTED: self._create_bursted_state_handler_gen,
            LveState.OVERUSING: self._create_overusing_state_handler_gen,
        }

        prev_state = next_state = self._state
        subhandler = state_to_handler_factory[next_state](next_state)
        request, response = None, None
        while True:
            try:
                request, response = (yield response), None

                # NOTE(vlebedev): "Input" variables mutations start here.
                #                 It's important that it happens before invoking state handler
                #                 so that state handler sees updated state.
                assert request.now >= self._now, "Time is not expected to go backward!"
                self._now = request.now
                if isinstance(request, self.UpdateReadings):
                    if request.normal_limits is not None:
                        self._normal_limits = request.normal_limits
                    if request.stats is not None:
                        self._stats = request.stats
                    if request.usage is not None:
                        self._usage = request.usage
                    next_state = infer_lve_state(
                        self._stats,
                        self._normal_limits,
                        self._usage,
                    )
                elif isinstance(request, self.Disappered):
                    next_state = LveState.EXISTED
                self._state = next_state
                # NOTE(vlebedev): "Input" variables mutations end here.

                # TODO(vlebedev): Refactor subhandler management - it turned out to be quite messy.
                if inspect.getgeneratorstate(subhandler) != inspect.GEN_CLOSED:
                    subhandler_exc = None
                    if isinstance(request, self.Disappered):
                        subhandler_exc = request
                    elif next_state != prev_state:
                        subhandler_exc = self._StateSwitched()

                    if subhandler_exc is not None:
                        try:
                            with contextlib.suppress(StopIteration, type(subhandler_exc)):
                                subhandler.throw(subhandler_exc)
                        except (StopIteration, type(subhandler_exc)):
                            logger.debug(
                                'LVE "%s": state subhandler for "%s" state terminated',
                                self._lve_id, next_state
                            )
                        except Exception:
                            if self._fail_fast:
                                raise
                            logger.exception(
                                'LVE "%s": state subhandler for "%s" state unexpectedly failed during termination!',
                                self._lve_id, next_state
                            )
                        else:
                            try:
                                subhandler.close()
                            except Exception:
                                if self._fail_fast:
                                    raise
                                logger.exception(
                                    'LVE "%s": state subhandler for "%s" state did not terminate properly '
                                    'and failed during force-close!',
                                    self._lve_id, next_state
                                )

                if inspect.getgeneratorstate(subhandler) == inspect.GEN_CLOSED and next_state != prev_state:
                    subhandler = state_to_handler_factory[next_state](next_state)
                    logger.debug('LVE "%s": state subhandler for "%s" state created', self._lve_id, next_state)

                if inspect.getgeneratorstate(subhandler) != inspect.GEN_CLOSED:
                    try:
                        response = subhandler.send(request)
                    except StopIteration:
                        logger.debug('LVE "%s": state subhandler for "%s" state finished', self._lve_id, next_state)
                    except Exception:
                        if self._fail_fast:
                            raise
                        logger.exception(
                            'LVE "%s": state subhandler for "%s" state failed while handling "%s" request!',
                            self._lve_id, next_state, request,
                        )

                # NOTE(vlebedev): Trigger signal handlers only after all internal state has been updated.
                if next_state != prev_state:
                    logger.debug(
                        'LVE "%s": transitioned from "%s" to "%s" state',
                        self._lve_id, prev_state, next_state,
                    )
                    try:
                        self._on_state_changed(prev_state, next_state)
                    except Exception:
                        if self._fail_fast:
                            raise
                        logger.exception('LVE "%s": some "on_state_changed" listeners failed!', self._lve_id)

                prev_state = next_state
            except Exception as e:
                if self._fail_fast:
                    raise e
                logger.critical('LVE "%s": top handler unexpectadly failed!', self._lve_id, exc_info=e)

    @bootstrap_gen
    def _create_unbursted_state_handler_gen(self, prev_state: LveState):
        request, response = None, None
        try:
            while True:
                request, response = (yield response), None

                if isinstance(request, self.Unburst):
                    response = InvalidStateError('Already unbursted!')
                elif isinstance(request, self.Burst):
                    try:
                        self._apply_lve_settings(request)
                    except Exception as e:
                        response = e
                    else:
                        break
        except self._StateSwitched:
            if self._state == LveState.BURSTED:
                logger.warning(
                    'LVE "%s": was set to bursted externally, without internal request from adjuster',
                    self._lve_id,
                )
            raise

        logger.debug('LVE "%s": waiting for bursting enabled confirmation', self._lve_id)

        try:
            yield from self._wait_for_confirmation_from_readings({LveState.BURSTED, LveState.OVERUSING})
        except self.Disappered:
            logger.debug(
                'LVE "%s": disappeared while waiting for bursting enabled or overusing started confirmation',
                self._lve_id,
            )
            raise
        except self._StateSwitched:
            logger.warning(
                'LVE "%s": was switched to "%s" while waiting for bursting enabled or overusing started confirmation',
                self._lve_id, self._state,
            )
            raise

        assert self._state in {LveState.BURSTED, LveState.OVERUSING}

        logger.debug(
            'LVE "%s": confirmed from readings that bursting switched to enabled%s',
            self._lve_id,
            " and overusing has started" if self._state == LveState.OVERUSING else "",
        )

    @bootstrap_gen
    def _create_bursted_state_handler_gen(self, prev_state: LveState):
        request, response = None, None
        try:
            while True:
                request, response = (yield response), None

                if isinstance(request, self.Unburst):
                    try:
                        self._apply_lve_settings(request)
                    except Exception as e:
                        response = e
                    else:
                        break
                elif isinstance(request, self.Burst):
                    response = InvalidStateError('Already bursted!')
        except self._StateSwitched:
            if self._state == LveState.UNBURSTED:
                logger.warning(
                    'LVE "%s": was set to unbursted externally, without internal request from adjuster',
                    self._lve_id,
                )
            raise

        logger.debug('LVE "%s": waiting for bursting disabled confirmation', self._lve_id)

        try:
            yield from self._wait_for_confirmation_from_readings({LveState.UNBURSTED})
        except self.Disappered:
            logger.debug('LVE "%s": disappeared while waiting for bursting disabled confirmation', self._lve_id)
            raise
        except self._StateSwitched:
            logger.warning(
                'LVE "%s": was switched to "%s" while waiting for bursting disabled confirmation',
                self._lve_id, self._state,
            )
            raise

        logger.debug('LVE "%s": confirmed from readings that bursting switched to disabled', self._lve_id)

    @bootstrap_gen
    def _create_overusing_state_handler_gen(self, prev_state: LveState):
        # TODO(vlebedev): Do not use absolute time but relative to the latest event with `now`.
        self._history = self._history.append(self._now, BurstingEventType.STARTED)

        request, response = None, None
        try:
            try:
                while True:
                    request, response = (yield response), None

                    if isinstance(request, self.Unburst):
                        try:
                            self._apply_lve_settings(request)
                        except Exception as e:
                            response = e
                        else:
                            break
                    elif isinstance(request, self.Burst):
                        response = InvalidStateError('LVE is overusing i.e. is also already bursted!')
            except self._StateSwitched:
                if self._state == LveState.UNBURSTED:
                    logger.warning(
                        'LVE "%s": was set to unbursted externally, without internal request from adjuster',
                        self._lve_id,
                    )
                raise

            yield from self._wait_for_unbursted_confirmation_from_readings(prev_state)
        finally:
            self._history = self._history.append(self._now, BurstingEventType.STOPPED)

    @bootstrap_gen
    def _create_existed_state_handler_gen(self, prev_state: LveState):
        request, response = None, None
        try:
            while True:
                try:
                    request, response = (yield response), None
                except self.Disappered:
                    logger.warning(
                        'LVE "%s": disappeared again while waiting for LVE to re-appear',
                        self._lve_id,
                    )

                if isinstance(request, (self.Burst, self.Unburst)):
                    response = InvalidStateError('LVE does not exist!')
        except self._StateSwitched:
            logger.debug('LVE "%s": got back from the dead', self._lve_id)
            return

    def _wait_for_unbursted_confirmation_from_readings(self, prev_state: LveState):
        try:
            yield from self._wait_for_confirmation_from_readings({LveState.UNBURSTED})
        except self.Disappered:
            logger.debug('LVE "%s": disappeared while waiting for bursting disabled confirmation', self._lve_id)
            raise
        except self._StateSwitched:
            logger.warning(
                'LVE "%s": was switched to "%s" while waiting for bursting disabled confirmation',
                self._lve_id, self._state,
            )
            raise

        logger.debug('LVE "%s": confirmed from readings that bursting switched to disabled', self._lve_id)

    def _wait_for_confirmation_from_readings(self, expected_states: AbstractSet[LveState]):
        request, response = None, None
        try:
            while True:
                request, response = (yield response), None

                if isinstance(request, (self.Burst, self.Unburst)):
                    response = InvalidStateError('Waiting for confirmation from readings that limits have switched!')
        except self._StateSwitched:
            if self._state not in expected_states:
                raise


_state_to_code = {
    LveState.EXISTED: "X",
    LveState.UNBURSTED: "U",
    LveState.BURSTED: "B",
    LveState.OVERUSING: "O",
}

assert _state_to_code.keys() == set(LveState)


class LveStateSummary(NamedTuple):
    @classmethod
    def for_lve(cls, manager: LveStateManager) -> Self:
        return cls(manager.state, manager.history, manager.now)

    state: LveState
    history: LveHistory
    now: Timestamp

    def __str__(self) -> str:
        intervals = self.history.get_overusing_intervals(self.now)
        duration = self.history.get_overusing_duration(self.now)
        state_code = _state_to_code[self.state]
        return f"{state_code};[{intervals}~={round(duration)}s]"

Youez - 2016 - github.com/yon3zu
LinuXploit