+ new_debt = nb.get_debt();
+ }
+ if (real_load <= new_debt) {
+ load_to_send = real_load;
+ nb.set_debt(new_debt - load_to_send);
+ real_load = 0.0;
+ } else {
+ load_to_send = new_debt;
+ nb.set_debt(0.0);
+ real_load -= load_to_send;
+ }
+ if (load_to_send > 0.0)
+ comm.send(nb.get_data_mbox(), new message(message::LOAD, load_to_send));
+}
+
+void process::send_all()
+{
+ using std::tr1::bind;
+ using std::tr1::placeholders::_1;
+
+ if (opt::bookkeeping) {
+ std::for_each(neigh.begin(), neigh.end(),
+ bind(&process::send1_bookkeeping, this, _1));
+ prev_load_broadcast = expected_load;
+ } else {
+ std::for_each(neigh.begin(), neigh.end(),
+ bind(&process::send1_no_bookkeeping, this, _1));
+ prev_load_broadcast = real_load;
+ }
+ comm.flush(false);
+}
+
+void process::receive(double timeout)
+{
+ message* msg;
+ m_host_t from;
+
+ DEBUG2("%sblocking receive (%g)", "\0non-" + !timeout, timeout);
+ while (may_receive() && comm.recv(msg, from, timeout)) {
+ switch (msg->get_type()) {
+ case message::INFO: {
+ neighbor* n = rev_neigh[from];
+ n->set_load(msg->get_amount());
+ break;
+ }
+ case message::CREDIT:
+ expected_load += msg->get_amount();
+ break;
+ case message::LOAD: {
+ double ld = msg->get_amount();
+ real_load += ld;
+ if (finalizing)
+ total_load_running -= ld;
+ break;
+ }
+ case message::CTRL_CLOSE:
+ ctrl_close_pending--;
+ close_received = true;
+ break;
+ case message::DATA_CLOSE:
+ data_close_pending--;
+ close_received = true;
+ break;
+ }
+ delete msg;
+ timeout = 0.0; // only wait on first recv
+ }
+ comm.flush(false);
+}
+
+void process::finalize1(neighbor& nb)
+{
+ comm.send(nb.get_ctrl_mbox(), new message(message::CTRL_CLOSE, 0.0));
+ comm.send(nb.get_data_mbox(), new message(message::DATA_CLOSE, 0.0));
+}
+
+void process::finalize()
+{
+ using std::tr1::bind;
+ using std::tr1::placeholders::_1;
+
+ finalizing = true;
+ total_load_running -= real_load;
+
+ DEBUG2("send CLOSE to %lu neighbor%s",
+ (unsigned long )neigh.size(), ESSE(neigh.size()));
+ std::for_each(neigh.begin(), neigh.end(),
+ bind(&process::finalize1, this, _1));
+
+ DEBUG2("wait for CLOSE from %lu neighbor%s",
+ (unsigned long )neigh.size(), ESSE(neigh.size()));
+ while (may_receive()) {
+ comm.flush(false);
+ receive(-1.0);