# Finding the earliest time we can schedule

We have a list of $$\j \in \{1,\ldots,N\}\$$ jobs with a processing time $$\p_j\$$ and a demand $$\d_j\$$ between 0 and 1. These are real numbers. Jobs require the fixed demand over the entirety of their processing time. We should not sort this list of jobs.

Our scheduling algorithm should do the following: Take the $$\i^{th}\$$ job from the list, and schedule it in the earliest time it is feasible. A job is feasibly scheduled at a time $$\t\$$ when the total demand of jobs that are being executed at time $$\t\$$ less than or equal to 1.

We construct a 2D array X of size N-by-3, where the columns are [start, completion, demand] and uses numpy.where to identify all jobs alive at time t. Using this construction, we scan over the time horizon. If a time t is feasible, we mark it as a candidate and keep scanning. If it has scanned a distance at least $$\p_j\$$ without breaking feasibility, then it means it can be scheduled at that candidate time point. Otherwise if it hit a snag before $$\p_j\$$, it means that none of the previous time points were feasible.

I am wondering if this can be sped up with smarter usage of numpy arrays or a different algorithm.

import numpy as np
from dataclasses import dataclass, field
from itertools import count
from numpy.testing import assert_almost_equal

@dataclass
class Job:
p: float = 0
d: float = 0
S: float = None

id: int = field(default_factory=count().__next__)  # Starts at 0

def schedule(jobs):
# jobs is a list of job. A job object contains 4 attributes: An index id, p, d, and start time S (which is defaulted to None)

# Three columns are start time, completion time, and  demand
X = np.ones(shape=(len(jobs), 3)) * -1
for job in jobs:
makespan = np.max(X[:, 1], initial=0)
t = 0
S = None
# Scan over the time horizon, while maintaining S that keeps track of earliest feasible start
while t <= makespan + job.p:
# Find all the jobs alive at time t. i.e. Start <= t < Completion
alive_jobs = X[np.where((X[:, 0] <= t) & (t < X[:, 1]))]
total_demand = np.sum(alive_jobs[:, 2])
if total_demand + job.d <= 1:
# t is a feasible start. Ensure the job fits for the entirety of its processing time
if S is None:
S = t

if S + job.p <= t:
# We have scanned over a horizon of at least p. This job is feasible
break
else:
# Could not feasibly schedule job over its processing time. Our candidate is not valid
S = None

# Advance time horizon by the earliest completion time of alive jobs.
t = min(alive_jobs[:, 1]) if alive_jobs.size > 0 else t + job.p

X[job.id, 0] = S
X[job.id, 1] = S + job.p
X[job.id, 2] = job.d

return X

def main():
# Random instance
processing_times = [1, 6, 2, 4, 7, 3, 7, 8]
demands = [0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.2]

desired_result = np.array([
[0,   1,   0.2],  # First job is assigned at t=0, current resource usage is 0.2
[0,   6,   0.3],  # Second job is assigned at t=0, since 0.2 + 0.3 = 0.5 <= 1
[0,   2,   0.4],  # Third job can also start at t=0, with a current total resource use of 0.9 <= 1
[2,   6,   0.5],  # Fourth job exceeds the current resource usage. It can't start when the first job finishes,
# because the total resource use at t=1 is 0.3 + 0.4 = 0.7. Therefore this job starts when
# the third job ends at t=2, with a total resource use 0.5 + 0.3 = 0.8
[6,  13,   0.6],  # Fifth job has the same story as the fourth. The earliest feasible time it can start is t=6
[13,  16,   0.7],
[16,  23,   0.8],
[1,   9,   0.2]  # The last job can actually start at t=1, because over the time t in [1,9) the resource use
# does not exceed 0.8. From [0, 2) the resource use is due to the second and third with total
# resource use 0.7. From [2, 6) the second and fourth jobs use 0.8 resources. From [6, 9),
# only the fifth job is alive, using only 0.8 resources.
])

jobs = []
for p, d in zip(processing_times, demands):
jobs.append(Job(p=p, d=d))

X = schedule(jobs)
assert_almost_equal(X, desired_result)
print(X)

if __name__ == '__main__':
main()

• What is the business requirement of not being able to sort you jobs array? Does this mean your jobs have to run in the submitted order? Mar 21 at 10:26
• Yes the sorting of the jobs is controlled by another part of the program that will eventually call this function. Mar 21 at 13:38
• Show us how to run this, please. I'm looking for a test suite, or at the very least def main(): which supplies example data to exercise the code and describes what a correct result would look like. If this code is following something from the literature, please offer a citation.
– J_H
Mar 28 at 18:48
• @J_H I have provided a test case Mar 29 at 19:39

A faster way would be to keep track of the amount of resources left. The one aspect that you need to keep in mind with this method is the variable max_required_time. You need to be 99.9% confident that this value will sufficient. You can determine this by looking at the logs of your previous scheduled tasks and applying a little statistics.

The closer max_required_time is the the minimum value required for the passed in jobs the faster this method will be.

def schedule(jobs):
# Using domain knowledge you can figure out an optimal value for max_required_time
# Even at 100 which is much higher than what is required this is still faster by ~1/3
max_required_time = 100
# How much resource is left at each point in time
free_resources = np.ones(max_required_time)

