BCP_tm_nodes_to_storage.cpp

Go to the documentation of this file.

// Copyright (C) 2007, International Business Machines
// Corporation and others.  All Rights Reserved.

#include "BCP_os.hpp"
#include "BCP_tm.hpp"
#include "BCP_tm_node.hpp"
#include "BCP_tm_user.hpp"
#include "BCP_tm_functions.hpp"

#define NUMNODES_BASED_ON_BUFSIZE

//#############################################################################

static bool BCP_tm_scan_children(BCP_tm_prob& p, BCP_tm_node* node,
                                 std::vector<BCP_tm_node*>& nodes_to_send,
                                 const long bufsize)
{
#ifndef NUMNODES_BASED_ON_BUFSIZE
  const size_t send_size = 100;
#endif
  BCP_vec<BCP_tm_node*>& children = node->_children;
  for (int i = children.size() - 1; i >= 0; --i) {
    BCP_tm_node* s = children[i];
    if (s == NULL)
      continue;
    if (BCP_tm_scan_children(p, s, nodes_to_send, bufsize))
      return true;
  }
  if (node->_data._desc.IsValid()) {
    nodes_to_send.push_back(node);
    const bool def = p.param(BCP_tm_par::ReportWhenDefaultIsExecuted);
    p.msg_buf.pack(node->_index);
    node->_data._desc->pack(p.packer, def, p.msg_buf);
    bool has_user_data = node->_data._user.IsValid();
    p.msg_buf.pack(has_user_data);
    if (has_user_data) {
      p.packer->pack_user_data(node->_data._user.GetRawPtr(), p.msg_buf);
    }
  }
#ifndef NUMNODES_BASED_ON_BUFSIZE
  return (nodes_to_send.size() >= send_size);
#else
  return (p.msg_buf.size() > bufsize);
#endif
}

//#############################################################################

static bool BCP_tm_scan_siblings(BCP_tm_prob& p, BCP_tm_node* node,
                                 std::vector<BCP_tm_node*>& nodes_to_send,
                                 const long bufsize)
{
#ifndef NUMNODES_BASED_ON_BUFSIZE
  const size_t send_size = 100;
#endif
  if (node->_data._desc.IsValid()) {
    nodes_to_send.push_back(node);
    const bool def = p.param(BCP_tm_par::ReportWhenDefaultIsExecuted);
    p.msg_buf.pack(node->_index);
    node->_data._desc->pack(p.packer, def, p.msg_buf);
    bool has_user_data = node->_data._user.IsValid();
    p.msg_buf.pack(has_user_data);
    if (has_user_data) {
      p.packer->pack_user_data(node->_data._user.GetRawPtr(), p.msg_buf);
    }
#ifndef NUMNODES_BASED_ON_BUFSIZE
    if (nodes_to_send.size() >= send_size)
#else
    if (p.msg_buf.size() > bufsize)
#endif
      {
        return true;
      }
  }
  BCP_tm_node* parent = node->_parent;
  if (parent == NULL)
    return false;
  BCP_vec<BCP_tm_node*>& siblings = parent->_children;
  for (int i = siblings.size() - 1; i >= 0; --i) {
    BCP_tm_node* s = siblings[i];
    if (s == node || s == NULL)
      continue;
    if (BCP_tm_scan_children(p, s, nodes_to_send, bufsize))
      return true;
  }
  return BCP_tm_scan_siblings(p, parent, nodes_to_send, bufsize);
}

//#############################################################################

bool BCP_tm_is_data_balanced(BCP_tm_prob& p)
{
  const int maxheap = p.param(BCP_tm_par::MaxHeapSize);
  if (maxheap == -1) {
    return true;
  }
#if 0
  // FIXME--DELETE (used to test Bonmin code)
  printf("local nodes: %i\n", BCP_tm_node::num_local_nodes);
  printf("remote nodes: %i\n", BCP_tm_node::num_remote_nodes);
#endif
//   return (BCP_tm_node::num_local_nodes < 10);

  const double usedheap = BCP_used_heap();
  if (usedheap == -1)
    return true;
  const double freeheap = maxheap - usedheap;
  printf("free: %f   used: %f   free/max: %f\n",
         freeheap, usedheap, freeheap/maxheap);
  return (freeheap > 1<<24 /* 16M */ && freeheap / maxheap > 0.15);
}

