/* If we continue to read it without having heard an update,
* we start to make the peer's value approach decayto, getting
* half of the way there each time. */
- if (remote->awol >= REMOTE_AWOL_THRESHOLD) {
+ if (remote->awol >= MESH_REMOTE_AWOL_THRESHOLD) {
printlog(LOG_WARN, "AWOL remote limiter detected.\n");
remote->rate += ((decayto - remote->rate) / 2);
} else {
}
*aggregate += comm->local_rate;
} else if (comm->comm_fabric == COMM_GOSSIP) {
+ int i;
+ int threshold = GOSSIP_REMOTE_AWOL_THRESHOLD;
double value = 0;
value = (comm->gossip.value / comm->gossip.weight);
value *= (comm->remote_node_count + 1);
/* Keep around the last value so that we don't stupidly pick 0 when
* we're negative. If we pick 0, it looks to the limiter like it
* has free reign and it will take 100% of the rate allocation for
- * itself. */
+ * itself. This is a lie. Open question what to do here... FIXME: Use decayto?*/
if (value <= 0) {
//*aggregate = comm->gossip.last_nonzero;
*aggregate = 0;
- //printf("*****Gossip value is %.3f (%u) ((%d))\n", value, *aggregate, (int) *aggregate);
+ printlog(LOG_DEBUG, "Gossip: Read aggregate of 0 from comm layer.\n");
} else {
*aggregate = value;
comm->gossip.last_nonzero = *aggregate;
- //printf("Gossip value is %.3f (%u) ((%d))\n", value, *aggregate, (int) *aggregate);
+ printlog(LOG_DEBUG, "Gossip: Read aggregate of %.3f from comm layer.\n", value);
+ }
+
+ for (i = 0; i < comm->remote_node_count; ++i) {
+ if (comm->remote_limiters[i].awol == threshold) {
+ /* Re-claim any value/weight sent. */
+ comm->gossip.value += comm->remote_limiters[i].outgoing.saved_value;
+ comm->gossip.weight += comm->remote_limiters[i].outgoing.saved_weight;
+
+ comm->remote_limiters[i].outgoing.saved_value = 0.0;
+ comm->remote_limiters[i].outgoing.saved_weight = 0.0;
+
+ comm->remote_limiters[i].awol += 1;
+ } else if (comm->remote_limiters[i].awol < threshold) {
+ comm->remote_limiters[i].awol += 1;
+ }
}
} else {
printlog(LOG_CRITICAL, "read_comm: Invalid comm fabric: %d.\n",
}
pthread_mutex_unlock(&comm->lock);
- //printf("read: %.3f\n", *aggregate);
-
return 0;
}
sigemptyset(&signal_mask);
sigaddset(&signal_mask, SIGHUP);
+ sigaddset(&signal_mask, SIGUSR1);
+ sigaddset(&signal_mask, SIGUSR2);
pthread_sigmask(SIG_BLOCK, &signal_mask, NULL);
pthread_setcanceltype(PTHREAD_CANCEL_DEFERRED, NULL);