X = np.ones(shape=(len(jobs), 3)) * -1
for job in jobs:
delta_t = job.p
cost = job.d
# Is there enough resource at the specific time
# Determine what ranges are valid locations.
valid_ranges = np.convolve(has_space, np.ones(delta_t) , mode='valid') / delta_t >= 1
# Get first valid index
idx = valid_ranges.view(bool).argmax() // valid_ranges.itemsize
# If you know that [max_required_time] will always be sufficient you won't need to run checks.
# else uncomment and handle
#idx = idx if valid_ranges[idx] else -1
# Spend the resources for the process.
free_resources[idx:idx + delta_t] = np.round(free_resources[idx:idx+delta_t] - cost,5)

X[job.id, 0] = idx
X[job.id, 1] = idx + job.p
X[job.id, 2] = job.d

return X


If you don't have logs what you can use:

def schedule2(jobs, max_required_time = 20):

# How much resource is left at each point in time
free_resources = np.ones(max_required_time)

X = np.ones(shape=(len(jobs), 3)) * -1
for job in jobs:
delta_t = job.p
cost = job.d
# Is there enough resource at the specific time
# Determine what ranges are valid locations.
valid_ranges = np.convolve(has_space, np.ones(delta_t) , mode='valid') / delta_t >= 1
# Get first valid index
idx = valid_ranges.view(bool).argmax() // valid_ranges.itemsize
# If you know that [max_required_time] will always be sufficient you won't need to run checks.
# else uncomment and handle
idx = idx if valid_ranges[idx] else -1
if idx == -1:
results = schedule2(jobs, max_required_time*2)
return results
# Spend the resources for the process.
free_resources[idx:idx + delta_t] = np.round(free_resources[idx:idx+delta_t] - cost,5)

X[job.id, 0] = idx
X[job.id, 1] = idx + job.p
X[job.id, 2] = job.d

return X


Running both yours and my second method 3 times for n number of jobs

N = [2,4,8,16,32,64,128,256,512]

Gave me the following grapy

For reference here is the complete code used to generate the graph.

import numpy as np
from dataclasses import dataclass, field
from itertools import count
from numpy.testing import assert_almost_equal
import timeit

@dataclass
class Job:
p: float = 0
d: float = 0
S: float = None

id: int = field(default_factory=count().__next__)  # Starts at 0

def schedule(jobs):
# jobs is a list of job. A job object contains 4 attributes: An index id, p, d, and start time S (which is defaulted to None)

# Three columns are start time, completion time, and  demand
X = np.ones(shape=(len(jobs), 3)) * -1
for job in jobs:
makespan = np.max(X[:, 1], initial=0)
t = 0
S = None
# Scan over the time horizon, while maintaining S that keeps track of earliest feasible start
while t <= makespan + job.p:
# Find all the jobs alive at time t. i.e. Start <= t < Completion
alive_jobs = X[np.where((X[:, 0] <= t) & (t < X[:, 1]))]
total_demand = np.sum(alive_jobs[:, 2])
if total_demand + job.d <= 1:
# t is a feasible start. Ensure the job fits for the entirety of its processing time
if S is None:
S = t

if S + job.p <= t:
# We have scanned over a horizon of at least p. This job is feasible
break
else:
# Could not feasibly schedule job over its processing time. Our candidate is not valid
S = None

# Advance time horizon by the earliest completion time of alive jobs.
t = min(alive_jobs[:, 1]) if alive_jobs.size > 0 else t + job.p

X[job.id, 0] = S
X[job.id, 1] = S + job.p
X[job.id, 2] = job.d

return X

def schedule2(jobs, max_required_time = 20):
# Using domain knowledge you can figure out an optimal value for max_required_time
# Even at 100 which is much higher than what is required this is still faster by ~1/3

# How much resource is left at each point in time
free_resources = np.ones(max_required_time)

X = np.ones(shape=(len(jobs), 3)) * -1
for job in jobs:
delta_t = job.p
cost = job.d
# Is there enough resource at the specific time
# Determine what ranges are valid locations.
valid_ranges = np.convolve(has_space, np.ones(delta_t) , mode='valid') / delta_t >= 1
# Get first valid index
idx = valid_ranges.view(bool).argmax() // valid_ranges.itemsize
# If you know that [max_required_time] will always be sufficient you won't need to run checks.
# else uncomment and handle
idx = idx if valid_ranges[idx] else -1
if idx == -1:
results = schedule2(jobs, max_required_time*2)
return results
# Spend the resources for the process.
free_resources[idx:idx + delta_t] = np.round(free_resources[idx:idx+delta_t] - cost,5)

X[job.id, 0] = idx
X[job.id, 1] = idx + job.p
X[job.id, 2] = job.d

return X
import random

def main():
Xs = []
S1 = []
S2 = []
for i in range(10):
job_count = 2 ** i
Xs.append(job_count)

# Random instance
processing_times = []
for i in range(job_count):
processing_times.append(random.randint(1,10))
demands = []

for i in range(job_count):
v = round(random.random(),2)
v = v if 1 > v > 0 else 0.5
demands.append(v)

jobs = []
i = 0
for p, d in zip(processing_times, demands):
jobs.append(Job(p=p, d=d, id= i))
i += 1
run_count = 3

start = timeit.default_timer()
for i in range(run_count):
_ = schedule(jobs)
stop = timeit.default_timer()
execution_time = stop - start
S1.append(execution_time)
start = timeit.default_timer()
for i in range(run_count):
_ = schedule2(jobs)
stop = timeit.default_timer()
execution_time = stop - start
S2.append(execution_time)
print(Xs, S1, S2)
import matplotlib.pyplot as plt
plt.plot(Xs, S1)
plt.plot(Xs, S2)
plt.show()

if __name__ == '__main__':
main()