//#############################################################################

bool BCP_tm_balance_data(BCP_tm_prob& p)
{
  if (! p.need_a_TS)
    return false;

  int pid = -1;

  /* check if any of the TS processes accept data */
  int max_space = 0;
  for (std::map<int,int>::const_iterator tsi = p.ts_space.begin();
       tsi != p.ts_space.end(); ++tsi) {
    if (tsi->second > max_space) {
      max_space = tsi->second;
      pid = tsi->first;
    }
  }
  if (max_space < 1 << 22 /* 4M */) {
    pid = -1;
  }

  BCP_buffer& buf = p.msg_buf;

  if (pid == -1) {
    /* None of the TS processes accept data, we need a new one. */
    if (p.lp_scheduler.maxNodeIds() > 1) {
      pid = p.lp_scheduler.request_node_id();
      if (pid == -1) {
        return true;
      }
      std::vector<int>::iterator ipid =
        std::find(p.lp_procs.begin(), p.lp_procs.end(), pid);
      if (ipid == p.lp_procs.end()) {
        throw BCP_fatal_error("\
Trying to convert an LP into TS, but its pid is not in lp_procs!\n");
      }
      p.lp_procs.erase(ipid);
      p.ts_procs.push_back(pid);
      printf("Turning LP (#%i) into a TS\n", pid);
      BCP_tm_notify_process_type(p, BCP_ProcessType_TS, 1, &pid);
    }
  }

  if (pid == -1)
    return true;

  /* OK we can balance */

  /*
    FIXME: For now we just balance node data. When we run hybrid (cuts are
    FIXME: added) then we must balance cuts, too
  */
  std::vector<BCP_tm_node*> nodes_to_send;
  const std::vector<CoinTreeSiblings*>& candList =
    p.candidate_list.getTree()->getCandidates();
  BCP_tm_node* node =
    dynamic_cast<BCP_tm_node*>(candList.back()->currentNode());

  /* 'node' is a leaf. Starting from it traverse the tree until there is
     enough data to be sent off. The "enough" means that the message buffer
     takes up 25% of the free space. */
  int usedheap = BCP_used_heap();
  assert(usedheap > 0);
  const int maxheap = p.param(BCP_tm_par::MaxHeapSize);
  assert(maxheap > 0);
  int freeheap = maxheap - usedheap;
  printf("Before sending off: freeheap: %i   usedheap: %i\n",
         freeheap, usedheap);
  buf.clear();
  BCP_tm_scan_siblings(p, node, nodes_to_send, freeheap >> 2);
  int num = nodes_to_send.size();
  if (num == 0) {
    // Everything is already sent out, but we are still having memory problems
    throw BCP_fatal_error("No memory left in TM\n");
  }
  int fake_index = -1;
  buf.pack(fake_index);
  p.msg_env->send(pid, BCP_Msg_NodeList, buf);
  buf.clear();
  int saved = 0, fm = 0;
  p.msg_env->receive(pid, BCP_Msg_NodeListReply, buf, -1);
  buf.unpack(saved);
  buf.unpack(fm);
  p.ts_space[pid] = fm;
  for (int i = 0; i < saved; ++i) {
    node = nodes_to_send[i];
    node->_locally_stored = false;
    node->_data._desc = NULL;
    node->_data._user = NULL;
    node->_data_location = pid;
  }
  nodes_to_send.clear();

  usedheap = BCP_used_heap();
  freeheap = maxheap - usedheap;
  printf("After sending off: freeheap: %i   usedheap: %i\n",
         freeheap, usedheap);

  if (saved == 0) {
    // Something is wrong. The least full TS did not accept anything...
    //    sleep(10000);
    throw BCP_fatal_error("TS did not accept anything\n");
  }
  
  BCP_tm_node::num_local_nodes -= saved;
  BCP_tm_node::num_remote_nodes += saved;

#if 0
  // FIXME--DELETE (used to test Bonmin code)
  for (size_t k = 0; k < p.search_tree.size(); ++k) {
    BCP_tm_node* n = p.search_tree[k];
    if (n && n->_locally_stored) {
      assert(n->_data._desc.IsNull() && n->_data._user.IsNull());
    }
  }
#endif

  return false;
}