4
\$\begingroup\$

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()
\$\endgroup\$
4
  • \$\begingroup\$ 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? \$\endgroup\$ Mar 21 at 10:26
  • \$\begingroup\$ Yes the sorting of the jobs is controlled by another part of the program that will eventually call this function. \$\endgroup\$
    – Titan
    Mar 21 at 13:38
  • \$\begingroup\$ 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. \$\endgroup\$
    – J_H
    Mar 28 at 18:48
  • \$\begingroup\$ @J_H I have provided a test case \$\endgroup\$
    – Titan
    Mar 29 at 19:39

1 Answer 1

3
\$\begingroup\$

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
        has_space = [free_resource >= cost for free_resource in free_resources]
        # 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
        has_space = [free_resource >= cost for free_resource in free_resources]
        # 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

enter image description here

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
        has_space = [free_resource >= cost for free_resource in free_resources]
        # 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()
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.