待ち行列(M/M/1)の再現 – 時刻制御

1 考え方
2 クラス構成
3 時刻制御の実装
- 3.1 考え方
- 3.2 テスト実行
4 実行結果

考え方

時刻制御(time driven)による待ち行列の計算の考え方は次の通り。

時刻t₀から始めて、以下時間間隔Δtずつ増やしながら計算を進めていく
ある時刻で一様乱数の値がλΔtより小さければ到着が発生
- 新たにトランザクションを生成し、到着時刻を記録し、システムに投入する
ある時刻で到着が発生せず、システムにトランザクションがある場合、一様乱数の値がμΔｔより小さければサービスが終了
- サービス中のトランザクションのサービスを終了させ、終了時刻を記録する
- このとき待ち行列にトランザクションがあれば、先頭のトランザクションのサービスを開始し、サービス開始時刻を記録する
予め設定していた時刻(またはトランザクション数)に達したら終了

queue-mm1-time-driven-fig1

たとえば上の図でトランザクション②に注目すると、

時刻t₄でトランザクションが発生
t₇でトランザクション①のサービスが終了し、②のサービスが開始
t₈でサービス終了

イベント制御のケースではRを用いたが、今回はPythonを使う。なおM/M/1待ち行列の解析的アプローチはこちら。

待ち行列システムのクラスをつくり、システムの外からはトランザクションの到着、サービス終了の操作を行うだけで、内部的に処理が進むようにする。

クラス構成

待ち行列システムを表すQueueSystemクラス、システムのサービス窓口を表すServiceクラス、システム中の1つのトランザクションを表すTransactionクラスから構成される。システム中の待ち行列はインスタンス変数のリストとして持つ。

M/M/1型の行列システムとして、QueueSystemクラスは1つのServiceオブジェクトと1つの待ち行列リストをメンバーに持つ。

トランザクションの到着の指示が出るとTransactionオブジェクトが生成され、システムに登録される。

その一方で実行中のサービス終了の指示が出るとサービスを受けているトランザクションが解放される。

各Transactionオブジェクトに対して、システムの到着、サービスの開始、サービスの終了の時刻が自動的に登録される。

Transactionクラス

識別子としての整数値のindexのほか、到着・サービス開始・サービス終了の時刻、到着時のシステム内トランザクション数と待ち行列長をメンバーとして持つ
- これらのメンバーに対しては直接アクセスして参照・代入する
このクラスのオブジェクトの文字列表現を__str__メソッドで実装している

class Transaction:
    def __init__(self, index, arrive_time):
        self.index = index
        self.arrive_time = arrive_time
        self.start_time = 0
        self.end_time = 0
        self.transaction_number_when_arrived = 0
        self.queue_length_when_arrived = 0

    def __str__(self):
        return "<{}:{},{},{}>".format(
            self.index, self.arrive_time, self.start_time, self.end_time)

class Transaction:

def __init__(self, index, arrive_time):

self.index = index

self.arrive_time = arrive_time

self.start_time = 0

self.end_time = 0

self.transaction_number_when_arrived = 0

self.queue_length_when_arrived = 0

def __str__(self):

return "<{}:{},{},{}>".format(

self.index, self.arrive_time, self.start_time, self.end_time)

Serviceクラス

privateなメンバ_transactionを持ち、これはサービス中のトランザクションを保持する
- サービス中のトランザクションがない場合はNone
サービス窓口が空いているか(vacant)、サービス中か(occupied)を判定するメソッドを持つ
サービス中のトランサクションへの参照を返すtransaction_in_service()メソッドを持つ
サービス開始時の処理を行うstart()メソッドを持つ
- サービスを開始するTransactionオブジェクトとサービス開始時刻を指定する
サービス終了時の処理を行うend()メソッドを持つ
- サービス終了時刻を指定し、サービス中のTransactionオブジェクトにその時刻を設定し、オブジェクトへの参照を返す

