The scheduler isn't able to send the result because it
+requires more resources than any client has.
+
+
+
+BOINC provides a form of redundant computing
+in which each computation is performed on multiple clients,
+the results are compared,
+and are accepted only when a 'consensus' is reached.
+In some cases new results must be created and sent.
+
+BOINC manages most of the details;
+however, there are two places where the application developer gets involved:
+
+
+- Validation:
+This performs two functions.
+First, when a sufficient number (a 'quorum') of successful results
+have been returned, it compares them and sees if there is a 'consensus'.
+The method of comparing results (which may need to take into
+account platform-varying floating point arithmetic)
+and the policy for determining consensus (e.g., best two out of three)
+are supplied by the application.
+If a consensus is reached, a particular result is designated
+as the 'canonical' result.
+Second, if a result arrives after a consensus has already been reached,
+the new result is compared with the canonical result;
+this determines whether the user gets credit.
+
+
- Assimilation:
+This is the mechanism by which the project is notifed of the completion
+(success or unsuccessful) of a work unit.
+It is performed exactly once per work unit.
+If the work unit was completed successfully
+(i.e. if there is a canonical result)
+the project-supplied function reads the output file(s)
+and handles the information, e.g. by recording it in a database.
+If the workunit failed,
+the function might write an entry in a log,
+send an email, etc.
+
+
+
+In the following example,
+the project creates a workunit with
+
+min_quorum = 2
+
+target_nresults = 3
+
+max_delay = 10
+
+BOINC automatically creates three results,
+which are sent at various times.
+At time 8, two successful results have returned
+so the validater is invoked.
+It finds a consensus, so the work unit is assimilated.
+At time 10 result 3 arrives;
+validation is performed again,
+this time to check whether result 3 gets credit.
+
+time 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14
+
+ created validate; assimilate
+WU x x x
+ created sent success
+result 1 x x---------------x
+ created sent success
+result 2 x x-------------------x
+ created sent success
+result 3 x x-----------------------x
+
+
+In the next example,
+result 2 is lost (i.e., there's no reply to the BOINC scheduler).
+When result 3 arrives a consensus is found
+and the work unit is assimilated.
+At time 13 the scheduler 'gives up' on result 2
+(this allows it to delete the canonical result's output files,
+which are needed to validate late-arriving results).
+
+time 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14
+
+ created validate; assimilate
+WU x x x
+ created sent success
+result 1 x x---------------x
+ created sent lost giveup
+result 2 x x-------- x
+ created sent success
+result 3 x x-----------------------x
+
+
+In the next example,
+results 2 returns an error at time 5.
+This reduces the number of outstanding results to 2;
+because target_nresults is 3, BOINC creates another result (result 4).
+A consensus is reached at time 9, before result 4 is returned.
+
+time 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14
+
+ created validate; assimilate
+WU x x x
+ created sent success
+result 1 x x---------------x
+ created sent error
+result 2 x x-------x
+ created sent success
+result 3 x x-------------------x
+ created sent success
+result 4 x x----------------------x
+
+";
+page_tail();
+?>
diff --git a/sched/feeder.C b/sched/feeder.C
index ffcc593d8f..b0eca165a5 100644
--- a/sched/feeder.C
+++ b/sched/feeder.C
@@ -26,10 +26,6 @@
//
// -asynch fork and run in a separate process
-// TODO:
-// - check for wu/results that don't get sent for a long time;
-// generate a warning message
-
// Trigger files:
// The feeder program periodically checks for two trigger files:
//
@@ -86,8 +82,8 @@ int check_reread_trigger() {
return 0;
}
-// Try keep the wu_results array filled.
-// This is actually a little tricky.
+// Try to keep the work array filled.
+// This is a little tricky.
// We use an enumerator.
// The inner loop scans the wu_result table,
// looking for empty slots and trying to fill them in.
@@ -102,11 +98,154 @@ int check_reread_trigger() {
// Crude approach: if a "collision" (as above) occurred on
// a pass through the array, wait a long time (5 sec)
//
-void feeder_loop() {
- int i, j, nadditions, ncollisions, retval;
+// Checking for infeasible results (i.e. can't sent to any host):
+// - the "infeasible_count" field of WU_RESULT keeps track of
+// how many times the WU_RESULT was infeasible for a host
+// - the scheduler gives priority to results that have infeasible_count > 0
+// - the feeder tries to ensure that the number of WU_RESULTs
+// with infeasible_count > 0 doesn't exceed MAX_INFEASIBLE
+// (compiled into feeder).
+// If it does, then the feeder picks the WU_RESULT with
+// the largest infeasible_count,
+// flags the result as OVER with outcome COULDNT_SEND,
+// flags the WU for the transitioner,
+// and repeats this until the infeasible count is low enough again
+
+static void scan_work_array(
+ DB_RESULT& result, char* clause,
+ int& nadditions, int& ncollisions, int& ninfeasible,
+ bool& no_wus
+) {
+ int i, j, retval;
+ DB_WORKUNIT wu;
+ bool collision, restarted_enum = false;
+
+ for (i=0; inwu_results; i++) {
+ WU_RESULT& wu_result = ssp->wu_results[i];
+ if (wu_result.present) {
+ if (wu_result.infeasible_count > 0) {
+ ninfeasible++;
+ }
+ } else {
+try_again:
+ retval = result.enumerate(clause);
+ if (retval) {
+
+ // if we already restarted the enum on this array scan,
+ // there's no point in doing it again.
+ //
+ if (restarted_enum) {
+ log_messages.printf(SchedMessages::DEBUG, "already restarted enum on this array scan\n");
+ break;
+ }
+
+ // restart the enumeration
+ //
+ restarted_enum = true;
+ retval = result.enumerate(clause);
+ log_messages.printf(SchedMessages::DEBUG, "restarting enumeration\n");
+ if (retval) {
+ log_messages.printf(SchedMessages::DEBUG, "enumeration restart returned nothing\n");
+ no_wus = true;
+ break;
+ }
+ }
+
+ // there's a chance this result was sent out
+ // after the enumeration started.
+ // So read it from the DB again
+ //
+ retval = result.lookup_id(result.id);
+ if (retval) {
+ log_messages.printf(SchedMessages::NORMAL,
+ "[%s] can't reread result: %d\n", result.name, retval
+ );
+ goto try_again;
+ }
+ if (result.server_state != RESULT_SERVER_STATE_UNSENT) {
+ log_messages.printf(
+ SchedMessages::NORMAL,
+ "[%s] RESULT STATE CHANGED\n",
+ result.name
+ );
+ goto try_again;
+ }
+ collision = false;
+ for (j=0; jnwu_results; j++) {
+ if (ssp->wu_results[j].present
+ && ssp->wu_results[j].result.id == result.id
+ ) {
+ ncollisions++;
+ collision = true;
+ break;
+ }
+ }
+ if (!collision) {
+ log_messages.printf(
+ SchedMessages::NORMAL,
+ "[%s] adding result in slot %d\n",
+ result.name, i
+ );
+ retval = wu.lookup_id(result.workunitid);
+ if (retval) {
+ log_messages.printf(
+ SchedMessages::CRITICAL,
+ "[%s] can't read workunit #%d: %d\n",
+ result.name, result.workunitid, retval
+ );
+ continue;
+ }
+ wu_result.result = result;
+ wu_result.workunit = wu;
+ wu_result.present = true;
+ wu_result.infeasible_count = 0;
+ nadditions++;
+ }
+ }
+ }
+}
+
+int remove_most_infeasible() {
+ int i, max, imax=-1, retval;
DB_RESULT result;
DB_WORKUNIT wu;
- bool no_wus, collision, restarted_enum;
+
+ max = 0;
+ for (i=0; inwu_results; i++) {
+ WU_RESULT& wu_result = ssp->wu_results[i];
+ if (wu_result.present && wu_result.infeasible_count > max) {
+ imax = i;
+ max = wu_result.infeasible_count;
+ }
+ }
+ if (max == 0) return -1; // nothing is infeasible
+
+ WU_RESULT& wu_result = ssp->wu_results[imax];
+ wu_result.present = false; // mark as absent
+ result = wu_result.result;
+ wu = wu_result.workunit;
+
+ log_messages.printf(
+ SchedMessages::NORMAL,
+ "[%s] declaring result as unsendable\n",
+ result.name
+ );
+
+ result.server_state = RESULT_SERVER_STATE_OVER;
+ result.outcome = RESULT_OUTCOME_COULDNT_SEND;
+ retval = result.update();
+ if (retval) return retval;
+ wu.transition_time = time(0);
+ retval = wu.update();
+ if (retval) return retval;
+
+ return 0;
+}
+
+void feeder_loop() {
+ int i, n, retval, nadditions, ncollisions, ninfeasible;
+ DB_RESULT result;
+ bool no_wus;
char clause[256];
sprintf(clause, "where server_state=%d order by random limit %d",
@@ -116,84 +255,22 @@ void feeder_loop() {
while (1) {
nadditions = 0;
ncollisions = 0;
+ ninfeasible = 0;
no_wus = false;
- restarted_enum = false;
- for (i=0; inwu_results; i++) {
- if (!ssp->wu_results[i].present) {
-try_again:
- retval = result.enumerate(clause);
- if (retval) {
- // if we already restarted the enum on this pass,
- // there's no point in doing it again.
- //
- if (restarted_enum) {
- log_messages.printf(SchedMessages::DEBUG, "already restarted enum on this pass\n");
- break;
- }
+ scan_work_array(
+ result, clause, nadditions, ncollisions, ninfeasible, no_wus
+ );
- // restart the enumeration
- //
- restarted_enum = true;
- retval = result.enumerate(clause);
- log_messages.printf(SchedMessages::DEBUG, "restarting enumeration\n");
- if (retval) {
- log_messages.printf(SchedMessages::DEBUG, "enumeration restart returned nothing\n");
- no_wus = true;
- break;
- }
- }
+ ssp->ready = true;
- // there's a chance this result was sent out
- // after the enumeration started.
- // So read it from the DB again
- //
- retval = result.lookup_id(result.id);
- if (retval) {
- log_messages.printf(SchedMessages::NORMAL, "can't reread result %s\n", result.name);
- goto try_again;
- }
- if (result.server_state != RESULT_SERVER_STATE_UNSENT) {
- log_messages.printf(
- SchedMessages::NORMAL,
- "[%s] RESULT STATE CHANGED\n",
- result.name
- );
- goto try_again;
- }
- collision = false;
- for (j=0; jnwu_results; j++) {
- if (ssp->wu_results[j].present
- && ssp->wu_results[j].result.id == result.id
- ) {
- ncollisions++;
- collision = true;
- break;
- }
- }
- if (!collision) {
- log_messages.printf(
- SchedMessages::NORMAL,
- "[%s] adding result in slot %d\n",
- result.name, i
- );
- retval = wu.lookup_id(result.workunitid);
- if (retval) {
- log_messages.printf(
- SchedMessages::CRITICAL,
- "[%s] can't read workunit #%d: %d\n",
- result.name, result.workunitid, retval
- );
- continue;
- }
- ssp->wu_results[i].result = result;
- ssp->wu_results[i].workunit = wu;
- ssp->wu_results[i].present = true;
- nadditions++;
- }
+ if (ninfeasible > MAX_INFEASIBLE) {
+ n = ninfeasible - MAX_INFEASIBLE;
+ for (i=0; iready = true;
if (nadditions == 0) {
log_messages.printf(SchedMessages::DEBUG, "No results added; sleeping 1 sec\n");
sleep(1);
diff --git a/sched/handle_request.C b/sched/handle_request.C
index c8c6907a8d..5b8bd6ba6c 100644
--- a/sched/handle_request.C
+++ b/sched/handle_request.C
@@ -653,39 +653,31 @@ static int update_wu_transition_time(WORKUNIT wu, time_t x) {
return 0;
}
-int send_work(
+// Make a pass through the wu/results array, sending work.
+// If "infeasible_only" is true, send only results that were
+// previously infeasible for some host
+//
+static void scan_work_array(
+ bool infeasible_only, double& seconds_to_fill, int& nresults,
SCHEDULER_REQUEST& sreq, SCHEDULER_REPLY& reply, PLATFORM& platform,
SCHED_SHMEM& ss
) {
- int i, retval, nresults = 0;
- double seconds_to_fill;
+ int i, retval;
WORKUNIT wu;
DB_RESULT result, result_copy;
- log_messages.printf(
- SchedMessages::NORMAL,
- "[HOST#%d] got request for %d seconds of work\n",
- reply.host.id, sreq.work_req_seconds
- );
-
- if (sreq.work_req_seconds <= 0) return 0;
-
- seconds_to_fill = sreq.work_req_seconds;
- if (seconds_to_fill > MAX_SECONDS_TO_SEND) {
- seconds_to_fill = MAX_SECONDS_TO_SEND;
- }
- if (seconds_to_fill < MIN_SECONDS_TO_SEND) {
- seconds_to_fill = MIN_SECONDS_TO_SEND;
- }
-
for (i=0; i0; i++) {
+ WU_RESULT& wu_result = ss.wu_results[i];
// the following should be a critical section
//
- if (!ss.wu_results[i].present) {
+ if (!wu_result.present) {
continue;
}
- wu = ss.wu_results[i].workunit;
+ if (infeasible_only && wu_result.infeasible_count==0) {
+ continue;
+ }
+ wu = wu_result.workunit;
double wu_seconds_filled = estimate_duration(wu, reply.host);
@@ -694,11 +686,12 @@ int send_work(
SchedMessages::DEBUG, "[HOST#%d] [WU#%d %s] WU is infeasible\n",
reply.host.id, wu.id, wu.name
);
+ wu_result.infeasible_count++;
continue;
}
- result = ss.wu_results[i].result;
- ss.wu_results[i].present = false;
+ result = wu_result.result;
+ wu_result.present = false;
retval = add_wu_to_reply(wu, reply, platform, ss);
if (retval) continue;
@@ -717,7 +710,10 @@ int send_work(
retval = update_wu_transition_time(wu, result.report_deadline);
if (retval) {
- log_messages.printf(SchedMessages::CRITICAL, "send_work: can't update WU transition time\n");
+ log_messages.printf(
+ SchedMessages::CRITICAL,
+ "send_work: can't update WU transition time\n"
+ );
}
// copy the result so we don't overwrite its XML fields
@@ -739,6 +735,35 @@ int send_work(
nresults++;
if (nresults == MAX_WUS_TO_SEND) break;
}
+}
+
+int send_work(
+ SCHEDULER_REQUEST& sreq, SCHEDULER_REPLY& reply, PLATFORM& platform,
+ SCHED_SHMEM& ss
+) {
+ int nresults = 0;
+ double seconds_to_fill;
+
+ log_messages.printf(
+ SchedMessages::NORMAL,
+ "[HOST#%d] got request for %d seconds of work\n",
+ reply.host.id, sreq.work_req_seconds
+ );
+
+ if (sreq.work_req_seconds <= 0) return 0;
+
+ seconds_to_fill = sreq.work_req_seconds;
+ if (seconds_to_fill > MAX_SECONDS_TO_SEND) {
+ seconds_to_fill = MAX_SECONDS_TO_SEND;
+ }
+ if (seconds_to_fill < MIN_SECONDS_TO_SEND) {
+ seconds_to_fill = MIN_SECONDS_TO_SEND;
+ }
+
+ // give priority to results that were infeasible for some other host
+ //
+ scan_work_array(true, seconds_to_fill, nresults, sreq, reply, platform, ss);
+ scan_work_array(false, seconds_to_fill, nresults, sreq, reply, platform, ss);
log_messages.printf(
SchedMessages::NORMAL, "[HOST#%d] Sent %d results\n",
diff --git a/sched/sched_shmem.h b/sched/sched_shmem.h
index 0f2665b704..72de2f6d3b 100644
--- a/sched/sched_shmem.h
+++ b/sched/sched_shmem.h
@@ -26,10 +26,14 @@
#define MAX_APPS 10
#define MAX_APP_VERSIONS 1000
#define MAX_WU_RESULTS 1000
+#define MAX_INFEASIBLE 500
+ // if # of elements in work array that were infeasible for some host
+ // exceeds this, classify some of them as COULDNT_SEND
// a workunit/result pair
struct WU_RESULT {
bool present;
+ int infeasible_count;
WORKUNIT workunit;
RESULT result;
};