<s> import math <EOL> import numpy <EOL> import random <EOL> import Queue <EOL> from util import Job , TaskDistributions <EOL> MEDIAN_TASK_DURATION = <NUM_LIT:100> <EOL> NETWORK_DELAY = <NUM_LIT:0.5> <EOL> TASKS_PER_JOB = <NUM_LIT:100> <EOL> SLOTS_PER_WORKER = <NUM_LIT:4> <EOL> TOTAL_WORKERS = <NUM_LIT> <EOL> PROBE_RATIO = <NUM_LIT:2> <EOL> PER_TASK = True <EOL> def get_percentile ( N , percent , key = lambda x : x ) : <EOL> if not N : <EOL> return <NUM_LIT:0> <EOL> k = ( len ( N ) - <NUM_LIT:1> ) * percent <EOL> f = math . floor ( k ) <EOL> c = math . ceil ( k ) <EOL> if f == c : <EOL> return key ( N [ int ( k ) ] ) <EOL> d0 = key ( N [ int ( f ) ] ) * ( c - k ) <EOL> d1 = key ( N [ int ( c ) ] ) * ( k - f ) <EOL> return d0 + d1 <EOL> def plot_cdf ( values , filename ) : <EOL> values . sort ( ) <EOL> f = open ( filename , "<STR_LIT:w>" ) <EOL> for percent in range ( <NUM_LIT:100> ) : <EOL> fraction = percent / <NUM_LIT> <EOL> f . write ( "<STR_LIT>" % ( fraction , get_percentile ( values , fraction ) ) ) <EOL> f . close ( ) <EOL> class Event ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> raise NotImplementedError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def run ( self , current_time ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class JobArrival ( Event ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , simulation , interarrival_delay , task_distribution ) : <EOL> self . simulation = simulation <EOL> self . interarrival_delay = interarrival_delay <EOL> self . task_distribution = task_distribution <EOL> def run ( self , current_time ) : <EOL> job = Job ( TASKS_PER_JOB , current_time , self . task_distribution , MEDIAN_TASK_DURATION ) <EOL> probe_event = self . simulation . send_probes ( job , current_time ) <EOL> arrival_delay = random . expovariate ( <NUM_LIT:1.0> / self . interarrival_delay ) <EOL> job_arrival_event = ( current_time + arrival_delay , self ) <EOL> return [ probe_event , job_arrival_event ] <EOL> class ProbeEvent ( Event ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , simulation , job_id ) : <EOL> self . simulation = simulation <EOL> self . job_id = job_id <EOL> def run ( self , current_time ) : <EOL> task_arrival_events = [ ] <EOL> if PER_TASK : <EOL> task_arrival_events = self . simulation . assign_tasks_per_task ( current_time , self . job_id ) <EOL> else : <EOL> task_arrival_events = self . simulation . assign_tasks ( current_time , self . job_id ) <EOL> assert ( len ( task_arrival_events ) == self . simulation . jobs [ self . job_id ] . num_tasks ) <EOL> return task_arrival_events <EOL> class TaskArrival ( Event ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , worker , task_duration , job_id ) : <EOL> self . worker = worker <EOL> self . task_duration = task_duration <EOL> self . job_id = job_id <EOL> def run ( self , current_time ) : <EOL> return self . worker . add_task ( current_time , self . task_duration , self . job_id ) <EOL> class TaskEndEvent ( ) : <EOL> def __init__ ( self , worker ) : <EOL> self . worker = worker <EOL> def run ( self , current_time ) : <EOL> return self . worker . free_slot ( current_time ) <EOL> class Worker ( object ) : <EOL> def __init__ ( self , simulation , num_slots , id ) : <EOL> self . simulation = simulation <EOL> self . free_slots = num_slots <EOL> self . queued_tasks = Queue . Queue ( ) <EOL> self . num_queued_tasks = <NUM_LIT:0> <EOL> self . id = id <EOL> def queue_length ( self ) : <EOL> queue_length = self . num_queued_tasks <EOL> if self . free_slots > <NUM_LIT:0> : <EOL> assert self . num_queued_tasks == <NUM_LIT:0> <EOL> return - self . free_slots <EOL> return queue_length <EOL> def add_task ( self , current_time , task_duration , job_id ) : <EOL> self . queued_tasks . put ( ( task_duration , job_id ) ) <EOL> self . num_queued_tasks += <NUM_LIT:1> <EOL> return self . maybe_start_task ( current_time ) <EOL> def free_slot ( self , current_time ) : <EOL> """<STR_LIT>""" <EOL> self . free_slots += <NUM_LIT:1> <EOL> get_task_events = self . maybe_start_task ( current_time ) <EOL> return get_task_events <EOL> def maybe_start_task ( self , current_time ) : <EOL> if not self . queued_tasks . empty ( ) and self . free_slots > <NUM_LIT:0> : <EOL> self . free_slots -= <NUM_LIT:1> <EOL> task_duration , job_id = self . queued_tasks . get ( ) <EOL> self . num_queued_tasks -= <NUM_LIT:1> <EOL> task_end_time = current_time + task_duration <EOL> self . simulation . add_task_completion_time ( job_id , task_end_time ) <EOL> return [ ( task_end_time , TaskEndEvent ( self ) ) ] <EOL> return [ ] <EOL> class Simulation ( object ) : <EOL> def __init__ ( self , num_jobs , file_prefix , load , task_distribution ) : <EOL> avg_used_slots = load * SLOTS_PER_WORKER * TOTAL_WORKERS <EOL> self . interarrival_delay = ( <NUM_LIT:1.0> * MEDIAN_TASK_DURATION * TASKS_PER_JOB / avg_used_slots ) <EOL> print ( "<STR_LIT>" % <EOL> ( self . interarrival_delay , avg_used_slots ) ) <EOL> self . jobs = { } <EOL> self . remaining_jobs = num_jobs <EOL> self . event_queue = Queue . PriorityQueue ( ) <EOL> self . workers = [ ] <EOL> self . file_prefix = file_prefix <EOL> while len ( self . workers ) < TOTAL_WORKERS : <EOL> self . workers . append ( Worker ( self , SLOTS_PER_WORKER , len ( self . workers ) ) ) <EOL> self . worker_indices = range ( TOTAL_WORKERS ) <EOL> self . task_distribution = task_distribution <EOL> def assign_tasks ( self , current_time , job_id ) : <EOL> job = self . jobs [ job_id ] <EOL> random . shuffle ( self . worker_indices ) <EOL> num_probes = PROBE_RATIO * job . num_tasks <EOL> loads = [ ( i , self . workers [ i ] . queue_length ( ) ) <EOL> for i in self . worker_indices [ : num_probes ] ] <EOL> loads . sort ( key = lambda x : x [ <NUM_LIT:1> ] ) <EOL> task_arrival_events = [ ] <EOL> for i , ( worker_index , load ) in enumerate ( loads [ : job . num_tasks ] ) : <EOL> task_arrival_events . append ( <EOL> ( current_time + <NUM_LIT:2> * NETWORK_DELAY , <EOL> TaskArrival ( self . workers [ worker_index ] , job . unscheduled_tasks [ i ] , job_id ) ) ) <EOL> return task_arrival_events <EOL> def assign_tasks_per_task ( self , current_time , job_id ) : <EOL> """<STR_LIT>""" <EOL> job = self . jobs [ job_id ] <EOL> random . shuffle ( self . worker_indices ) <EOL> task_arrival_events = [ ] <EOL> for i , task_duration in enumerate ( job . unscheduled_tasks ) : <EOL> loads = [ ( i , self . workers [ i ] . queue_length ( ) ) <EOL> for i in self . worker_indices [ PROBE_RATIO * i : PROBE_RATIO * ( i + <NUM_LIT:1> ) ] ] <EOL> loads . sort ( key = lambda x : x [ <NUM_LIT:1> ] ) <EOL> task_arrival_events . append ( <EOL> ( current_time + <NUM_LIT:2> * NETWORK_DELAY , <EOL> TaskArrival ( self . workers [ loads [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] ] , task_duration , job_id ) ) ) <EOL> return task_arrival_events <EOL> def send_probes ( self , job , current_time ) : <EOL> """<STR_LIT>""" <EOL> self . jobs [ job . id ] = job <EOL> return ( current_time + NETWORK_DELAY , ProbeEvent ( self , job . id ) ) <EOL> def add_task_completion_time ( self , job_id , completion_time ) : <EOL> job_complete = self . jobs [ job_id ] . task_completed ( completion_time ) <EOL> if job_complete : <EOL> self . remaining_jobs -= <NUM_LIT:1> <EOL> def run ( self ) : <EOL> self . event_queue . put ( ( <NUM_LIT:0> , JobArrival ( self , self . interarrival_delay , self . task_distribution ) ) ) <EOL> last_time = <NUM_LIT:0> <EOL> while self . remaining_jobs > <NUM_LIT:0> : <EOL> current_time , event = self . event_queue . get ( ) <EOL> assert current_time >= last_time <EOL> last_time = current_time <EOL> new_events = event . run ( current_time ) <EOL> for new_event in new_events : <EOL> self . event_queue . put ( new_event ) <EOL> print ( "<STR_LIT>" % <EOL> ( last_time , len ( self . jobs ) ) ) <EOL> complete_jobs = [ j for j in self . jobs . values ( ) if j . completed_tasks_count == j . num_tasks ] <EOL> print "<STR_LIT>" % len ( complete_jobs ) <EOL> response_times = [ job . end_time - job . start_time for job in complete_jobs <EOL> if job . start_time > <NUM_LIT> ] <EOL> print "<STR_LIT>" % len ( response_times ) <EOL> plot_cdf ( response_times , "<STR_LIT>" % self . file_prefix ) <EOL> print "<STR_LIT>" , numpy . mean ( response_times ) <EOL> longest_tasks = [ job . longest_task for job in complete_jobs ] <EOL> plot_cdf ( longest_tasks , "<STR_LIT>" % self . file_prefix ) <EOL> return response_times <EOL> def main ( ) : <EOL> sim = Simulation ( <NUM_LIT:10> , "<STR_LIT>" , <NUM_LIT> , TaskDistributions . CONSTANT ) <EOL> sim . run ( ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> main ( ) </s>
