// 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 . // 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 "str_util.h" #include "util.h" #include "client_msgs.h" #include "scheduler_op.h" #include "sandbox.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) #ifndef SIM // 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, sizeof(buf)); FILE* f = boinc_fopen(buf, "wb"); if (!f) return ERR_FOPEN; 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; } mf.init_file(f); fprintf(f, "\n" " %s\n" " %d\n" " %d\n" " %d\n" " %d\n" " %d\n" " %f\n" " %f\n" " %f\n" " %f\n" " %f\n" " %d\n", p->authenticator, p->hostid, p->rpc_seqno, core_client_version.major, core_client_version.minor, core_client_version.release, resource_share_fraction, rrs_fraction, prrs_fraction, work_fetch.estimated_delay, p->duration_correction_factor, g_use_sandbox?1:0 ); work_fetch.write_request(p, f); // write client capabilities // fprintf(f, " 1\n" ); write_platforms(p, mf); // send supported app_versions for anonymous platform clients // 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->cpid_time < winner->cpid_time) { winner = project; } else if (project->cpid_time == winner->cpid_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; // can't return without closing file 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, config.suppress_net_info); //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 ); if (coprocs.coprocs.size()) { fprintf(f, " \n"); for (i=0; iwrite_xml(mf); } fprintf(f, " \n"); } // 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 ); } // NOTE: there's also a send_file_list flag, not currently used if (p->send_time_stats_log) { fprintf(f, "\n"); time_stats.get_log_after(p->send_time_stats_log, mf); fprintf(f, "\n"); } if (p->send_job_log) { fprintf(f, "\n"); job_log_filename(*p, buf, sizeof(buf)); send_log_after(buf, p->send_job_log, mf); fprintf(f, "\n"); } // 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_time_remaining(false); if (x == 0) continue; fprintf(f, " \n" " %s\n" " %f\n" " %f\n" " \n", rp->name, rp->report_deadline, x ); } fprintf(f, "\n"); fprintf(f, "\n"); fclose(f); return 0; } // 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; static double last_work_fetch_time = 0; // check only every 5 sec // if (now - last_time < 5.0) return false; last_time = now; switch(scheduler_op->state) { case SCHEDULER_OP_STATE_IDLE: if (scheduler_op->check_master_fetch_start()) { action = true; break; } // check for various reasons to contact particular projects. // If we need to contact a project, // see if we should ask it for work as well. // p = next_project_sched_rpc_pending(); if (p) { work_fetch.compute_work_request(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) { work_fetch.compute_work_request(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) { work_fetch.compute_work_request(p); scheduler_op->init_op_project(p, RPC_REASON_RESULTS_DUE); action = true; break; } // should we check work fetch? Do this at most once/minute if (exit_when_idle && contacted_sched_server) break; if (tasks_suspended) break; if (must_check_work_fetch) { last_work_fetch_time = 0; } if (now - last_work_fetch_time < 60) return false; must_check_work_fetch = false; last_work_fetch_time = now; p = work_fetch.choose_project(); if (p) { scheduler_op->init_op_project(p, RPC_REASON_NEED_WORK); action = true; 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) { 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; vectornew_results; contacted_sched_server = true; project->last_rpc_time = now; get_sched_reply_filename(*project, filename, sizeof(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) { msg_printf(project, MSG_INFO, "Scheduler request completed: got %d new tasks", (int)sr.results.size() ); } if (log_flags.sched_op_debug) { if (sr.scheduler_version) { msg_printf(project, MSG_INFO, "[sched_op_debug] Server version %d", sr.scheduler_version ); } } // 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_USER_ERROR, "You used the wrong URL for this project" ); msg_printf(project, MSG_USER_ERROR, "The correct URL is %s", sr.master_url ); p2 = 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 // bool dup_name = false; for (i=0; iproject_name, project->project_name)) { dup_name = true; break; } } if (dup_name) { msg_printf(project, MSG_USER_ERROR, "Already attached to a project named %s (possibly with wrong URL)", project->project_name ); msg_printf(project, MSG_USER_ERROR, "Consider detaching this project, then trying again" ); } // show messages from server // 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 computer location: %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_INTERNAL_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_INTERNAL_ERROR, "New code signing key doesn't validate" ); } } else { msg_printf(project, MSG_INTERNAL_ERROR, "Missing code sign key signature" ); } } } // copy new entities to client state // for (i=0; iuser_friendly_name, sr.apps[i].user_friendly_name); } else { app = new APP; *app = sr.apps[i]; retval = link_app(project, app); if (retval) { msg_printf(project, MSG_INTERNAL_ERROR, "Can't handle application %s in scheduler reply", app->name ); 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_INTERNAL_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; iavg_ncpus = avpp.avg_ncpus; avp->max_ncpus = avpp.max_ncpus; avp->flops = avpp.flops; strcpy(avp->cmdline, avpp.cmdline); avp->coprocs.delete_coprocs(); avp->coprocs = avpp.coprocs; strlcpy(avp->api_version, avpp.api_version, sizeof(avp->api_version)); // if we had download failures, clear them // avp->clear_errors(); continue; } avp = new APP_VERSION; *avp = avpp; retval = link_app_version(project, avp); if (retval) { delete avp; continue; } app_versions.push_back(avp); } for (i=0; iproject = project; retval = link_workunit(project, wup); if (retval) { msg_printf(project, MSG_INTERNAL_ERROR, "Can't handle task %s in scheduler reply", wup->name ); delete wup; continue; } wup->clear_errors(); workunits.push_back(wup); } double est_duration = 0; for (i=0; iname ); delete rp; continue; } if (strlen(rp->platform) == 0) { strcpy(rp->platform, get_primary_platform()); rp->version_num = latest_version(rp->wup->app, rp->platform); } rp->avp = lookup_app_version( rp->wup->app, rp->platform, rp->version_num, rp->plan_class ); if (!rp->avp) { msg_printf(project, MSG_INTERNAL_ERROR, "No app version for result: %s %d", rp->platform, rp->version_num ); delete rp; continue; } rp->wup->version_num = rp->version_num; results.push_back(rp); new_results.push_back(rp); rp->set_state(RESULT_NEW, "handle_scheduler_reply"); est_duration += rp->estimated_duration(false); } if (log_flags.sched_op_debug) { if (sr.results.size()) { msg_printf(project, MSG_INFO, "[sched_op_debug] estimated total job duration: %.0f seconds", est_duration ); } } // update records for ack'ed results // for (i=0; igot_server_ack = true; } else { msg_printf(project, MSG_INTERNAL_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 - no longer usable" ); } else { rp->abort_inactive(ERR_ABORTED_BY_PROJECT); } } else { msg_printf(project, MSG_INTERNAL_ERROR, "Server requested abort of unknown task %s", sr.result_abort[i].name ); } } for (i=0; iabort_inactive(ERR_ABORTED_BY_PROJECT); } } else { msg_printf(project, MSG_INTERNAL_ERROR, "Server requested conditional abort of unknown task %s", sr.result_abort_if_not_started[i].name ); } } // remove acked trickle files // if (sr.message_ack) { remove_trickle_files(project); } if (sr.send_file_list) { project->send_file_list = true; } project->send_time_stats_log = sr.send_time_stats_log; project->send_job_log = sr.send_job_log; project->sched_rpc_pending = 0; project->trickle_up_pending = false; // 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 computer cross-project ID: %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(); } // the following must precede the backoff and request_delay checks, // since it overrides them // if (sr.next_rpc_delay) { project->next_rpc_time = now + sr.next_rpc_delay; } else { project->next_rpc_time = 0; } work_fetch.handle_reply(project, new_results); project->nrpc_failures = 0; project->min_rpc_time = 0; if (sr.request_delay) { double x = now + sr.request_delay; project->set_min_rpc_time(x, "requested by project"); } return 0; } #endif // SIM 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) return; if (next_rpc_time && (future_time > next_rpc_time)) return; min_rpc_time = future_time; possibly_backed_off = true; if (log_flags.sched_op_debug) { msg_printf(this, MSG_INFO, "[sched_op_debug] Deferring communication for %s", timediff_format(min_rpc_time - gstate.now).c_str() ); msg_printf(this, MSG_INFO, "[sched_op_debug] 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 // PROJECT* CLIENT_STATE::next_project_sched_rpc_pending() { unsigned int i; PROJECT* p; for (i=0; inext_rpc_time && p->next_rpc_timesched_rpc_pending = RPC_REASON_PROJECT_REQ; p->next_rpc_time = 0; } else { if (p->waiting_until_min_rpc_time()) continue; } // 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; } // 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 within a day of the report deadline, // or at least a day has elapsed since the 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; iready_to_report) continue; PROJECT* p = r->project; if (p->waiting_until_min_rpc_time()) continue; if (p->suspended_via_gui) continue; if (config.report_results_immediately) { return p; } 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 + SECONDS_PER_DAY) { return p; } } return 0; } // trigger work fetch // void CLIENT_STATE::request_work_fetch(const char* where) { if (log_flags.work_fetch_debug) { msg_printf(0, MSG_INFO, "[work_fetch_debug] Request work fetch: %s", where); } must_check_work_fetch = true; } const char *BOINC_RCSID_d35a4a7711 = "$Id$";