Locality scheduling redesign

Assumptions

• A batch consists of a sequence of files F1 ... Fn and a set of jobs J1 ... Jm operating on these files.
• Each job uses a contiguous set of files.
• A given file may be used by many jobs.
• The density of jobs in the file sequence may be variable.
• Several batches may be in progress concurrently, for the same or different applications.

Goals

• To complete batches quickly.
• To minimize the amount of data transfer to hosts.

Scheduling policy

The policy in which we dispatch the jobs in order essentially sends every file to every host, so it fails to achieve the 2nd goal.

The ideal policy would start each host at a different point in the job space, separated according to their speeds. This would potentially send each file only to a single host. However, it's impractical for various reasons: replication, unreliability of hosts, and unpredictability of their speed.

Instead, we use a policy in which the set of hosts is divided into teams, and each team works on a different area of the job space. Teams should have these properties:

• The fastest host in a team should be no faster than the total of the other hosts. Otherwise, it could get unboundedly far ahead of the others.
• Subject to the above, teams should as small as possible. A good size might be 10 or 20.
• The hosts in a team should belong to different users (for validation purposes).

Because of host churn, team membership is dynamic; e.g. a team may be empty for a period.

Design

A cursor consists of

• a team of hosts
• a range of jobs
• status information (see below)

Note: we discussed having a separate notion of "job range", allowing cursors to move from one job range to another, and allowing job ranges to be subdivided. I think this is needlessly complex.

Database

New tables:

batch
    // this table already exists; we may need to add fields to it

batch_host              // batch/host association table
    host_id integer
    batch_id integer
    cursor_id integer

locality_cursor
    batch_id integer
    first_job_num integer
    last_job_num integer
        // range of jobs to be done
    expavg_credit double
        // sum of expavg_credit of hosts in the team
    remaining_credit double
        // estimated credit of unfinished jobs
        // zero means all jobs finished
    first_unfinished_job_num integer
        // all jobs before this have been completed
    first_ungenerated_job_num integer
        // we've generated workunit records for all jobs before this

workunit (new fields)
    cursor_id integer
    job_num integer

Creating a batch

To create a batch:

• create batch record
• based on # of hosts, # of jobs, and job density, create locality_cursor records

Scheduler

Assign host to cursors

Define

est_time_left(cursor) = cursor.remaining_credit/cursor.expavg_credit

This is an estimate of the time needed to complete the cursor's jobs, given its current team.

For each batch:

If this is a new host (i.e. no batch_host record) then

• assign host to cursor for this batch with least greatest est_time_left().
• create batch_host record
• add host's expavg_credit to cursor's expavg_credit

Otherwise, consider moving this host to a different cursor. Let C = host's cursor, and let D = the cursor for which est_time_left() is greatest. If est_time_left(C) < .5*est_time_left(D), then move this host to D. (This policy may need to be refined a bit to reduce moving hosts between cursors).

Assigning jobs

For each batch:

Enumerate unsent results for this host's cursor in order of increasing job num (i.e. finish old jobs before starting new ones). Send as many as the host can handle.

Some type of synchronization is needed; maybe

• use a DB transaction (how much would this lock?)
• explicitly lock the locality_cursor record (is this possible?)
• use a semaphore per cursor

Deleting files

If the host has a sticky file that's not used by an unfinished job in its cursor, tell client to delete that file.

Note: names of sticky files should encode the batch and file number.

Work generator

Loop over batches and cursors. Try to maintain a cushion of N unsent jobs per cursor. Start generating jobs at cursor.first_ungenerated_job_num.

Transitioner

When a workunit is completed (successfully or not):

while cursor.first_unfinished_job_num is finished
    cursor.first_unfinished_job_num++
    update cursor.remaining_credit

locality_daemon

This runs periodically (every hour or day) and recomputes locality_cursor.expavg_credit. It doesn't have to be a separate program; it could be added to an existing daemon, like the feeder