class Service:
    def __init__(self):
        self._transaction = None

    def is_vacant(self):
        return self._transaction is None

    def is_occupied(self):
        return self._transaction is not None

    def transaction_in_service(self):
        return self._transaction

    def start(self, transaction, time):
        transaction.start_time = time
        self._transaction = transaction

    def end(self, time):
        transaction = self._transaction
        transaction.end_time = time
        self._transaction = None
        return transaction

class Service:

def __init__(self):

self._transaction = None

def is_vacant(self):

return self._transaction is None

def is_occupied(self):

return self._transaction is not None

def transaction_in_service(self):

return self._transaction

def start(self, transaction, time):

transaction.start_time = time

self._transaction = transaction

def end(self, time):

transaction = self._transaction

transaction.end_time = time

self._transaction = None

return transaction

QueueSystemクラス

メンバとなるインスタンス変数は4つ

待ち行列リスト_queue
サービス窓口オブジェクト_service
到着済みのトランザクションを登録する_arrived_transaction_list
サービス完了済みのトランザクションを登録する_departed_transaction_list

getter系のメソッドは

システム中にあるトランザクション数を返すtransaction_number_in_system()
待ち行列長を返すqueue_length()
システム中にあるトランザクションへの参照を返すtransaction_in_service()
待ち行列中にあるトランザクションへの参照を返すtransaction_in_queue()
到着済みのトランザクション数を返すarrived_transaction_number()
完了済みのトランザクション数を返すdeparted_transaction_number()
到着済みのトランザクションのn番目を返すarrived_transaction()
完了済みのトランザクションのn番目を返すdeparted_transaction()

システムを操作するメソッドは

システムに到着したトランザクションを登録するarrive()
サービスが終了したトランザクションをシステムから取り出すdepart()

システムの状態を表示するメソッドとして以下の2つを実装している。

システムの現在の状態を表示するdisplay_current_status()
サービスを完了したトランザクション群の各パラメータを表示するdisplay_transaction_summary()

class QueueSystem:
    def __init__(self):
        self._queue = []
        self._service = Service()
        self._arrived_transaction_list = []
        self._departed_transaction_list = []

    def transaction_number_in_system(self):
        return (len(self._queue)
                + (1 if self._service.is_occupied() else 0))

    def queue_length(self):
        return len(self._queue)

    def transaction_in_service(self):
        return self._service.transaction_in_service()

    def transaction_in_queue(self, n):
        return self._queue[n]

    def arrived_transaction_number(self):
        return len(self._arrived_transaction_list)

    def departed_transaction_number(self):
        return len(self._departed_transaction_list)

    def arrived_transaction(self, n):
        return self._arrived_transaction_list[n]

    def departed_transaction(self, n):
        return self._departed_transaction_list[n]

    def arrive(self, transaction):
        self._arrived_transaction_list.append(transaction)
        transaction.transaction_number_when_arrived = \
            self.transaction_number_in_system()
        transaction.queue_length_when_arrived = self.queue_length()
        if self._service.is_vacant():
            self._service.start(transaction, transaction.arrive_time)
        else:
            self._queue.append(transaction)

    def depart(self, time):
        departing_transaction = self._service.end(time)
        self._departed_transaction_list.append(departing_transaction)
        if len(self._queue) > 0:
            waiting_transaction = self._queue.pop(0)
            self._service.start(waiting_transaction, time)

    def display_current_status(self):
        print()
        print("--- Queue System Status ---")
        print("arrived :{0:5d}".format(self.arrived_transaction_number()))
        print("departed:{0:5d}".format(self.departed_transaction_number()))
        print("[Service]")
        if self.transaction_number_in_system() > 0:
            print(' ' + str(self.transaction_in_service()))
        print("[Queue]")
        if self.queue_length() > 0:
            for n in range(0, self.queue_length()):
                print(self.transaction_in_queue(n))

    def display_transaction_summary(self):
        print()
        print("---------------------------- " \
            "All Transaction Summary " \
            "---------------------------")
        print(
            "     index    arrive     start       end" \
            "   trs num   que len wait time  trs time")
        for tr in self._departed_transaction_list:
            print("{0:10d}{1:10.2f}{2:10.2f}{3:10.2f}" \
                "{4:10d}{5:10d}{5:10.2f}{6:10.2f}".format(
                tr.index, tr.arrive_time, tr.start_time, tr.end_time,
                tr.transaction_number_when_arrived,
                tr.queue_length_when_arrived,
                tr.start_time - tr.arrive_time, tr.end_time - tr.arrive_time))

