Broadway Snippet

scheduler.py

# This was modified from a class assignment
# I did the following:
# - translated from c++ to python
# - optimized broadcast to signal via condition variable map to reduce unnecessary wakeups and context switches
# - switched shortest seek first to combined metric to also optimize on resource starvation
# Since this was translated, there might be some inconsistencies. Nonetheless, I can produce high quality software.
# Areas to be improved: docs/comments, unit/functional testing, static analysis (hopefully for threading) and quality linting
# globals could be raised to service and communication could happen over channels or other higher level synchronization primitives
# This was tested 20k times to check for numerous thread interleavings

from threading import Lock, Condition, Thread
import sys
from math import abs
from functools import reduce

lock = Lock()
free_spot = Condition(lock)
spot_filled = Condition(lock)
request_complete = Condition(lock)
service: [Condition] = []
files = []
tracks_in_queue: [tuple] = []
completed = []
number_serviced = []
queue_size = -1
max_size = -1
allowed_size = 0
total_left = -1
total_completed = -1

def main():
    for i in range(len(sys.argv)):
        if not i:
            continue
        cur = sys.argv[i]
        if i == 1:
            max_size = int(cur)
        else:
            files.append(cur)
    completed = [False] * len(files)
    number_serviced = [0] * len(files)
    service = [Condition(lock) for _ in files]
    total_left = len(files)
    tracks_in_queue = [(-1, -1)] * max_size
    server()

def requestor(id):
    with lock:
        filename = files[id]
        with open(filename) as file:
            for line in file:
                track = int(line)
                while not ((queue_size < max_size) and queue_size < allowed_size):
                    free_spot.wait()
                i = tracks_in_queue.index((-1, -1))
                cur = (track, id)
                tracks_in_queue[i] = cur
                print_request(id, track)
                queue_size += 1
                spot_filled.signal()
                while tracks_in_queue[i] != (-1, -1):
                    service[i].wait()
                request_complete.signal()
                total_completed = id
        completed[id] = True
        total_left -= 1

def signal_free():
    if (completed[check]) return
    oldSize = queue_size
    free_spot.signal()
    while queue_size != oldSize + 1:
        spot_filled.wait()

def get_next(prev):
    def helper(best_idx, cur):
      best_track, best_requestor = tracks_in_queue[best_idx]
      (idx, (cur_track, cur_requestor)) = cur
      if best_track < 0:
        return idx
      # we want to minimize disk head travel time and minimize starvation
      # we want to penalize this metric for high distance and number_serviced
      # this needs tuning and perfomance benchmarks to best determine k
      k = 50
      newDiff = abs(cur_track - prev) + k * number_serviced
      oldDiff = abs(cur_track - prev) + k * number_serviced
      if newDiff < oldDiff:
        return idx
      return cur
      
    return reduce(helper, enumerate(tracks_in_queue))

def service_one(prev):
    next_ind = get_next(prev)
    track, requestor = tracks_in_queue[next_ind]
    print_service(requestor, track)
    service[i].signal()
    queue_size -= 1
    tracks_in_queue[next_ind] = (-1, -1)
    while total_completed != requestor:
        request_complete.wait()
    total_completed = -1
    number_serviced[requestor] += 1
    if not completed[requestor] or total_left >= max_size: 
        signal_free()
    return track

def server():
    with lock:
        prev_track = 0
        queue_size = 0
        for filename in files:
            Thread(requestor, filename)
        while queue_size < max_size and queue_size != total_left:
            allowed_size += 1
            signal_free()
        while(total_left):
            prev_track = service_one(prev_track, needSignal)

if __name__ == "__main__":
    main()