// This file is part of BOINC. // http://boinc.berkeley.edu // Copyright (C) 2008 University of California // // BOINC is free software; you can redistribute it and/or modify it // under the terms of the GNU Lesser General Public License // as published by the Free Software Foundation, // either version 3 of the License, or (at your option) any later version. // // BOINC is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. // See the GNU Lesser General Public License for more details. // // You should have received a copy of the GNU Lesser General Public License // along with BOINC. If not, see . #include "cpp.h" #ifdef _WIN32 #include "boinc_win.h" #else #include "config.h" #endif #include "util.h" #include "client_msgs.h" #include "client_state.h" #include "project.h" #include "result.h" #include "scheduler_op.h" #include "work_fetch.h" //#define DEBUG(x) x #define DEBUG(X) using std::vector; RSC_WORK_FETCH rsc_work_fetch[MAX_RSC]; WORK_FETCH work_fetch; static inline bool dont_fetch(PROJECT* p, int rsc_type) { if (p->no_rsc_pref[rsc_type]) return true; if (p->no_rsc_config[rsc_type]) return true; if (p->no_rsc_apps[rsc_type]) return true; if (p->no_rsc_ams[rsc_type]) return true; return false; } // if the configuration file disallows the use of a GPU type // for a project, set a flag to that effect // void set_no_rsc_config() { for (unsigned int i=0; iproject != p) continue; if (rp->state() <= RESULT_FILES_DOWNLOADED) { return true; } } return false; } inline bool has_coproc_app(PROJECT* p, int rsc_type) { unsigned int i; for (i=0; iproject != p) continue; if (avp->gpu_usage.rsc_type == rsc_type) return true; } return false; } /////////////// RSC_PROJECT_WORK_FETCH /////////////// bool RSC_PROJECT_WORK_FETCH::compute_may_have_work(PROJECT* p, int rsc_type) { if (dont_fetch(p, rsc_type)) return false; if (p->rsc_defer_sched[rsc_type]) return false; return (backoff_time < gstate.now); } void RSC_PROJECT_WORK_FETCH::rr_init(PROJECT* p, int rsc_type) { may_have_work = compute_may_have_work(p, rsc_type); fetchable_share = 0; n_runnable_jobs = 0; sim_nused = 0; nused_total = 0; deadlines_missed = 0; } void RSC_PROJECT_WORK_FETCH::resource_backoff(PROJECT* p, const char* name) { if (backoff_interval) { backoff_interval *= 2; if (backoff_interval > WF_MAX_BACKOFF_INTERVAL) backoff_interval = WF_MAX_BACKOFF_INTERVAL; } else { backoff_interval = WF_MIN_BACKOFF_INTERVAL; } double x = (.5 + drand())*backoff_interval; backoff_time = gstate.now + x; if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch] backing off %s %.0f sec", name, x ); } } /////////////// RSC_WORK_FETCH /////////////// RSC_PROJECT_WORK_FETCH& RSC_WORK_FETCH::project_state(PROJECT* p) { return p->rsc_pwf[rsc_type]; } void RSC_WORK_FETCH::rr_init() { shortfall = 0; nidle_now = 0; sim_nused = 0; total_fetchable_share = 0; deadline_missed_instances = 0; saturated_time = 0; busy_time_estimator.reset(); sim_used_instances = 0; } void RSC_WORK_FETCH::update_stats(double sim_now, double dt, double buf_end) { double idle = ninstances - sim_nused; if (idle > 1e-6 && sim_now < buf_end) { double dt2; if (sim_now + dt > buf_end) { dt2 = buf_end - sim_now; } else { dt2 = dt; } shortfall += idle*dt2; } if (idle < 1e-6) { saturated_time = sim_now + dt - gstate.now; } } void RSC_WORK_FETCH::update_busy_time(double dur, double nused) { busy_time_estimator.update(dur, nused); } static bool wacky_dcf(PROJECT* p) { if (p->dont_use_dcf) return false; double dcf = p->duration_correction_factor; return (dcf < 0.02 || dcf > 80.0); } // request this project's share of shortfall and instances. // don't request anything if project is backed off. // void RSC_WORK_FETCH::set_request(PROJECT* p) { // if backup project, fetch 1 job per idle instance // if (p->resource_share == 0) { req_instances = nidle_now; req_secs = 1; return; } if (config.fetch_minimal_work) { req_instances = ninstances; req_secs = 1; return; } RSC_PROJECT_WORK_FETCH& w = project_state(p); double non_excl_inst = ninstances - w.ncoprocs_excluded; if (shortfall) { if (wacky_dcf(p)) { // if project's DCF is too big or small, // its completion time estimates are useless; just ask for 1 second // req_secs = 1; } else { req_secs = shortfall; if (w.ncoprocs_excluded) { req_secs *= non_excl_inst/ninstances; } } } double instance_share = ninstances*w.fetchable_share; if (instance_share > non_excl_inst) { instance_share = non_excl_inst; } instance_share -= w.nused_total; req_instances = std::max(nidle_now, instance_share); if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch] set_request() for %s: ninst %d nused_total %f nidle_now %f fetch share %f req_inst %f req_secs %f", rsc_name(rsc_type), ninstances, w.nused_total, nidle_now, w.fetchable_share, req_instances, req_secs ); } if (req_instances && !req_secs) { req_secs = 1; } } // We're fetching work because some instances are starved because // of exclusions. // See how many N of these instances are not excluded for this project. // Ask for N instances and for N*work_buf_min seconds. // void RSC_WORK_FETCH::set_request_excluded(PROJECT* p) { RSC_PROJECT_WORK_FETCH& pwf = project_state(p); int inst_mask = sim_excluded_instances & pwf.non_excluded_instances; int n = 0; for (int i=0; iresource_share == 0 || config.fetch_minimal_work) { req_secs = 1; } else { req_secs = n*gstate.work_buf_total(); } } void RSC_WORK_FETCH::print_state(const char* name) { msg_printf(0, MSG_INFO, "[work_fetch] --- state for %s ---", name); msg_printf(0, MSG_INFO, "[work_fetch] shortfall %.2f nidle %.2f saturated %.2f busy %.2f", shortfall, nidle_now, saturated_time, busy_time_estimator.get_busy_time() ); for (unsigned int i=0; inon_cpu_intensive) continue; RSC_PROJECT_WORK_FETCH& pwf = project_state(p); bool no_rsc_pref = p->no_rsc_pref[rsc_type]; bool no_rsc_config = p->no_rsc_config[rsc_type]; bool no_rsc_apps = p->no_rsc_apps[rsc_type]; bool no_rsc_ams = p->no_rsc_ams[rsc_type]; double bt = pwf.backoff_time>gstate.now?pwf.backoff_time-gstate.now:0; if (bt) { sprintf(buf, " (resource backoff: %.2f, inc %.2f)", bt, pwf.backoff_interval ); } else { strcpy(buf, ""); } msg_printf(p, MSG_INFO, "[work_fetch] fetch share %.3f%s%s%s%s%s", pwf.fetchable_share, buf, no_rsc_pref?" (blocked by prefs)":"", no_rsc_apps?" (no apps)":"", no_rsc_ams?" (blocked by account manager)":"", no_rsc_config?" (blocked by configuration file)":"" ); } } void RSC_WORK_FETCH::clear_request() { req_secs = 0; req_instances = 0; } /////////////// PROJECT_WORK_FETCH /////////////// int PROJECT_WORK_FETCH::compute_cant_fetch_work_reason(PROJECT* p) { if (p->non_cpu_intensive) return CANT_FETCH_WORK_NON_CPU_INTENSIVE; if (p->suspended_via_gui) return CANT_FETCH_WORK_SUSPENDED_VIA_GUI; if (p->master_url_fetch_pending) return CANT_FETCH_WORK_MASTER_URL_FETCH_PENDING; if (p->dont_request_more_work) return CANT_FETCH_WORK_DONT_REQUEST_MORE_WORK; if (p->some_download_stalled()) return CANT_FETCH_WORK_DOWNLOAD_STALLED; if (p->some_result_suspended()) return CANT_FETCH_WORK_RESULT_SUSPENDED; if (p->too_many_uploading_results) return CANT_FETCH_WORK_TOO_MANY_UPLOADS; // this goes last // if (p->min_rpc_time > gstate.now) return CANT_FETCH_WORK_MIN_RPC_TIME; return 0; } void PROJECT_WORK_FETCH::reset(PROJECT* p) { for (int i=0; irsc_pwf[i].reset(); } } /////////////// WORK_FETCH /////////////// // mark the projects from which we can fetch work // void WORK_FETCH::compute_cant_fetch_work_reason() { for (unsigned int i=0; ipwf.cant_fetch_work_reason = p->pwf.compute_cant_fetch_work_reason(p); } } void WORK_FETCH::rr_init() { for (int i=0; ipwf.n_runnable_jobs = 0; for (int j=0; jrsc_pwf[j].rr_init(p, j); } } } #if 0 // if the given project is highest-priority among the projects // eligible for the resource, set request fields // void RSC_WORK_FETCH::supplement(PROJECT* pp) { double x = pp->sched_priority; for (unsigned i=0; ipwf.cant_fetch_work_reason) continue; if (!project_state(p).may_have_work) continue; RSC_PROJECT_WORK_FETCH& rpwf = project_state(p); if (rpwf.anon_skip) continue; if (p->sched_priority > x) { if (log_flags.work_fetch_debug) { msg_printf(pp, MSG_INFO, "[work_fetch]: not requesting work for %s: %s has higher priority", rsc_name(rsc_type), p->get_project_name() ); } return; } } // didn't find a better project; ask for work // set_request(pp); } // we're going to ask the given project for work of the given type. // (or -1 if none) // Set requests for this type and perhaps other types // void WORK_FETCH::set_all_requests_hyst(PROJECT* p, int rsc_type) { for (int i=0; i gstate.work_buf_total()) { continue; } // don't fetch work if backup project and no idle instances // if (p->resource_share==0 && rsc_work_fetch[i].nidle_now==0) { continue; } if (i>0 && !gpus_usable) { continue; } rsc_work_fetch[i].supplement(p); } } } void WORK_FETCH::set_all_requests(PROJECT* p) { for (int i=0; ipwf.cant_fetch_work_reason) { sprintf(buf, "can't req work: %s", cant_fetch_work_string(p->pwf.cant_fetch_work_reason) ); } else { strcpy(buf, "can req work"); } if (p->min_rpc_time > gstate.now) { char buf2[256]; sprintf(buf2, " (backoff: %.2f sec)", p->min_rpc_time - gstate.now); strcat(buf, buf2); } msg_printf(p, MSG_INFO, "[work_fetch] REC %.3f prio %.6f %s", p->pwf.rec, p->sched_priority, buf ); } for (int i=0; idont_request_more_work) return; if (p->non_cpu_intensive) { if (!has_a_job(p)) { rsc_work_fetch[0].req_secs = 1; } return; } // if project was updated from manager and config says so, // fetch work for a resource even if there are higher-prio projects // able to fetch it // bool check_higher_priority_projects = true; if (p->sched_rpc_pending && config.fetch_on_update) { check_higher_priority_projects = false; } setup(); // For each resource, scan projects in decreasing priority, // seeing if there's one that's higher-priority than this // able to fetch work for the resource. // If not, and the resource needs topping off, do so // for (int i=0; ipwf.cant_fetch_work_reason) { DEBUG(msg_printf(p, MSG_INFO, "piggyback: %s can't fetch work", p2->project_name);) continue; } if (rwf.can_fetch(p2) && !rwf.backed_off(p2)) { DEBUG(msg_printf(p, MSG_INFO, "piggyback: better proj %s", p2->project_name);) break; } } if (p != p2) continue; } DEBUG(msg_printf(p, MSG_INFO, "piggyback: requesting %s", rsc_name(i));) if (buffer_low) { rwf.set_request(p); } else { rwf.set_request_excluded(p); } } } // see if there's a fetchable non-CPU-intensive project without work // PROJECT* WORK_FETCH::non_cpu_intensive_project_needing_work() { for (unsigned int i=0; inon_cpu_intensive) continue; if (!p->can_request_work()) continue; if (p->rsc_pwf[0].backoff_time > gstate.now) continue; if (has_a_job(p)) continue; clear_request(); rsc_work_fetch[0].req_secs = 1; return p; } return 0; } static bool higher_priority(PROJECT *p1, PROJECT *p2) { return (p1->sched_priority > p2->sched_priority); } // check resource-level backoff // bool RSC_WORK_FETCH::backed_off(PROJECT* p) { if (project_state(p).backoff_time > gstate.now) { DEBUG(msg_printf(p, MSG_INFO, "skip: backoff");) return true; } return false; } // a variety of checks for whether we should ask this project // for work of this type // bool RSC_WORK_FETCH::can_fetch(PROJECT *p) { // see whether work fetch for this resource is banned // by prefs, config, project, or acct mgr // if (dont_fetch(p, rsc_type)) { DEBUG(msg_printf(p, MSG_INFO, "skip: dont_fetch");) return false; } RSC_PROJECT_WORK_FETCH& rpwf = project_state(p); // if project has zero resource share, // only fetch work if a device is idle // if (p->resource_share == 0 && nidle_now == 0) { DEBUG(msg_printf(p, MSG_INFO, "skip: zero share");) return false; } // if project has excluded GPUs of this type, // we need to avoid fetching work just because there's an idle instance // or a shortfall; // fetching work might not alleviate either of these, // and we'd end up fetching unbounded work. // At the same time, we want to respect work buf params if possible. // // Current policy: // don't fetch work if remaining time of this project's jobs // exceeds work_buf_min * (#usable instances / #instances) // // TODO: THIS IS FAIRLY CRUDE. Making it smarter would require // computing shortfall etc. on a per-project basis // int nexcl = rpwf.ncoprocs_excluded; if (rsc_type && nexcl) { int n_not_excluded = ninstances - nexcl; if (rpwf.queue_est > (gstate.work_buf_min() * n_not_excluded)/ninstances) { DEBUG(msg_printf(p, MSG_INFO, "skip: too much work");) return false; } } if (rpwf.anon_skip) { DEBUG(msg_printf(p, MSG_INFO, "skip: anon");) return false; } return true; } // return true if there is exclusion starvation // and this project can use the starved instances // bool RSC_WORK_FETCH::uses_starved_excluded_instances(PROJECT* p) { RSC_PROJECT_WORK_FETCH& rpwf = project_state(p); if (!sim_excluded_instances) return false; if ((sim_excluded_instances & rpwf.non_excluded_instances) == 0) { DEBUG(msg_printf(p, MSG_INFO, "skip: excl");) return false; } return true; } // setup for choose_project() and piggyback() // void WORK_FETCH::setup() { gstate.compute_nuploading_results(); rr_simulation(); compute_shares(); project_priority_init(true); clear_request(); // Decrement the priority of projects that have work queued. // Specifically, subtract // (FLOPs queued for P)/(FLOPs of max queue) // which will generally be between 0 and 1. // This is a little arbitrary but I can't think of anything better. // double max_queued_flops = gstate.work_buf_total()*total_peak_flops(); for (unsigned int i=0; iproject; p->sched_priority -= rp->estimated_flops_remaining()/max_queued_flops; } std::sort( gstate.projects.begin(), gstate.projects.end(), higher_priority ); if (log_flags.work_fetch_debug) { print_state(); } } // Choose a project to fetch work from, // and set the request fields of resource objects. // If "strict_hyst" is set, require that some resource be below min buf; // otherwise require below max buf // (or // PROJECT* WORK_FETCH::choose_project() { PROJECT* p; if (log_flags.work_fetch_debug) { msg_printf(0, MSG_INFO, "[work_fetch] work fetch start"); } p = non_cpu_intensive_project_needing_work(); if (p) return p; setup(); for (int i=0; ipwf.cant_fetch_work_reason) { DEBUG(msg_printf(p, MSG_INFO, "skip: cfwr %d", p->pwf.cant_fetch_work_reason);) continue; } // For each resource type: // - See if we can ask this project for work of that type; // if so set a flag so that lower-priority projects // won't request it // - If so, see if work is needed for this type; // if so, set "found" // int rsc_index = -1; for (int i=0; i= 0) { for (int i=0; isched_rpc_pending = RPC_REASON_NEED_WORK; } else { if (log_flags.work_fetch_debug) { msg_printf(0, MSG_INFO, "[work_fetch] No project chosen for work fetch"); } p = NULL; } return p; } void WORK_FETCH::accumulate_inst_sec(ACTIVE_TASK* atp, double dt) { APP_VERSION* avp = atp->result->avp; PROJECT* p = atp->result->project; double x = dt*avp->avg_ncpus; p->rsc_pwf[0].secs_this_rec_interval += x; rsc_work_fetch[0].secs_this_rec_interval += x; int rt = avp->gpu_usage.rsc_type; if (rt) { x = dt*avp->gpu_usage.usage; p->rsc_pwf[rt].secs_this_rec_interval += x; rsc_work_fetch[rt].secs_this_rec_interval += x; } } // find total and per-project resource shares for each resource // void WORK_FETCH::compute_shares() { unsigned int i; PROJECT* p; for (i=0; inon_cpu_intensive) continue; if (p->pwf.cant_fetch_work_reason) continue; for (int j=0; jrsc_pwf[j].may_have_work) { rsc_work_fetch[j].total_fetchable_share += p->resource_share; } } } for (i=0; inon_cpu_intensive) continue; if (p->pwf.cant_fetch_work_reason) continue; for (int j=0; jrsc_pwf[j].may_have_work) { p->rsc_pwf[j].fetchable_share = rsc_work_fetch[j].total_fetchable_share?p->resource_share/rsc_work_fetch[j].total_fetchable_share:1; } } } } void WORK_FETCH::request_string(char* buf) { char buf2[256]; sprintf(buf, "[work_fetch] request: CPU (%.2f sec, %.2f inst)", rsc_work_fetch[0].req_secs, rsc_work_fetch[0].req_instances ); for (int i=1; ianonymous_platform) { for (int i=1; i work_req) { work_req = rsc_work_fetch[i].req_secs; } } } } fprintf(f, " %f\n" " %f\n" " %f\n" " %f\n", work_req, rsc_work_fetch[0].req_secs, rsc_work_fetch[0].req_instances, rsc_work_fetch[0].req_secs?rsc_work_fetch[0].busy_time_estimator.get_busy_time():0 ); if (log_flags.work_fetch_debug) { char buf[256]; request_string(buf); msg_printf(p, MSG_INFO, buf); } } // we just got a scheduler reply with the given jobs; update backoffs // void WORK_FETCH::handle_reply( PROJECT* p, SCHEDULER_REPLY*, vector new_results ) { bool got_work[MAX_RSC]; bool requested_work[MAX_RSC]; for (int i=0; i 0); } for (unsigned int i=0; iavp->gpu_usage.rsc_type] = true; } for (int i=0; irsc_pwf[i].backoff_time < gstate.now) { switch (p->sched_rpc_pending) { case RPC_REASON_RESULTS_DUE: case RPC_REASON_NEED_WORK: case RPC_REASON_TRICKLE_UP: p->rsc_pwf[i].resource_backoff(p, rsc_name(i)); } } } // if we did get jobs, clear backoff // if (got_work[i]) { p->rsc_pwf[i].clear_backoff(); } } } // set up for initial RPC. // arrange to always get one job, even if we don't need it or can't handle it. // (this is probably what user wants) // void WORK_FETCH::set_initial_work_request(PROJECT* p) { for (int i=0; irsc_pwf[i].ncoprocs_excluded) { rsc_work_fetch[i].req_secs = 0; } } rsc_work_fetch[i].req_instances = 0; rsc_work_fetch[i].busy_time_estimator.reset(); } } // called once, at client startup // void WORK_FETCH::init() { rsc_work_fetch[0].init(0, gstate.ncpus, 1); double cpu_flops = gstate.host_info.p_fpops; // use 20% as a rough estimate of GPU efficiency for (int i=1; ianonymous_platform) continue; for (int k=0; krsc_pwf[k].anon_skip = true; } for (j=0; jproject != p) continue; p->rsc_pwf[avp->gpu_usage.rsc_type].anon_skip = false; } } } // clear backoff for app's resource // void WORK_FETCH::clear_backoffs(APP_VERSION& av) { av.project->rsc_pwf[av.gpu_usage.rsc_type].clear_backoff(); } //////////////////////// void CLIENT_STATE::compute_nuploading_results() { unsigned int i; for (i=0; inuploading_results = 0; projects[i]->too_many_uploading_results = false; } for (i=0; istate() == RESULT_FILES_UPLOADING) { rp->project->nuploading_results++; } } int n = gstate.ncpus; for (int j=1; j n) { n = coprocs.coprocs[j].count; } } n *= 2; for (i=0; inuploading_results > n) { projects[i]->too_many_uploading_results = true; } } } // Returns the estimated total elapsed time of this task. // Compute this as a weighted average of estimates based on // 1) the workunit's flops count (static estimate) // 2) the current elapsed time and fraction done (dynamic estimate) // double ACTIVE_TASK::est_dur() { if (fraction_done >= 1) return elapsed_time; double wu_est = result->estimated_runtime(); if (fraction_done <= 0) return wu_est; if (wu_est < elapsed_time) wu_est = elapsed_time; double frac_est = fraction_done_elapsed_time / fraction_done; double fd_weight = fraction_done * fraction_done; double wu_weight = 1 - fd_weight; double x = fd_weight*frac_est + wu_weight*wu_est; #if 0 //if (log_flags.rr_simulation) { msg_printf(result->project, MSG_INFO, "[rr_sim] %s frac_est %f = %f/%f", result->name, frac_est, fraction_done_elapsed_time, fraction_done ); msg_printf(result->project, MSG_INFO, "[rr_sim] %s dur: %.2f = %.3f*%.2f + %.3f*%.2f", result->name, x, fd_weight, frac_est, wu_weight, wu_est ); //} #endif return x; } // the fraction of time BOINC is processing // double CLIENT_STATE::overall_cpu_frac() { double x = time_stats.on_frac * time_stats.active_frac; if (x < 0.01) x = 0.01; if (x > 1) x = 1; return x; } double CLIENT_STATE::overall_gpu_frac() { double x = time_stats.on_frac * time_stats.gpu_active_frac; if (x < 0.01) x = 0.01; if (x > 1) x = 1; return x; } double CLIENT_STATE::overall_cpu_and_network_frac() { double x = time_stats.on_frac * time_stats.cpu_and_network_available_frac; if (x < 0.01) x = 0.01; if (x > 1) x = 1; return x; } // called when benchmarks change // void CLIENT_STATE::scale_duration_correction_factors(double factor) { if (factor <= 0) return; for (unsigned int i=0; idont_use_dcf) continue; p->duration_correction_factor *= factor; } if (log_flags.dcf_debug) { msg_printf(NULL, MSG_INFO, "[dcf] scaling all duration correction factors by %f", factor ); } } // Choose a new host CPID. // If using account manager, do scheduler RPCs // to all acct-mgr-attached projects to propagate the CPID // void CLIENT_STATE::generate_new_host_cpid() { host_info.generate_host_cpid(); for (unsigned int i=0; iattached_via_acct_mgr) { projects[i]->sched_rpc_pending = RPC_REASON_ACCT_MGR_REQ; projects[i]->set_min_rpc_time(now + 15, "Sending new host CPID"); } } }