// Berkeley Open Infrastructure for Network Computing // http://boinc.berkeley.edu // Copyright (C) 2005 University of California // // This 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 2.1 of the License, or (at your option) any later version. // // This software 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. // // To view the GNU Lesser General Public License visit // http://www.gnu.org/copyleft/lesser.html // or write to the Free Software Foundation, Inc., // 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA // High-level logic for communicating with scheduling servers, // and for merging the result of a scheduler RPC into the client state // The scheduler RPC mechanism is in scheduler_op.C #include "cpp.h" #ifdef _WIN32 #include "boinc_win.h" #endif #ifndef _WIN32 #include "config.h" #include #include #include #include #include #include #endif #include "crypt.h" #include "error_numbers.h" #include "file_names.h" #include "filesys.h" #include "parse.h" #include "util.h" #include "client_msgs.h" #include "scheduler_op.h" #include "client_state.h" using std::max; using std::vector; using std::string; // quantities like avg CPU time decay by a factor of e every week // #define EXP_DECAY_RATE (1./(SECONDS_PER_DAY*7)) // try to report results this much before their deadline // #define REPORT_DEADLINE_CUSHION ((double)SECONDS_PER_DAY) static const char* urgency_name(int urgency) { switch(urgency) { case WORK_FETCH_DONT_NEED: return "Don't need"; case WORK_FETCH_OK: return "OK"; case WORK_FETCH_NEED: return "Need"; case WORK_FETCH_NEED_IMMEDIATELY: return "Need immediately"; } return "Unknown"; } // how many CPUs should this project occupy on average, // based on its resource share relative to a given set // int CLIENT_STATE::proj_min_results(PROJECT* p, double subset_resource_share) { if (p->non_cpu_intensive) { return 1; } if (!subset_resource_share) return 1; // TODO - fix return (int)(ceil(ncpus*p->resource_share/subset_resource_share)); } void CLIENT_STATE::check_project_timeout() { unsigned int i; for (i=0; ipossibly_backed_off && now > p->min_rpc_time) { p->possibly_backed_off = false; request_work_fetch("Project backoff ended"); } } } void PROJECT::set_min_rpc_time(double future_time, const char* reason) { if (future_time > min_rpc_time) { min_rpc_time = future_time; possibly_backed_off = true; msg_printf(this, MSG_INFO, "Deferring communication for %s", timediff_format(min_rpc_time - gstate.now).c_str() ); msg_printf(this, MSG_INFO, "Reason: %s\n", reason); } } // Return true if we should not contact the project yet. // bool PROJECT::waiting_until_min_rpc_time() { return (min_rpc_time > gstate.now); } // find a project that needs to have its master file fetched // PROJECT* CLIENT_STATE::next_project_master_pending() { unsigned int i; PROJECT* p; for (i=0; iwaiting_until_min_rpc_time()) continue; if (p->suspended_via_gui) continue; if (p->master_url_fetch_pending) { return p; } } return 0; } // find a project for which a scheduler RPC is pending // and we're not backed off // PROJECT* CLIENT_STATE::next_project_sched_rpc_pending() { unsigned int i; PROJECT* p; for (i=0; iwaiting_until_min_rpc_time()) continue; if (p->next_rpc_time && p->next_rpc_timesched_rpc_pending = RPC_REASON_PROJECT_REQ; p->next_rpc_time = 0; } //if (p->suspended_via_gui) continue; // do the RPC even if suspended. // This is critical for acct mgrs, to propagate new host CPIDs // if (p->sched_rpc_pending) { return p; } } return 0; } PROJECT* CLIENT_STATE::next_project_trickle_up_pending() { unsigned int i; PROJECT* p; for (i=0; iwaiting_until_min_rpc_time()) continue; if (p->suspended_via_gui) continue; if (p->trickle_up_pending) { return p; } } return 0; } // Return the best project to fetch work from, NULL if none // // Pick the one with largest (long term debt - amount of current work) // // PRECONDITIONS: // - work_request_urgency and work_request set for all projects // - CLIENT_STATE::overall_work_fetch_urgency is set // (by previous call to compute_work_requests()) // PROJECT* CLIENT_STATE::next_project_need_work() { PROJECT *p, *p_prospect = NULL; unsigned int i; for (i=0; iwork_request_urgency == WORK_FETCH_DONT_NEED) continue; if (p->work_request == 0) continue; if (!p->contactable()) continue; // if we don't need work, only get work from non-cpu intensive projects. // if (overall_work_fetch_urgency == WORK_FETCH_DONT_NEED && !p->non_cpu_intensive) continue; // if we don't really need work, // and we don't really need work from this project, pass. // if (overall_work_fetch_urgency == WORK_FETCH_OK) { if (p->work_request_urgency <= WORK_FETCH_OK) { continue; } } if (p_prospect) { if (p->work_request_urgency == WORK_FETCH_OK && p_prospect->work_request_urgency > WORK_FETCH_OK ) { continue; } if (p->long_term_debt + p->cpu_shortfall < p_prospect->long_term_debt + p_prospect->cpu_shortfall && !p->non_cpu_intensive ) { continue; } } p_prospect = p; } if (p_prospect && (p_prospect->work_request <= 0)) { p_prospect->work_request = 1.0; if (log_flags.work_fetch_debug) { msg_printf(0, MSG_INFO, "[work_fetch_debug] next_project_need_work: project picked %s", p_prospect->project_name ); } } return p_prospect; } // Write a scheduler request to a disk file, // to be sent to a scheduling server // int CLIENT_STATE::make_scheduler_request(PROJECT* p) { char buf[1024]; MIOFILE mf; unsigned int i; RESULT* rp; int retval; double disk_total, disk_project; get_sched_request_filename(*p, buf); FILE* f = boinc_fopen(buf, "wb"); double trs = total_resource_share(); double rrs = runnable_resource_share(); double prrs = potentially_runnable_resource_share(); double resource_share_fraction, rrs_fraction, prrs_fraction; if (trs) { resource_share_fraction = p->resource_share / trs; } else { resource_share_fraction = 1; } if (rrs) { rrs_fraction = p->resource_share / rrs; } else { rrs_fraction = 1; } if (prrs) { prrs_fraction = p->resource_share / prrs; } else { prrs_fraction = 1; } // if hostid is zero, rpc_seqno better be also // if (!p->hostid) { p->rpc_seqno = 0; } if (!f) return ERR_FOPEN; mf.init_file(f); fprintf(f, "\n" " %s\n" " %d\n" " %d\n" " %s\n" " %d\n" " %d\n" " %d\n" " %f\n" " %f\n" " %f\n" " %f\n" " %f\n" " %f\n", p->authenticator, p->hostid, p->rpc_seqno, p->anonymous_platform?"anonymous":platform_name, core_client_version.major, core_client_version.minor, core_client_version.release, p->work_request, resource_share_fraction, rrs_fraction, prrs_fraction, time_until_work_done(p, proj_min_results(p, prrs)-1, prrs), p->duration_correction_factor ); if (p->anonymous_platform) { fprintf(f, " \n"); for (i=0; iproject != p) continue; avp->write(mf); } fprintf(f, " \n"); } if (strlen(p->code_sign_key)) { fprintf(f, " \n%s\n", p->code_sign_key); } // send working prefs // fprintf(f, "\n"); global_prefs.write(mf); fprintf(f, "\n"); // send master global preferences if present and not host-specific // if (!global_prefs.host_specific && boinc_file_exists(GLOBAL_PREFS_FILE_NAME)) { FILE* fprefs = fopen(GLOBAL_PREFS_FILE_NAME, "r"); if (fprefs) { copy_stream(fprefs, f); fclose(fprefs); } PROJECT* pp = lookup_project(global_prefs.source_project); if (pp && strlen(pp->email_hash)) { fprintf(f, "%s\n", pp->email_hash ); } } // Of the projects with same email hash as this one, // send the oldest cross-project ID. // Use project URL as tie-breaker. // PROJECT* winner = p; for (i=0; iemail_hash, p->email_hash)) continue; if (project->user_create_time < winner->user_create_time) { winner = project; } else if (project->user_create_time == winner->user_create_time) { if (strcmp(project->master_url, winner->master_url) < 0) { winner = project; } } } fprintf(f, "%s\n", winner->cross_project_id ); retval = time_stats.write(mf, true); if (retval) return retval; retval = net_stats.write(mf); if (retval) return retval; // update hardware info, and write host info // host_info.get_host_info(); retval = host_info.write(mf); if (retval) return retval; // get and write disk usage // total_disk_usage(disk_total); project_disk_usage(p, disk_project); fprintf(f, " \n" " %f\n" " %f\n" " \n", disk_total, disk_project ); // report results // p->nresults_returned = 0; for (i=0; iproject == p && rp->ready_to_report) { p->nresults_returned++; rp->write(mf, true); } } read_trickle_files(p, f); // report sticky files as needed // for (i=0; iproject != p) continue; if (!fip->report_on_rpc) continue; if (fip->marked_for_delete) continue; fprintf(f, " \n" " %s\n" " %f\n" " %d\n" " \n" " \n", fip->name, fip->nbytes, fip->status ); } // send names of results in progress for this project // fprintf(f, "\n"); for (i=0; iproject == p && !rp->ready_to_report) { fprintf(f, " \n" " %s\n" " \n", rp->name ); } } fprintf(f, "\n"); // send summary of in-progress results // to give scheduler info on our CPU commitment // fprintf(f, "\n"); for (i=0; iestimated_cpu_time_remaining(); if (x == 0) continue; fprintf(f, " \n" " %f\n" " %f\n" " \n", rp->report_deadline, x ); } fprintf(f, "\n"); fprintf(f, "\n"); fclose(f); return 0; } // find a project with finished results that should be reported. // This means: // - we're not backing off contacting the project // - the result is ready_to_report (compute done; files uploaded) // - we're either within a day of the report deadline, // or at least work_buf_min_days time has elapsed since // result was completed, // or we have a sporadic connection // PROJECT* CLIENT_STATE::find_project_with_overdue_results() { unsigned int i; RESULT* r; for (i=0; iproject; if (p->waiting_until_min_rpc_time()) continue; if (p->suspended_via_gui) continue; if (!r->ready_to_report) continue; if (net_status.have_sporadic_connection) { return p; } double cushion = std::max(REPORT_DEADLINE_CUSHION, work_buf_min()); if (gstate.now > r->report_deadline - cushion) { return p; } if (gstate.now > r->completed_time + work_buf_min()) { return p; } if (gstate.now > r->report_deadline - work_buf_min()) { return p; // Handles the case where the report is due before the next reconnect is // likely. } } return 0; } // the fraction of time a given CPU is working for BOINC // double CLIENT_STATE::overall_cpu_frac() { double running_frac = time_stats.on_frac * time_stats.active_frac * time_stats.cpu_efficiency; if (running_frac < 0.01) running_frac = 0.01; if (running_frac > 1) running_frac = 1; return running_frac; } // the expected number of CPU seconds completed by the client // in a second of wall-clock time. // May be > 1 on a multiprocessor. // double CLIENT_STATE::avg_proc_rate() { return ncpus*overall_cpu_frac(); } // estimate wall-clock time until the number of uncompleted results // for project p will reach k, // given the total resource share of a set of competing projects // double CLIENT_STATE::time_until_work_done( PROJECT *p, int k, double subset_resource_share ) { int num_results_to_skip = k; double est = 0; // total up the estimated time for this project's unstarted // and partially completed results, // omitting the last k // for (vector::reverse_iterator iter = results.rbegin(); iter != results.rend(); iter++ ) { RESULT *rp = *iter; if (rp->project != p || rp->state > RESULT_FILES_DOWNLOADED || rp->ready_to_report ) continue; if (num_results_to_skip > 0) { --num_results_to_skip; continue; } if (rp->project->non_cpu_intensive) { // if it is a non_cpu intensive project, // it needs only one at a time. // est = max(rp->estimated_cpu_time_remaining(), work_buf_min()); } else { est += rp->estimated_cpu_time_remaining(); } } if (log_flags.work_fetch_debug) { msg_printf(NULL, MSG_INFO, "[work_fetch_debug] time_until_work_done(): est %f ssr %f apr %f prs %f", est, subset_resource_share, avg_proc_rate(), p->resource_share ); } if (subset_resource_share) { double apr = avg_proc_rate()*p->resource_share/subset_resource_share; return est/apr; } else { return est/avg_proc_rate(); // TODO - fix } } // Top-level function for work fetch policy. // Outputs: // - overall_work_fetch_urgency // - for each contactable project: // - work_request and work_request_urgency // // Notes: // - at most 1 CPU-intensive project will have a nonzero work_request // and a work_request_urgency higher than DONT_NEED. // This prevents projects with low LTD from getting work // even though there was a higher LTD project that should get work. // - all non-CPU-intensive projects that need work // and are contactable will have a work request of 1. // // return false // bool CLIENT_STATE::compute_work_requests() { unsigned int i; static double last_time = 0; if (gstate.now - last_time >= 60) { gstate.request_work_fetch("timer"); } if (!must_check_work_fetch) return 0; if (log_flags.work_fetch_debug) { msg_printf(0, MSG_INFO, "[work_fetch_debug] compute_work_requests(): start"); } last_time = gstate.now; must_check_work_fetch = false; adjust_debts(); rr_simulation(); // compute per-project and overall urgency // bool possible_deadline_miss = false; bool project_shortfall = false; bool non_cpu_intensive_needs_work = false; for (i=0; i< projects.size(); i++) { PROJECT* p = projects[i]; if (p->non_cpu_intensive) { if (p->runnable() || !p->contactable()) { p->work_request = 0; p->work_request_urgency = WORK_FETCH_DONT_NEED; } else { p->work_request = 1.0; p->work_request_urgency = WORK_FETCH_NEED_IMMEDIATELY; non_cpu_intensive_needs_work = true; if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch_debug] non-CPU-intensive project needs work" ); } return false; } } else { p->work_request_urgency = WORK_FETCH_DONT_NEED; p->work_request = 0; if (p->rr_sim_deadlines_missed) { possible_deadline_miss = true; } if (p->cpu_shortfall && p->long_term_debt > -global_prefs.cpu_scheduling_period_minutes * 60) { project_shortfall = true; } } } if (cpu_shortfall <= 0.0 && (possible_deadline_miss || !project_shortfall)) { overall_work_fetch_urgency = WORK_FETCH_DONT_NEED; } else if (no_work_for_a_cpu()) { overall_work_fetch_urgency = WORK_FETCH_NEED_IMMEDIATELY; } else if (cpu_shortfall > 0) { overall_work_fetch_urgency = WORK_FETCH_NEED; } else { overall_work_fetch_urgency = WORK_FETCH_OK; } if (log_flags.work_fetch_debug) { msg_printf(0, MSG_INFO, "[work_fetch_debug] compute_work_requests(): cpu_shortfall %f, overall urgency %s", cpu_shortfall, urgency_name(overall_work_fetch_urgency) ); } if (overall_work_fetch_urgency == WORK_FETCH_DONT_NEED) { if (non_cpu_intensive_needs_work) { overall_work_fetch_urgency = WORK_FETCH_NEED_IMMEDIATELY; } return false; } // loop over projects, and pick one to get work from // double prrs = potentially_runnable_resource_share(); PROJECT *pbest = NULL; for (i=0; inon_cpu_intensive) continue; if (!p->contactable()) { if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch_debug] work fetch: project not contactable"); } continue; } if (p->deadlines_missed && overall_work_fetch_urgency != WORK_FETCH_NEED_IMMEDIATELY ) { if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch_debug] project has %d deadline misses", p->deadlines_missed ); } continue; } if (p->some_download_stalled()) { if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch_debug] project has stalled download" ); } continue; } if (p->some_result_suspended()) { if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch_debug] project has suspended result"); } continue; } if (p->overworked() && overall_work_fetch_urgency < WORK_FETCH_NEED) { if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch_debug] project is overworked"); } continue; } if (p->cpu_shortfall == 0.0 && overall_work_fetch_urgency < WORK_FETCH_NEED) { if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch_debug] project has no shortfall"); } continue; } // see if this project is better than our current best // if (pbest) { // avoid getting work from a project in deadline trouble // if (p->deadlines_missed && !pbest->deadlines_missed) { if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch_debug] project has deadline misses, %s doesn't", pbest->get_project_name() ); } continue; } // avoid getting work from an overworked project // if (p->overworked() && !pbest->overworked()) { if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch_debug] project is overworked, %s isn't", pbest->get_project_name() ); } continue; } // get work from project with highest LTD // if (pbest->long_term_debt + pbest->cpu_shortfall > p->long_term_debt + p->cpu_shortfall) { if (log_flags.work_fetch_debug) { msg_printf(p, MSG_INFO, "[work_fetch_debug] project has less LTD than %s", pbest->get_project_name() ); } continue; } } pbest = p; if (log_flags.work_fetch_debug) { msg_printf(pbest, MSG_INFO, "[work_fetch_debug] best project so far"); } } if (pbest) { pbest->work_request = max( pbest->cpu_shortfall, cpu_shortfall * (prrs ? pbest->resource_share/prrs : 1) ); if (!pbest->nearly_runnable()) { pbest->work_request_urgency = WORK_FETCH_NEED_IMMEDIATELY; } else if (pbest->cpu_shortfall) { pbest->work_request_urgency = WORK_FETCH_NEED; } else { pbest->work_request_urgency = WORK_FETCH_OK; } if (log_flags.work_fetch_debug) { msg_printf(pbest, MSG_INFO, "[work_fetch_debug] compute_work_requests(): work req %f, shortfall %f, urgency %s\n", pbest->work_request, pbest->cpu_shortfall, urgency_name(pbest->work_request_urgency) ); } } else if (non_cpu_intensive_needs_work) { overall_work_fetch_urgency = WORK_FETCH_NEED_IMMEDIATELY; } return false; } // called from the client's polling loop. // initiate scheduler RPC activity if needed and possible // bool CLIENT_STATE::scheduler_rpc_poll() { PROJECT *p; bool action=false; static double last_time=0; // check only every 5 sec, unless there's a tentative (new) project // if (!have_tentative_project() && gstate.now - last_time < 5.0) return false; last_time = gstate.now; switch(scheduler_op->state) { case SCHEDULER_OP_STATE_IDLE: if (scheduler_op->check_master_fetch_start()) { action = true; break; } p = next_project_sched_rpc_pending(); if (p) { scheduler_op->init_op_project(p, p->sched_rpc_pending); action = true; break; } if (network_suspended) break; p = next_project_trickle_up_pending(); if (p) { scheduler_op->init_op_project(p, RPC_REASON_TRICKLE_UP); action = true; break; } // report overdue results // p = find_project_with_overdue_results(); if (p) { scheduler_op->init_op_project(p, RPC_REASON_RESULTS_DUE); action = true; break; } if (!(exit_when_idle && contacted_sched_server)) { scheduler_op->init_get_work(); if (scheduler_op->state != SCHEDULER_OP_STATE_IDLE) { break; } } break; default: scheduler_op->poll(); if (scheduler_op->state == SCHEDULER_OP_STATE_IDLE) { action = true; } break; } return action; } // Handle the reply from a scheduler // int CLIENT_STATE::handle_scheduler_reply( PROJECT* project, char* scheduler_url, int& nresults ) { SCHEDULER_REPLY sr; FILE* f; int retval; unsigned int i; bool signature_valid, update_global_prefs=false, update_project_prefs=false; char buf[256], filename[256]; std::string old_gui_urls = project->gui_urls; PROJECT* p2; nresults = 0; contacted_sched_server = true; project->last_rpc_time = now; get_sched_reply_filename(*project, filename); f = fopen(filename, "r"); if (!f) return ERR_FOPEN; retval = sr.parse(f, project); fclose(f); if (retval) return retval; if (log_flags.sched_ops) { if (sr.scheduler_version) { msg_printf(project, MSG_INFO, "Scheduler RPC succeeded [server version %d]", sr.scheduler_version ); } else { msg_printf(project, MSG_INFO, "Scheduler RPC succeeded"); } } // check that master URL is correct // if (strlen(sr.master_url)) { canonicalize_master_url(sr.master_url); if (strcmp(sr.master_url, project->master_url)) { msg_printf(project, MSG_ERROR, "You used the wrong URL for this project" ); msg_printf(project, MSG_ERROR, "The correct URL is %s", sr.master_url ); p2 = gstate.lookup_project(sr.master_url); if (p2) { msg_printf(project, MSG_INFO, "You seem to be attached to this project twice" ); msg_printf(project, MSG_INFO, "We suggest that you detach projects named %s,", project->project_name ); msg_printf(project, MSG_INFO, "then reattach to %s", sr.master_url ); } else { msg_printf(project, MSG_INFO, "Using the wrong URL can cause problems in some cases." ); msg_printf(project, MSG_INFO, "When convenient, detach this project, then reattach to %s", sr.master_url ); } } } // make sure we don't already have a project of same name // if (project->tentative) { bool dup_name = false; for (i=0; iproject_name, project->project_name)) { dup_name = true; break; } } if (dup_name) { msg_printf(project, MSG_ERROR, "Already attached to a project named %s (possibly with wrong URL)", project->project_name ); msg_printf(project, MSG_ERROR, "Consider detaching this project, then trying again" ); } } // on the off chance that this is the initial RPC for a project // being attached, copy messages to a safe place // for (i=0; iset_min_rpc_time(x, "project is down"); } return ERR_PROJECT_DOWN; } // if the scheduler reply includes global preferences, // insert extra elements, write to disk, and parse // if (sr.global_prefs_xml) { // skip this if we have host-specific prefs // and we're talking to an old scheduler // if (!global_prefs.host_specific || sr.scheduler_version >= 507) { retval = save_global_prefs( sr.global_prefs_xml, project->master_url, scheduler_url ); if (retval) { return retval; } update_global_prefs = true; } else { if (log_flags.sched_op_debug) { msg_printf(project, MSG_INFO, "ignoring prefs from old server; we have host-specific prefs" ); } } } // see if we have a new venue from this project // (this must go AFTER the above, since otherwise // global_prefs_source_project() is meaningless) // if (strcmp(project->host_venue, sr.host_venue)) { safe_strcpy(project->host_venue, sr.host_venue); msg_printf(project, MSG_INFO, "New host venue: %s", sr.host_venue); update_project_prefs = true; if (project == global_prefs_source_project()) { strcpy(main_host_venue, sr.host_venue); update_global_prefs = true; } } if (update_global_prefs) { read_global_prefs(); } // deal with project preferences (should always be there) // If they've changed, write to account file, // then parse to get our venue, and pass to running apps // if (sr.project_prefs_xml) { if (strcmp(project->project_prefs.c_str(), sr.project_prefs_xml)) { project->project_prefs = string(sr.project_prefs_xml); update_project_prefs = true; } } // the account file has GUI URLs and project prefs. // rewrite if either of these has changed // if (project->gui_urls != old_gui_urls || update_project_prefs) { retval = project->write_account_file(); if (retval) { msg_printf(project, MSG_ERROR, "Can't write account file: %s", boincerror(retval) ); return retval; } } if (update_project_prefs) { project->parse_account_file(); if (strlen(project->host_venue)) { project->parse_account_file_venue(); } project->parse_preferences_for_user_files(); active_tasks.request_reread_prefs(project); } // if the scheduler reply includes a code-signing key, // accept it if we don't already have one from the project. // Otherwise verify its signature, using the key we already have. // if (sr.code_sign_key) { if (!strlen(project->code_sign_key)) { safe_strcpy(project->code_sign_key, sr.code_sign_key); } else { if (sr.code_sign_key_signature) { retval = verify_string2( sr.code_sign_key, sr.code_sign_key_signature, project->code_sign_key, signature_valid ); if (!retval && signature_valid) { safe_strcpy(project->code_sign_key, sr.code_sign_key); } else { msg_printf(project, MSG_ERROR, "New code signing key doesn't validate" ); } } else { msg_printf(project, MSG_ERROR, "Missing code sign key signature" ); } } } // copy new entities to client state // for (i=0; iname ); delete app; } else { apps.push_back(app); } } } FILE_INFO* fip; for (i=0; imerge_info(sr.file_infos[i]); } else { fip = new FILE_INFO; *fip = sr.file_infos[i]; retval = link_file_info(project, fip); if (retval) { msg_printf(project, MSG_ERROR, "Can't handle file %s in scheduler reply", fip->name ); delete fip; } else { file_infos.push_back(fip); } } } for (i=0; iname ); fip->marked_for_delete = true; } } for (i=0; iclear_errors(); continue; } avp = new APP_VERSION; *avp = sr.app_versions[i]; retval = link_app_version(project, avp); if (retval) { msg_printf(project, MSG_ERROR, "Can't handle application version %s %d in scheduler reply", avp->app_name, avp->version_num ); delete avp; continue; } app_versions.push_back(avp); } for (i=0; iproject = project; int vnum = choose_version_num(wup, sr); if (vnum < 0) { msg_printf(project, MSG_ERROR, "Can't find application version for task %s", wup->name ); delete wup; continue; } wup->version_num = vnum; retval = link_workunit(project, wup); if (retval) { msg_printf(project, MSG_ERROR, "Can't handle task %s in scheduler reply", wup->name ); delete wup; continue; } wup->clear_errors(); workunits.push_back(wup); } for (i=0; iname ); delete rp; continue; } results.push_back(rp); rp->state = RESULT_NEW; nresults++; } // update records for ack'ed results // for (i=0; igot_server_ack = true; } else { msg_printf(project, MSG_ERROR, "Got ack for task %s, but can't find it", sr.result_acks[i].name ); } } // handle result abort requests // for (i=0; iabort_task(ERR_ABORTED_BY_PROJECT, "aborted by project"); } else { rp->abort_inactive(ERR_ABORTED_BY_PROJECT); } } } for (i=0; iabort_inactive(ERR_ABORTED_BY_PROJECT); } } } // remove acked trickle files // if (sr.message_ack) { remove_trickle_files(project); } if (sr.send_file_list) { project->send_file_list = true; } project->sched_rpc_pending = 0; project->trickle_up_pending = false; // handle delay request // if (sr.request_delay) { double x = gstate.now + sr.request_delay; project->set_min_rpc_time(x, "requested by project"); } else { project->min_rpc_time = 0; } if (sr.next_rpc_delay) { project->next_rpc_time = gstate.now + sr.next_rpc_delay; } else { project->next_rpc_time = 0; } // The project returns a hostid only if it has created a new host record. // In that case reset RPC seqno // if (sr.hostid) { if (project->hostid) { // if we already have a host ID for this project, // we must have sent it a stale seqno, // which usually means our state file was copied from another host. // So generate a new host CPID. // generate_new_host_cpid(); msg_printf(project, MSG_INFO, "Generated new host CPID: %s", host_info.host_cpid ); } //msg_printf(project, MSG_INFO, "Changing host ID from %d to %d", project->hostid, sr.hostid); project->hostid = sr.hostid; project->rpc_seqno = 0; } if (sr.auto_update.present) { if (!sr.auto_update.validate_and_link(project)) { auto_update = sr.auto_update; } } project->link_project_files(true); set_client_state_dirty("handle_scheduler_reply"); if (log_flags.state_debug) { msg_printf(0, MSG_INFO, "[state_debug] handle_scheduler_reply(): State after handle_scheduler_reply():" ); print_summary(); } return 0; } double CLIENT_STATE::work_needed_secs() { double total_work = 0; for(unsigned int i=0; iproject->non_cpu_intensive) continue; total_work += results[i]->estimated_cpu_time_remaining(); } double x = work_buf_min() * avg_proc_rate() - total_work; if (x < 0) { return 0; } return x; } void CLIENT_STATE::scale_duration_correction_factors(double factor) { if (factor <= 0) return; for (unsigned int i=0; iduration_correction_factor *= factor; } if (log_flags.cpu_sched_debug) { msg_printf(NULL, MSG_INFO, "[cpu_sched_debug] scaling 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"); } } } const char *BOINC_RCSID_d35a4a7711 = "$Id$";