myxce/src/scheduler/testsuite/test_job_move_priority.c

/*
 * test_job_move_priority.c - bug-1944: moving a pending job to another
 * queue must not change the job's own priority, so the moved job does not
 * overtake jobs submitted earlier to the target queue.
 *
 * Fixture: queue qa (priority 0) and queue qb (priority 10); jobs B1 and B2
 * pending in qb (queue_time 1000, 1001), job M pending in qa (1002). M is
 * moved to qb through the core's JOB_UPDATE event (core_events.c), the path
 * both drivers use for a job-fields update. One worker slot, so only the
 * first job the picker offers runs. FCFS orders a queue's jobs by job
 * priority, then queue time: B1 must run and M must stay pending with its
 * submit-time priority (0). Built like test_sim_contention_limit_driver_
 * parity.c's normal-mode fixture (real scheduler_srcs, real picker and
 * allocator threads, `static` fixture, exactly one RUN_TEST()).
 */
#include <errno.h>
#include <pthread.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>

#include <unity/unity.h>

#include "controller.capnp.h"
#include "decision.h"
#include "plugin.h"
#include "preason.h"
#include "resobj.h"
#include "sched/types.h"
#include "scheduler.h"
#include "xstring.h"

#define WORKER_NAME "jmove_worker"
#define QUEUE_A     "jmove_qa"
#define QUEUE_B     "jmove_qb"
#define TEST_RESREQ ((char*)"select[type==any] rusage[mem=1]")

#define JOB_B1_ID ((uint64_t)194401ULL)
#define JOB_B2_ID ((uint64_t)194402ULL)
#define JOB_M_ID  ((uint64_t)194403ULL)

void setUp(void) { return; }
void tearDown(void) { return; }

static void register_slots_resource(var resmgr)
{
    xce_resource_t res;

    memset(&res, 0, sizeof(res));
    res.name    = strdup("slots");
    res.valtype = XCE_VAL_NUMERIC;
    res.ordtype = XCE_ORD_DECR;
    res.flags   = XCE_RESF_LIMIT | XCE_RESF_CONSUMABLE | XCE_RESF_RELEASE
                | XCE_RESF_DEFAULT_HOST;

    TEST_ASSERT_NOT_NULL(makeResource(resmgr, &res));
}

static void add_worker(xce_sched_core_t* core, double slots)
{
    xce_sched_worker_t* w   = (xce_sched_worker_t*)calloc(1, sizeof(*w));
    xce_resource_t*     r   = NULL;
    xce_val_t*          val = NULL;
    int                 rc;

    TEST_ASSERT_NOT_NULL(w);
    w->name = strdup(WORKER_NAME);
    w->idx  = -1;
    zarray_init(&(w->resvals), 128, (zarray_item_killer_t)kill_sched_resval);
    zarray_init(&(w->resInUse), 128, (zarray_item_killer_t)kill_sched_resval);

    rc = key_find(core->resmgr, (void*)"slots", (void**)&r);
    TEST_ASSERT_TRUE(rc >= 0);
    val = (xce_val_t*)calloc(1, sizeof(*val));
    TEST_ASSERT_NOT_NULL(val);
    val->which    = Value_u_fvalue;
    val->u.fvalue = slots;
    zarray_put(&(w->resvals), r->idx, val);

    rc = sched_add_worker(core, w);
    TEST_ASSERT_TRUE(rc >= 0);
}

static void add_queue(xce_sched_core_t* core, xce_sched_plugin_t* picker,
                      const char* name, int priority)
{
    xce_sched_queue_t* q = (xce_sched_queue_t*)calloc(1, sizeof(*q));
    int                rc;

    TEST_ASSERT_NOT_NULL(q);
    q->name     = makeString((char*)name);
    q->priority = priority;
    rc          = sched_add_queue(core, q);
    TEST_ASSERT_TRUE(rc >= 0);
    picker->picker_iface->on_queue_update(picker, q);
}

static void push_job(xce_scheduler_t* sched, xce_sched_plugin_t* picker,
                     uint64_t jobid, const char* queue, uint64_t queue_time)
{
    job_submit_data_t submit;
    xce_job_t         j;
    xce_sched_job_t*  job;
    int               rc;

    memset(&submit, 0, sizeof(submit));
    submit.queue        = (char*)queue;
    submit.user         = (char*)"jmove_user";
    submit.project_name = (char*)"default";
    submit.resreq       = TEST_RESREQ;
    submit.mintasks     = 1;
    submit.maxtasks     = 1;

    memset(&j, 0, sizeof(j));
    j.jobid      = jobid;
    j.state      = JOB_STAT_PEND;
    j.sub_time   = queue_time;
    j.queue_time = queue_time;
    j.submit     = &submit;

    job = make_sched_job(sched, &j);
    TEST_ASSERT_NOT_NULL(job);
    job->schedobj->u.primary->resreq->minTasks = 1;
    job->schedobj->u.primary->resreq->maxTasks = 1;
    rc = sched_add_job(&(sched->core), job);
    TEST_ASSERT_TRUE(rc >= 0);
    picker->picker_iface->on_job_push(picker, job);
}

