本文主要是介绍project: queue simulation，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
标签：python、ipynb、queue simulation
Submit your assignment as a Jupyter notebook file called yourmacid_project.ipynb, where yourmacid is replaced with your MacID, the user name in your McMaster email address. (For example, since my McMaster email address is bolker@mcmaster.ca, I would submit a file called bolker_project.ipynb.) Upload this file to the appropriate dropbox on Avenue to Learn.
To complete your assignment, edit the function definitions in this template, deleting the raise NotImplementedError statements in the code; then rename the file appropriately, as above. (Don't forget to use appropriate return statements in your functions!) Do not alter any other part of this template.
While developing and testing your code, it might be helpful to use print statements, but please remove them before submitting your solution. The template contains several tests for your code.
Any file submitted for grading that does not conform to the specifications above will lead to a grade of zero on the assignment.
Before submitting your solution, you should make sure that it runs properly. To do this, you can open it in the Jupyter notebook server that comes installed with Anaconda, or you can use the McMaster Jupyter server, by going to the website https://mcmaster.syzygy.ca/. (Try running Kernel > Restart & Run all from the menu to run all of your code from scratch and make sure nothing is left out.) Feel free to use Spyder, or some other IDE, to develop your code before entering it into the Jupyter notebook that you submit.
Your grade for each question will depend on whether or not your code correctly handles not just the test cases provided in the template, but other test cases that will be run on your solutions.
Do not leave this assignment until the last minute; last-minute computer/internet/Avenue issues on your part are your problem, not ours ...
Late assignments will be penalized 25% per day.
All work submitted for grading must be your own. You may discuss homework problems and related material with other students, but you must not submit work copied from others or from the internet. If you use any internet resources for help, you must list them as comments (##) in your answer.
In this project you will write Python code to simulate a simple queue. A queue is any system where one or more servers process jobs: this could be bank clients (jobs) at an automated teller machine (servers), or shoppers (jobs) at checkout counters (servers) in a supermarket.
The usual mechanics of a queue are that new jobs arrive at the queue at random times, typically exponentially distributed: we can use the numpy.random.exponential(a) function to choose an exponentially distributed (pseudo)random variable with mean 𝑎a. Each job also has a processing time, which we will also choose from an exponential distribution with mean 𝑤w.
- Write a function queue_step(queue, atime, wtime) that simulates one step in the dynamics of the queue from a starting time time until the arrival of the next job (with processing time wtime) at time atime
- queue is a list (possibly empty) of jobs. Each element in the list is a float representing the processing time for that job, when it arrives at the server. The first element queue[0] (if it exists) is the "current job", or the job currently being processed; the second element queue[1] (if it exists) is the "first waiting job"; queue[-1] (if it exists) is the most recently arrived job.
- atime is the time until the next job arrives
- wtime is the processing time the job will take when it arrives at the head of the queue
-
Your function should return a tuple (proc_n, proc_t) that gives the number of jobs completed (integer) and the processing time (float) that has been used while waiting for the next job to arrive.
The basic machinery of this function should work as follows:
- append the job to the end of the queue (i.e., append its processing time to the queue list)
- if the queue is of length 1, it was empty before the new job arrived; return (0,0.0) (no jobs have been processed while waiting for the new job to arrive, and no time was spent processing)
- if the queue was not empty:
- if the current job's remaining processing time is longer than the time it took for the new job to arrive:
- subtract the arrival time from the current job's processing time
- return (0,atime) (no jobs have finished in this step, and the entire waiting time was spent processing)
-
- otherwise:
- initialize proc_n and proc_t to zero
- while there is more than one job in the queue (including the current job) and the remaining processing time of the current job is less than the remaining arrival time:
- subtract the processing time of the current job from the arrival time
- add the processing time of the current job to proc_t
- add one to the number of jobs completed;
- move the processing time of the first waiting job (the second job in the queue) into the processing slot (first in the queue) by deleting the current job from the beginning of the list (using del or .pop())
-
- return the number of jobs completed and the processing time for this step
-
-
import numpy as np
import numpy.random as npr
import matplotlib.pyplot as plt
import sys
print(sys.version)
def queue_step(queue, atime, wtime):
#答案代码-联系作者
return proc_n, proc_t
def run_test(q_in,atime,wtime,q_out,proc_n,proc_t,name=None):
"""testing function
q_in, atime, wtime: input queue, arrival and processing time for the arriving job
q_out, proc_n, proc_t: *expected* values of the ending queue, number of jobs processed,
processing time completed
"""
proc_n_obs, proc_t_obs = queue_step(q_in,atime,wtime)
## q_in is modified by queue_step (now equal to *ending* state of the queue)
if q_in != q_out:
raise Exception(f"expected queue out={q_out}, obs value={q_in}")
if proc_n_obs!= proc_n:
raise Exception(f"expected jobs finished={proc_n}, obs value={proc_n_obs}")
if proc_t_obs!= proc_t:
raise Exception(f"expected processing time={proc_t}, obs value={proc_t_obs}")
## we only get here if nothing went wrong ...
if not name is None:
print(f"{name}: correct!")
return True
## empty queue; add atime 1, wtime 1; add to queue, no jobs finish, no processing done
run_test(q_in=[], atime=1,wtime=1,q_out=[1],proc_n=0,proc_t=0, name="empty_11")
## queue with 1 job; add atime 1, wtime 1; 1 job finishes, 1 unit of processing
run_test(q_in=[1],atime=1,wtime=1,q_out=[1],proc_n=1,proc_t=1, name="1job_11")
## queue with 2 jobs
run_test(q_in=[1,1],atime=1,wtime=1,q_out=[1,1],proc_n=1,proc_t=1,name="2jobs_11")
## queue with 2 jobs; arrival time =2
run_test(q_in=[1,1],atime=2,wtime=1,q_out=[1],proc_n=2,proc_t=2,name="2jobs_21")
## queue with 2 jobs; arrival time =0
run_test(q_in=[1,1],atime=0,wtime=1,q_out=[1,1,1],proc_n=0,proc_t=0, name="2jobs_01")
## queue with 2 jobs; arrival time =3 (still only 2 jobs processed)
run_test(q_in=[1,1],atime=3,wtime=1,q_out=[1],proc_n=2,proc_t=2,name="2jobs_31")
Next, write a function queue_run(queue, atimes, wtimes) that starts at time 0 with a specified queue and simulates the arrival of a number of jobs with specified arrival times ( atimes , a list of numeric values) and processing times ( wtimes , also a list of numeric values). It should raise a ValueError if the lengths of atimes and wtimes are different. It should return a tuple (proc_n,proc_t,tot_t) containing (1) the total number of jobs that have been processed at that point; (2) the total amount of time spent processing jobs; and (3) the final time when the last job arrives at the queue (i.e., the sum of the arrival times).
def queue_run(queue,atimes,wtimes):
#答案代码联系作者
return proc_n, proc_t, tot_t
def run_test2(q_in,atimes,wtimes,answer,name=None):
if not name is None:
print(f"test: {name}", end=": ")
a0 = queue_run(q_in,atimes,wtimes)
if a0==answer:
print("correct!")
else:
print(f"your answer is {q0}, it should be {answer}")
return None
## 4 jobs arriving at an empty queue: num jobs processed=3, processing time=3, total time=4
run_test2([],atimes=[1,1,1,1],wtimes=[1,1,1,1],answer=(3,3,4), name="basic queue_run test")
## test that queue_run correctly raises a ValueError when atimes and wtimes lengths differ
try:
queue_run([],atimes=[],wtimes=[1])
except ValueError:
print("queue_run correctly catches unequal length error!")
except Exception:
print("queue_run raises wrong kind of error!")
## 4 jobs arriving at an empty queue with different arrival times
run_test2([],atimes=[1,3,1,2],wtimes=[2,2,2,2],answer=(2,4,7), name="queue_run test 2")
Write a function queue_exp_run(n, a, w) that builds on queue_run() to start with an empty queue at time 0 and simulate the arrival of n jobs, each with a different exponentially distributed pseudo-random arrival time a and a different exponentially distributed pseudo-random processing time w. The function should return the same tuple as queue_run() (total jobs processed, total processing time, total overall time)
def queue_exp_run(n, a, w):
#答案联系作者
return queue_run([],atimes,wtimes)
npr.seed(101)
res = queue_exp_run(1000,a=2,w=1)
print(res) ## should be APPROX 1000, 1000, 2000 - at least within 10%
np.isclose(res,np.array([1000,1000,2000]),rtol=0.10).all()
Write a function queue_summary(vals) that takes a tuple (vals) like the one output by queue_exp_run() and queue_run() and returns a tuple giving the throughput (jobs processed divided by total time) and utilization (processing time divided by total time)
def queue_summary(vals):
#答案联系作者
return throughput,utilization
qvals = queue_run([],[1,1,1,1],[1,1,1,1])
## 3 jobs processed in 4 time units. 3/4 of the total time is spent processing
queue_summary(qvals) == (0.75, 0.75)
Use all the functions you've written so far to create a 3D numpy array ans with two slices, 5 rows, and 2 columns that stores the results from exponential queue simulations with n=1000 steps. The answers stored in the first slice should use an expected processing time of 𝑤=1w=1, those in the second slice should use 𝑤=0.5w=0.5. The rows of each slice use exponential arrival times of 𝑎=a=2, 3, 4, 5, and 6. The first column of each slice holds the throughput; the second column of each slice holds the utilization.
a_vec = np.arange(2,7)
w_vec = (1,0.5)
#答案联系作者
print(ans)
atype = type(ans).__name__
if atype=="ndarray":
print("answer is correct type!")
else:
print("answer is wrong type (should be 'ndarray', actually {atype})")
if ans.shape==(2,5,2):
print("answer is correct shape!")
else:
print("answer is wrong shape (should be (2,5,2), actually {ans.shape})")
theor_vals = np.zeros((2,5,2))
theor_vals[0,:,0] = 1/a_vec ## throughput, w=1
theor_vals[0,:,1] = 1/a_vec ## utilization, w=1
theor_vals[1,:,0] = 1/a_vec ## throughput, w=0.5
theor_vals[1,:,1] = 0.5/a_vec ## utilization, w=0.5
print(theor_vals)
## are all of the simulated answers within 15% of theoretical values?
if np.isclose(ans,theor_vals,rtol=0.15).all():
print("all values are close!")
else:
print("some differences are larger than expected! relative differences:")
print(round((ans-theor_vals)/theor_vals,2))
fig, ax = plt.subplots(1,2)
ax[0].plot(a_vec,ans[0,:,0],label="w=1")
ax[0].plot(a_vec,ans[1,:,0],label="w=0.5")
ax[0].set_title("throughput")
ax[0].set_xlabel("arrival time")
ax[1].plot(a_vec,ans[0,:,1],label="w=1")
ax[1].plot(a_vec,ans[1,:,1],label="w=0.5")
ax[1].set_title("utilization")
ax[1].set_xlabel("arrival time")
ax[1].legend();
最后一步图示： 
联系邮箱535798943@qq.com
这篇关于project: queue simulation的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
