背景
どのくらいの待ち時間が有るのか計算する必要がでてきたので、試しに
１列の待ち行列のときにどういう計算になるのか調べてみました。
まずどうやって計算するか
計算方法は以下のサイトが
サルでもわかる待ち行列
とてもわかり易かったです。
待ち行列の自体の説明は上のサイトを参考にして、とりあえず
平均利用率
\rho \text{(平均利用率)}
が与えられれば、すぐに平均待ち時間
T_w = \frac{\rho}{1-\rho}T_s
が計算できるようです。
ここでTsは
T_s \text{(平均サービス時間)}
という量で、これは平均で１個処理するのに何秒かかるかということです。
順番が逆になってしまいましたが、
平均利用率ρは
\rho=\frac{\lambda}{\mu}
という定義で、ここにでてくるλとμは逆数が平均時間になっています。
それぞれ
\lambda=\frac{1}{T_a \text{(平均到着時間)}}
\mu=\frac{1}{T_s \text{(平均サービス時間)}}
という関係になっています。
というわけで上の平均到着時間と平均サービス時間がわかれば
待ち行列の待ち時間Twがすぐに計算できるため
簡単にpythonで計算できるモジュールを作ってみました。
待ち時間と待ち時間を計算する簡単なモジュール
# coding: UTF-8
# Copyright 2019 Hideto Manjo.
#
# Licensed under the MIT License
from collections import deque
import time
class Queueing:
"""Queueing.
This module is used to calculate latency and turnaround time
based on queuing theory when processing queues continuously.
Calling arrival and service will record the time.The required
parameters can be calculated by calling the respective method.
"""
def __init__(self, maxlen=10):
"""Init."""
self._arrivals = deque(maxlen=maxlen)
self._services = deque(maxlen=maxlen)
self._last_arrival = time.time()
self._last_service = time.time()
@classmethod
def __mean(cls, time_list):
return sum(time_list) / len(time_list)
@classmethod
def __record(cls, time_list, last_time):
"""Record the lap time and return the current time."""
now = time.time()
time_list.append(now - last_time)
return now
def clear_history(self):
"""Clear history only."""
self._arrivals.clear()
self._services.clear()
def clear(self):
"""Clear."""
self.clear_history()
self._last_arrival = time.time()
self._last_service = time.time()
def arrival(self):
"""Record arrival time."""
self._last_arrival = self.__record(self._arrivals, self._last_arrival)
return self._arrivals[-1]
def service(self):
"""Record service time."""
self._last_service = self.__record(self._services, self._last_service)
return self._services[-1]
def average_arrival_time(self):
"""Average arrival time (Ta)."""
return self.__mean(self._arrivals)
def average_service_time(self):
"""Average service time (Ts)."""
return self.__mean(self._services)
def average_arrival_rate(self):
"""Avarage arrival rate (lambda)."""
return 1.0 / self.__mean(self._arrivals)
def average_service_rate(self):
"""Avarage service rate (mu)."""
return 1.0 / self.__mean(self._services)
def average_utilization(self):
"""Avarage utilization."""
return self.average_service_time() / self.average_arrival_time()
def average_wait(self):
"""Avarage wait."""
rho = self.average_utilization()
if rho < 1.0:
return rho / (1.0 - rho)
return float('inf')
def average_wait_time(self):
"""Avarage wait time."""
return self.average_wait() * self.average_service_time()
def turnaround_time(self):
"""Turnaround time."""
return self.average_wait_time() + self.average_service_time()
if __name__ == "__main__":
import random
QUEUEING = Queueing()
#This means the avarage service time is fixed 0.5[sec].
QUEUEING.service()
time.sleep(1)
QUEUEING.service()
for i in range(100):
sleep_time = random.uniform(0, 2)
time.sleep(sleep_time)
QUEUEING.arrival()
print("-----------------------------")
print("Current arrival", sleep_time, "[sec]")
print("Ta", QUEUEING.average_arrival_time(), "[sec]")
print("Ts", QUEUEING.average_service_time(), "[sec]")
print("Rho", QUEUEING.average_utilization())
print("Average wait time", QUEUEING.average_wait_time(), "[sec]")
print("Turnaround time", QUEUEING.turnaround_time(), "[sec]")
動作確認用にメイン関数も書いています。
中身について
dequeの利用
後入れ先出しを効率よく使うためにdequeというのが有るようなので
listではなくdequeを使っています。
maxlenで指定するとappendすれば勝手に先頭は消えるそうです。
すごく便利ですね。
maxlenで指定した長さまで平均の時刻を記録しておき
必要な計算するときには読み出して計算しています。
使い方
使い方はキューが到着すればarrival()を呼び出し
サービスを開始するときにservice()を呼び出します。
それぞれの呼び出す間隔を記録していますので
あとはその記録を使って色々計算できる寸法です。
試しに実行
上のメイン関数を実行すると
まずservice()を２秒で２回呼び出しますので
平均サービスタイムは１秒となります。
それを固定したまま
１秒から2秒までランダムに到着arrival()を呼び出して
その後、計算結果を表示します。
ターンアラウンドタイムは自分の待ち時間を足したものとなっており
待ち時間を考慮した場合の応答時間を表しています。
まとめ
これで一応の計算はできそうです。
何かの参考になれば幸いです。
注意
まだ適当に書いてみたばっかりなので
どっかおかしいところが有るかも知れないです。