static void run_round(xce_sched_core_t* core)
{
    pthread_t       picker_tid;
    pthread_t       allocator_tid;
    xce_decision_t* d;
    int             rc;

    rc = pthread_create(&picker_tid, NULL, sched_job_picker, core);
    TEST_ASSERT_EQUAL_INT(0, rc);
    rc = pthread_create(&allocator_tid, NULL, sched_allocator, core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    rc = xce_core_go(core);
    TEST_ASSERT_EQUAL_INT(0, rc);

    do {
        d = NULL;
        cbuf_get(&(core->decisions), &d);
    } while (d != NULL);

    rc = xce_core_end_session(core);
    TEST_ASSERT_EQUAL_INT(0, rc);
}

static xce_scheduler_t g_sched;

static void test_moved_job_keeps_its_priority_and_place(void)
{
    xce_sched_plugin_t* fcfs = NULL;
    job_submit_data_t   move_submit;
    xce_job_t           move;
    xce_core_event_t    ev;
    xce_sched_job_t*    m;
    int                 rc;

    memset(&g_sched, 0, sizeof(g_sched));
    rc = xce_sched_core_init(&(g_sched.core));
    TEST_ASSERT_EQUAL_INT(0, rc);
    register_slots_resource(g_sched.core.resmgr);

    fcfs = builtin_fcfs_new_plugin();
    TEST_ASSERT_NOT_NULL(fcfs);
    rc = plugin_register(&g_sched, fcfs);
    TEST_ASSERT_TRUE(rc >= 0);
    g_sched.core.job_pick_plugin = (char*)"builtin_fcfs";
    rc = register_seqjob_allocator(&g_sched);
    TEST_ASSERT_TRUE(rc >= 0);
    rc = register_parjob_allocator(&g_sched);
    TEST_ASSERT_TRUE(rc >= 0);

    add_queue(&(g_sched.core), fcfs, QUEUE_A, 0);
    add_queue(&(g_sched.core), fcfs, QUEUE_B, 10);
    add_worker(&(g_sched.core), 1.0);

    push_job(&g_sched, fcfs, JOB_B1_ID, QUEUE_B, 1000ULL);
    push_job(&g_sched, fcfs, JOB_B2_ID, QUEUE_B, 1001ULL);
    push_job(&g_sched, fcfs, JOB_M_ID, QUEUE_A, 1002ULL);

    m = sched_get_job(&(g_sched.core), JOB_M_ID);
    TEST_ASSERT_NOT_NULL(m);
    TEST_ASSERT_EQUAL_INT(0, m->priority);

    /* move M from qa to qb */
    memset(&move_submit, 0, sizeof(move_submit));
    move_submit.queue = (char*)QUEUE_B;
    memset(&move, 0, sizeof(move));
    move.jobid  = JOB_M_ID;
    move.state  = JOB_STAT_PEND;
    move.submit = &move_submit;

    memset(&ev, 0, sizeof(ev));
    ev.type             = XCE_CORE_EVENT_JOB_UPDATE;
    ev.u.job_update.job = &move;
    rc = xce_core_push_events(&(g_sched.core), &ev, 1);
    TEST_ASSERT_TRUE(rc >= 0);

    TEST_ASSERT_EQUAL_STRING_MESSAGE(QUEUE_B, m->queue->name,
                                     "the move must reach the job");
    TEST_ASSERT_EQUAL_INT_MESSAGE(
        0, m->priority,
        "moving a job must not change its own priority (bug-1944)");

    run_round(&(g_sched.core));

    TEST_ASSERT_EQUAL_UINT_MESSAGE(
        JOB_STAT_RUN, sched_get_job(&(g_sched.core), JOB_B1_ID)->state,
        "B1, first in qb, must take the only slot");
    TEST_ASSERT_EQUAL_UINT_MESSAGE(
        JOB_STAT_PEND, m->state,
        "the moved job must not overtake jobs already pending in qb");
}

int main(void)
{
    UNITY_BEGIN();
    RUN_TEST(test_moved_job_keeps_its_priority_and_place);
    return UNITY_END();
}