class QueueSystem:

def __init__(self):

self._queue = []

self._service = Service()

self._arrived_transaction_list = []

self._departed_transaction_list = []

def transaction_number_in_system(self):

return (len(self._queue)

+ (1 if self._service.is_occupied() else 0))

def queue_length(self):

return len(self._queue)

def transaction_in_service(self):

return self._service.transaction_in_service()

def transaction_in_queue(self, n):

return self._queue[n]

def arrived_transaction_number(self):

return len(self._arrived_transaction_list)

def departed_transaction_number(self):

return len(self._departed_transaction_list)

def arrived_transaction(self, n):

return self._arrived_transaction_list[n]

def departed_transaction(self, n):

return self._departed_transaction_list[n]

def arrive(self, transaction):

self._arrived_transaction_list.append(transaction)

transaction.transaction_number_when_arrived = \

self.transaction_number_in_system()

transaction.queue_length_when_arrived = self.queue_length()

if self._service.is_vacant():

self._service.start(transaction, transaction.arrive_time)

else:

self._queue.append(transaction)

def depart(self, time):

departing_transaction = self._service.end(time)

self._departed_transaction_list.append(departing_transaction)

if len(self._queue) > 0:

waiting_transaction = self._queue.pop(0)

self._service.start(waiting_transaction, time)

def display_current_status(self):

print()

print("--- Queue System Status ---")

print("arrived :{0:5d}".format(self.arrived_transaction_number()))

print("departed:{0:5d}".format(self.departed_transaction_number()))

print("[Service]")

if self.transaction_number_in_system() > 0:

print(' ' + str(self.transaction_in_service()))

print("[Queue]")

if self.queue_length() > 0:

for n in range(0, self.queue_length()):

print(self.transaction_in_queue(n))

def display_transaction_summary(self):

print()

print("---------------------------- " \

"All Transaction Summary " \

"---------------------------")

print(

" index arrive start end" \

" trs num que len wait time trs time")

for tr in self._departed_transaction_list:

print("{0:10d}{1:10.2f}{2:10.2f}{3:10.2f}" \

"{4:10d}{5:10d}{5:10.2f}{6:10.2f}".format(

tr.index, tr.arrive_time, tr.start_time, tr.end_time,

tr.transaction_number_when_arrived,

tr.queue_length_when_arrived,

tr.start_time - tr.arrive_time, tr.end_time - tr.arrive_time))

動作テスト

上記のクラス群を使って以下を実行。規則正しく刻まれた時刻に4つのトランザクションが到着し、その後規則正しくサービスが終了していく様子と、その結果を表示させている。

trs = []
for n in range(1, 10):
    trs.append(Transaction(n, n*10))

system = QueueSystem()
system.display_current_status()

system.arrive(trs[0])
system.display_current_status()

system.arrive(trs[1])
system.display_current_status()

system.arrive(trs[2])
system.display_current_status()

system.arrive(trs[3])
system.display_current_status()

system.depart(45)
system.display_current_status()

system.depart(55)
system.display_current_status()

system.depart(65)
system.display_current_status()

system.depart(75)
system.display_current_status()

print()
system.display_transaction_summary()

print()
print("-- arrived --")
na = system.arrived_transaction_number()
for i in range(0, na):
    print(system.arrived_transaction(i))

print()
print("-- departed --")
na = system.departed_transaction_number()
for i in range(0, na):
    print(system.departed_transaction(i))

trs = []

for n in range(1, 10):

trs.append(Transaction(n, n*10))

system = QueueSystem()