def queuePerformance(arrivalRate, serviceRate, servers=1): """ Determine the average performance characteristics of a service queue using the M/M/c forumla arrivalRate = The rate of new customers in the queue serviceRate = The number of customers served in a time interval by all servers servers = The number of servers available (defaults to 1) returns a dictionary containing the following: emptyProbability = The probability the queue will be empty fullProbability = The probability the queue will be full averageLineLength = The average length of the queue averagePopulation = Average number of customers in the system as a whole (waiting and being served) averageWaitTime = Average time to wait until service averageTime = Average time in the system as a whole Doctests follow. To use these, run python interactively then type: import mmc import doctest doctest.testmod(mmc) Should return (0, 3) (no failures, 3 tests run) >>> queuePerformance(1, 1.1, 1) {'averageWaitTime': 9.0909090909090882, 'averageLineLength': 9.0909090909090882, 'averageTime': 9.9999999999999964, 'emptyProbability': 0.090909090909090939, 'averagePopulation': 9.9999999999999964, 'fullProbability': 0.082644628099173584} >>> queuePerformance(5, 6.1, 5) {'averageWaitTime': 6.3714079566131468e-05, 'averageLineLength': 0.00031857039783065737, 'averageTime': 0.16399814030907434, 'emptyProbability': 0.44055114515780114, 'averagePopulation': 0.81999070154537168, 'fullProbability': 0.0013583632864877702} >>> queuePerformance(3.4,4.2,3) {'averageWaitTime': 0.0058284681698965819, 'averageLineLength': 0.019816791777648377, 'averageTime': 0.24392370626513465, 'emptyProbability': 0.44281457712583089, 'averagePopulation': 0.82934060130145781, 'fullProbability': 0.039152503642860853} """ # Input validation arrivalRate = float(arrivalRate) serviceRate = float(serviceRate) throughput = arrivalRate/serviceRate if arrivalRate >= serviceRate: raise ValueError("Arrival rate must be lower than Service rate, otherwise the result is an infinite wait") if servers < 1: raise ValueError("Servers must be greater than 1, otherwise the result is an infinite wait (or other strangeness)") """ Rather than a for loop, I have used a list comprehension with a sum. I did this because it's fun, but it is functionally equivalent to: serviceRateSum = 0.0 for server in range(servers): serviceRateSum += ((throughput ** server) / factorial(server)) """ serviceRateSum = sum([((throughput ** server) / factorial(server)) for server in range(servers)]) # I really don't understand this number pwait = (throughput ** servers) / (factorial(servers) * (1 - throughput/servers)) # Work out the queue characteristics emptyProbability = 1.0/(serviceRateSum + pwait) fullProbability = emptyProbability * throughput**servers / factorial(servers) averageLineLength = fullProbability * (throughput / servers) / ((1-throughput / servers) ** 2) averagePopulation = averageLineLength + throughput averageWaitTime = averageLineLength / arrivalRate averageTime = averagePopulation / arrivalRate return dict(emptyProbability=emptyProbability, fullProbability=fullProbability, averageLineLength=averageLineLength, averagePopulation=averagePopulation, averageWaitTime=averageWaitTime, averageTime=averageTime) def mmc(lam, mu, c=1): """ Interface wrapper for queuePerformance """ try: result = queuePerformance(lam, mu, c) except ValueError: return dict() return dict( lam=lam, mu=mu, rho=lam/mu, c=c, L=result['averagePopulation'], W=result['averageTime'], Lq=result['averageLineLength'], Wq=result['averageWaitTime'], P0=result['emptyProbability'], Pc=result['fullProbability'] )