負荷（コンピューティング）

#!/usr/bin/env python3インポート時間import osfrom datetime import datetimefrom math import exp , logfrom dataclasses import dataclassLOGSYSPERIODS = [ log ( x ) for x in [ 60 , 300 , 900 ]]REFRESH_RATE = (( 5 ** 0.5 ) + 1 ) / 2 # リップケの黄金比周期= [ 10 , 30 , 60 , 120 , 300 , 900 , 1800 , 3600 ]COUNT_DISKWAIT = True # ディスク待機を負荷計算に含めるかどうか@dataclassクラスLoadEntry :平均: floatexp : floatdef initialize_loads ( now : int | float , lavgs : dict [ int , LoadEntry ]):システム負荷平均と瞬間負荷数の線形補間に基づいて負荷平均を初期化します。sys = getloadavg ()傾斜= (( sys [ 0 ] - now ) / 60 , ( sys [ 1 ] - sys [ 0 ]) / 240 , ( sys [ 2 ] - sys [ 1 ]) / 600 )期間が[ 10 , 30 , 60 , 120 , 300 , 900 , 1800 , 3600 ]の場合:exp_factor = exp ( - REFRESH_RATE / period )期間が60未満の場合：est_avg = now + slopes [ 0 ] * ( period - 60 )elif period < 300 :est_avg = sys [ 0 ] + slopes [ 1 ] * ( period - 300 )それ以外：est_avg = sys [ 1 ] + slopes [ 2 ] * ( period - 900 )lavgs [ period ] = LoadEntry ( avg = max ( est_avg , 0 ), exp = exp_factor )def update_loads ( lavgs : dict [ int , LoadEntry ], current_load : int | float ) -> None :for _ 、lavgs.items ( )のエントリ:entry.avg = entry.avg * entry.exp + current_load * ( 1 - entry.exp )​​​​​​​​if os.name == " posix " :uname = os . uname ()[ 0 ] .より低い（）getloadavg = os.getloadavg​​if uname == "linux" :def get_current_load () -> int :負荷= 0with open ( "/proc/stat" , "r" ) as f :for line in f :行が「procs_running」で始まる場合：load += int ( line . split ()[ 1 ]) # procs_running を読み込むload -= 1 # 自分自身のために1を引くelif line.startswith ( " procs_blocked " ) and COUNT_DISKWAIT :load += int ( line . split ()[ 1 ]) # procs_blocked を読み込むリターンロードそれ以外：PS_THREAD_OPTION = "-H" if os.uname ()[ 0 ] .lower () . endswith ( " bsd" ) else " -M "PS_DISK_WAIT = "U" if os.uname ( )[ 0 ] == "Darwin" else "D "PS_STATES = ( "R" + PS_DISK_WAIT ) if COUNT_DISKWAIT else "R"def get_current_load () -> int :with os.popen ( f " ps { PS_THREAD_OPTION } ax -o stat" , " r" ) as f :states = map ( f , lambda line : line . split ()[ - 1 ]) # 最後の列を取得します。macOS では必須です。return sum ( if state [ 0 ] in PS_STATES else 0 for state in states ) - 1​elif os.name == " nt " :# Windowsのパフォーマンスカウンターを使用してキューの長さを取得することが可能です。# 実際、マイクロソフトはCPU使用率に加えて、負荷の指標としてこれを推奨しています。# https://learn.microsoft.com/en-us/biztalk/technical-guides/using-the-performance-analysis-of-logs-pal-tool#processor-queue-length-analysis# これを使用すると、Unix システムと同様の負荷を得ることもできます。from pyperfmon import pyperfmonpm = pyperfmon.pyperfmon ( )​ncores = os.cpu_count ( )​get_counter = lambda x : pm . getCounter ( x )def get_current_load () -> float :戻る（# CPUを待機しているスレッド（実行中のスレッドではない）get_counter ( r "System\Processor Queue Length" )# CPUを使用しているスレッドのおおよその数+ get_counter ( r "Processor\_Total\% Processor Time" ) * ncores# ディスクI/Oを待機しているスレッド+ get_counter ( r "PhysicalDisk\_Total\Current Disk Queue Length" )COUNT_DISKWAITの場合それ以外の場合は0）def getloadavg () -> tuple [ float , float , float ]:# Windows 用のダミー実装load = get_current_load ()return ( load , load , load )def main ():lavgs : dict [ int , LoadEntry ] = {}current_load = get_current_load ()initialize_loads ( current_load , lavgs )見出し= [ "SYSTIME" , " CURR" ] + [ str ( x ) for x in PERIODS ]print ( " \t " .join ( heading ) )while True :# 再計算する前に必ず印刷するentries = [ f " { datetime . now () . strftime ( '%H:%M:%S' ) } { current_load : .4f } " ]entries += [ f " { entry . avg : .4f } " for entry in lavgs . values ()]print ( " \t " .join ( entries ) , flush = True )# スリープして待機します。これは、新しい行を更新するのにかかる時間が# 睡眠時間に比べてごくわずかです。そうでない場合は、# sleepuntil() 型の計算方法を使用する必要があります。time.sleep ( REFRESH_RATE )​​# 印刷current_load = get_current_load ()update_loads ( lavgs , current_load )if __name__ == "__main__" :主要（）