gs_smart_netpm.c

Go to the documentation of this file.

#include "config.h"
#ifdef GS_SMART_GRIDSOLVE
#include "gs_smart_netpm.h"
#include "comm_encode.h"


/*
 * 
 * This function generates a network performance model based on the 
 * list of servers received from the agent.
 *
 * @param (out) netpm -- network performance model.
 * @param (in) server_list -- list of servers received from agent.
 * @param (in) nb_servers -- number of servers in the server list.
 */

int gs_smart_netpm_construct_netpm(gs_smart_netpm * netpm,
                                    gs_server_t ** server_list, int nb_servers){
  
  if((!netpm) || (!server_list)) return -1;

  if(gs_smart_netpm_aggregate_ping_info(server_list, nb_servers)<0){
    ERRPRINTF("SMART: Error agrgregating ping info\n");
    return -1;
  }

  if(gs_smart_netpm_measure_client_links_perf(server_list, nb_servers)<0){
    ERRPRINTF("SMART: Error measuring client links of network pm\n");
    return -1;
  }
 
  if(gs_smart_netpm_construct_nodes(netpm, server_list, nb_servers)<0){
    ERRPRINTF("SMART: Error constructing nodes of network performance model\n");
    return -1;
  }
  
  if(gs_smart_netpm_construct_links(netpm, server_list, nb_servers)<0){
    ERRPRINTF("SMART: Error constructing links of network performance model\n");
    return -1;
  }

  if(gs_smart_netpm_add_bw_to_links(netpm, server_list, nb_servers)<0){
    ERRPRINTF("SMART: Error adding bandwidth links of network pm\n");
    return -1;
  }

  return 0;
}


/*
 * 
 * This function constructs each network performance model node.
 *
 * @param (out) netpm -- network performance model.
 * @param (in) server_list -- list of servers received from agent.
 * @param (in) nb_servers -- number of servers in the server list.
 */


int gs_smart_netpm_construct_nodes(gs_smart_netpm * netpm,
                                   gs_server_t ** server_list, int nb_servers){

  gs_server_t * this_server=NULL;
  gs_smart_netpm_node * netpm_node;
  int  i;
  
  if((!netpm) || (!server_list) || (nb_servers<=0)) return -1;
  
  netpm->nb_nodes=nb_servers;

  /*
   * Construct a single client node
   */  
  netpm->netpm_clnt_node=(gs_smart_netpm_node *)
                     calloc(1, sizeof(gs_smart_netpm_node ));

  if(!netpm->netpm_clnt_node) return -1;

  netpm->netpm_clnt_node->node_type=GS_SMART_NETPM_CLIENT_NODE;

  /*
   * Construct array of server nodes
   */
  netpm->netpm_nodes=(gs_smart_netpm_node **)
                     calloc(netpm->nb_nodes, sizeof(gs_smart_netpm_node *));
  
  if(!netpm->netpm_nodes) return -1;


  for(i=0; i<netpm->nb_nodes; i++){

    /*
     * Set up each server node..
     */    
    netpm->netpm_nodes[i]=(gs_smart_netpm_node *)
                           calloc(1, sizeof(gs_smart_netpm_node));
   
    if(!netpm->netpm_nodes[i]) return -1;

    netpm_node=netpm->netpm_nodes[i];
    netpm_node->id=i;
    netpm_node->node_type=GS_SMART_NETPM_SERVER_NODE;

    if(!server_list[i]) return -1;

    netpm_node->server=server_list[i];
    this_server = netpm_node->server;
  }

  return 0;
}




/*
 * 
 * This function constructs the links between network nodes.
 *
 * @param (out) netpm -- network performance model.
 * @param (in) server_list -- list of servers received from agent.
 * @param (in) nb_servers -- number of servers in the server list.
 */





int gs_smart_netpm_construct_links(gs_smart_netpm * netpm,
                                   gs_server_t ** server_list, int nb_servers){

  gs_smart_netpm_node * netpm_node, * netpm_clnt_node, * dest_netpm_node;
  gs_server_t * this_server=NULL;
  int i, j;
  int alloc_links=0; 
 
  if((!netpm) || (!server_list) || (nb_servers<=0)) return -1;
  if(!netpm->netpm_clnt_node) return -1;


  /*
   * Construct client node links.
   * There is one client node link for each server
   * in the network..
   */ 
  netpm_clnt_node=netpm->netpm_clnt_node;
  netpm_clnt_node->nb_links=nb_servers;
  netpm_clnt_node->netpm_links=(gs_smart_netpm_link **)
                              calloc(nb_servers, sizeof(gs_smart_netpm_link *));

  if(!netpm_clnt_node->netpm_links) return -1;

  /*
   * Set up server node links
   */
  for(i=0; i<netpm->nb_nodes;i++){
    if(!netpm->netpm_nodes[i]) return -1;
    netpm_node=netpm->netpm_nodes[i];

    if(!netpm_node->server) return -1;
    this_server = netpm_node->server;
    

 
    if(this_server->smart==1){
     /*
      * If this server node is a smart server node
      * then the links connecting it to other smart servers
      * need to be constructed (One for each smart server).
      */   
      netpm_node->nb_links=this_server->nb_smart_servers;
      netpm_node->netpm_links=(gs_smart_netpm_link **)
                   calloc(netpm_node->nb_links, sizeof(gs_smart_netpm_link *));
 
      if(!netpm_node->netpm_links) return -1;

      /*
       * Constructing server node link to client
       */
      netpm_node->netpm_links[0]=(gs_smart_netpm_link *)
                                       calloc(1, sizeof(gs_smart_netpm_link ));

      if(!netpm_node->netpm_links[0]) return -1;
 
      /*
       * Set up links to other servers 
       * For each two-way smart link between servers,
       * memory is only allocated once 
       */  
      int link_node=0;
      for(j=0; j<alloc_links; j++){
        if(!netpm->netpm_nodes[j]) return -1;
        if(!netpm->netpm_nodes[j]->server) return -1;

        while(netpm->netpm_nodes[link_node]->server->smart==0){
          if(link_node>netpm->nb_nodes) return -1;
          link_node++;
        }
       
        if(!netpm->netpm_nodes[link_node]) return -1;
        dest_netpm_node=netpm->netpm_nodes[link_node];
        
        netpm_node->netpm_links[j+1]=(gs_smart_netpm_link *)
                                    calloc(1, sizeof(gs_smart_netpm_link));
        
        if(!netpm_node->netpm_links[j+1]) return -1;

        /*
         * Pointing this source sever node link to
         * another destination server. 
         */ 
        netpm_node->netpm_links[j+1]->linked_nodeA=netpm_node;
        netpm_node->netpm_links[j+1]->linked_nodeB=dest_netpm_node;
 
        /*
         * Letting the corresponding link of the destination node
         * equal the link of the source node.
         */ 
        dest_netpm_node->netpm_links[alloc_links]=netpm_node->netpm_links[j+1];
        link_node++;
      }
      alloc_links++;
    }
    else{

      /*
       * If this server is not a smart server then only one link
       * needs to be set up. That link which points to the client.
       */ 
      netpm_node->nb_links=1;
      netpm_node->netpm_links=(gs_smart_netpm_link **)
                              calloc(1, sizeof(gs_smart_netpm_link *));

      if(!netpm_node->netpm_links) return -1;
 
      netpm_node->netpm_links[0]=(gs_smart_netpm_link *)
                                    calloc(1, sizeof(gs_smart_netpm_link));

      if(!netpm_node->netpm_links[0]) return -1;
    }

    /*
     * Pointing the client link  (link 0) of each server node
     * back to the client. 
     */ 
    netpm_node->netpm_links[0]->linked_nodeA=netpm_node;
    netpm_node->netpm_links[0]->linked_nodeB=netpm_clnt_node;
    
    /*
     * Letting the corresponding link from the client node equal the client link of 
     * this server node.
     */ 
    netpm_clnt_node->netpm_links[i]=netpm_node->netpm_links[0];
  }

  return 0;
}



/*
 * 
 * This function adds the bandwidth values to each link in the network 
 * performance model.
 *
 * @param (out) netpm -- network performance model.
 * @param (in) server_list -- list of servers received from agent.
 * @param (in) nb_servers -- number of servers in the server list.
 */


int gs_smart_netpm_add_bw_to_links(gs_smart_netpm * netpm,
                                   gs_server_t ** server_list, int nb_servers){
  gs_server_t * this_server=NULL, * search_server=NULL;
  gs_smart_netpm_node * netpm_node;
  int i, j, k;
  int link_num=0;
  double ping_byte_size, bw;
  double avg_s_bw, avg_c_bw, s_sum, c_sum;
  int nnzs, nnzc;
  int nb_smart_servers=0;

  if((!netpm) || (!server_list) || (nb_servers<=0)) return -1;

  s_sum = c_sum = avg_c_bw = avg_s_bw = 0.0;
  nnzs = nnzc = 0;

  ping_byte_size=(32*1024);
  for(i=0; i<netpm->nb_nodes;i++){

    if(!netpm->netpm_nodes[i]) return -1;
    netpm_node=netpm->netpm_nodes[i];

    if(!netpm_node->server) return -1;
    this_server = netpm_node->server;

    /*
     * Adding bandwidth value to the links (link 0)
     * pointing from server node to the client link..
     */ 
    if(!netpm_node->netpm_links[0]) return -1;
    netpm_node->netpm_links[0]->bw=this_server->comm_bw;

    if(this_server->comm_bw != 0.0) {
      c_sum += this_server->comm_bw;
      nnzc++;
    }


    /*
     * Adding bandwidth value from smart server links 
     * pointing from smart servers to other smart servers.
     * This involves examining the aggregated ping info
     * and using the componentid value to determine
     * which ping info below to which links.
     */ 

    if(this_server->smart==1){
      for(j=0;j<this_server->nb_my_pings ;j++){
      link_num=1;
        for(k=0; k<netpm->nb_nodes; k++){
          if(!netpm->netpm_nodes[k]) return -1;
          if(!netpm->netpm_nodes[k]->server) return -1;
     
          search_server=netpm->netpm_nodes[k]->server;

          if(!this_server->my_pings[j]) return -1;
          if((!search_server->componentid) || 
             (!this_server->my_pings[j]->componentid) ) return -1;

          if(memcmp(search_server->componentid, 
                    this_server->my_pings[j]->componentid, CID_LEN)==0){

            bw =(ping_byte_size/this_server->my_pings[j]->comm_time); 
            if(bw<0) bw=0.0;


            if(!netpm_node->netpm_links[link_num]) return -1;

            netpm_node->netpm_links[link_num]->bw=bw;
            if(bw != 0.0){
              s_sum+=bw; 
              nnzs++;
            }
              


          }
          if((search_server->smart==1) && (i!=k)) link_num++;
        }
      }
    nb_smart_servers++;
    }
  }
  if(nb_smart_servers<2){
    netpm->avg_s_bw=0.0;
  }
  else{
    if(nnzs > 0){
      netpm->avg_s_bw=s_sum/(double) nnzs;
    }
    if(netpm->avg_s_bw == 0.0)
      netpm->avg_s_bw = 20000.0;
  }

  if(nnzc > 0){
    netpm->avg_c_bw=c_sum/(double) nnzc;
  }
  if(netpm->avg_c_bw == 0.0)
    netpm->avg_c_bw = 20000.0;

  return 0;
}




/*
 *
 * This function aggregates all the ping info of each server into a 
 * single structure (all_pings) which is stored as part of the task graph.
 *
 *
 * Each smart server only stores the ping info for only a subset of the links 
 * connecting it to all other servers. This is to avoid duplicate 
 * pings, i.e if server A pings server B to get the  ping info for the link 
 * connecting A to B, B does not ping server A to get the ping info for the 
 * same link.
 *
 *
 * @param (out) netpm -- network performance model.
 * @param (in) server_list -- list of servers received from agent.
 * @param (in) nb_servers -- number of servers in the server list.
 */

int gs_smart_netpm_aggregate_ping_info(gs_server_t ** server_list, 
                                                     int nb_servers){
  gs_server_pings_t * ping_info1, * ping_info2;
  int nb_smart_servers=0, i, j,k,l;
  int arr_cnt;

  if((!server_list) || (nb_servers<=0)){
    ERRPRINTF("SMART: Server List is NULL\n");
    return -1;
  }

 /*
  * Setting up the my_ping for this server
  * This structure holds the its subset of links connecting 
  * it to all other servers.
  */
  for(i=0;i<nb_servers;i++){
    if(!server_list[i]) return -1;
    if(server_list[i]->smart==1){
 
      if(!server_list[i]->my_ping_str) return -1;

      if(gs_smart_decode_ping_info(server_list[i]->my_ping_str,
                    &server_list[i]->my_pings,&server_list[i]->nb_my_pings)<0){
          ERRPRINTF("SMART: Error decoding smart info\n");
          return -1;
       }
       if(server_list[i]->my_ping_str) free(server_list[i]->my_ping_str);
       nb_smart_servers++;
    }
  }

  /*
   * Setting up the component ids in the all_pings info structure.
   * This structure is used to aggregate all the info from the my_ping
   * structures of all the servers into a single structure.
   */

  for(i=0;i<nb_servers;i++){
    server_list[i]->nb_smart_servers=nb_smart_servers;
    server_list[i]->all_pings=(gs_server_pings_t **)
                        calloc( nb_smart_servers, sizeof(gs_server_pings_t *));

    if(!server_list[i]->all_pings){
      ERRPRINTF("SMART: Error allocating all ping\n");
      return -1;
    }

    arr_cnt=0;
    for(j=0;j<nb_servers;j++){
      if(server_list[j]->smart==1){
        server_list[i]->all_pings[arr_cnt]=(gs_server_pings_t *)
                                        calloc(1, sizeof(gs_server_pings_t));

        if (!server_list[i]->all_pings[arr_cnt]) {
          ERRPRINTF("SMART: Cannot malloc all ping %d\n", i);
          return -1;
        }

        if(memcpy( server_list[i]->all_pings[arr_cnt]->componentid,
                            server_list[j]->componentid, CID_LEN)<0){
          ERRPRINTF("SMART: Error copying component ids\n");
          return -1;
        }

        if(!server_list[i]->all_pings[arr_cnt]->componentid){
          ERRPRINTF("SMART: Error copying component id data to ping data\n");
          return -1;
        }

        server_list[i]->all_pings[arr_cnt]->comm_time=0.0;
        arr_cnt++;
      }
    }
  }



 /*
  * For each smart server add the comm time of their own my_ping structure
  */ 

 for(i=0;i<nb_servers;i++){                  
   if(!server_list[i]) return -1; 
   if( server_list[i]->smart==1){
     for(j=0;j< server_list[i]->nb_my_pings;j++){      
       for(k=0;k<nb_smart_servers;k++){ 
               
         if((!server_list[i]->all_pings[k]) ||
            (!server_list[i]->my_pings[j])) return -1;

         if((!server_list[i]->all_pings[k]->componentid) || 
            (!server_list[i]->my_pings[j]->componentid)) return -1;

         if(memcmp(server_list[i]->all_pings[k]->componentid, 
                   server_list[i]->my_pings[j]->componentid, CID_LEN)==0){ 
           server_list[i]->all_pings[k]->comm_time=server_list[i]->my_pings[j]->comm_time;
         }
       }
     }
   }
 }


 /*
  * For each smart server add the comm time of from the my_ping 
  * structure of the other smart servers. This completes
  * the all_ping structure for each server.
  */ 


 for(i=0;i<nb_servers;i++){                    
   if(!server_list[i]) return -1; 
   if( server_list[i]->smart==1){
     for(j=0;j< server_list[i]->nb_my_pings;j++){      

       for(k=0;k<nb_servers;k++){ 

         if(!server_list[k]) return -1; 
             
         if( server_list[k]->smart==1){

           if((!server_list[k]->componentid) || 
              (!server_list[i]->my_pings[j])) return -1;

           if(!(server_list[i]->my_pings[j]->componentid)) return -1;
 
           if(memcmp(server_list[k]->componentid, 
                    server_list[i]->my_pings[j]->componentid, CID_LEN)==0){ 


             for(l=0;l<nb_smart_servers;l++){     
         
               if((!server_list[i]->componentid) ||
                  (!server_list[k]->all_pings[l])) return -1;

               if(!server_list[k]->all_pings[l]->componentid) return -1;
               
               if(memcmp(server_list[i]->componentid, 
                         server_list[k]->all_pings[l]->componentid, CID_LEN)==0){  
                 ping_info1=server_list[k]->all_pings[l];
                 ping_info2=server_list[i]->my_pings[j];
                 ping_info1->comm_time=ping_info2->comm_time;
               }
             }
           }
         }
       }
     }
   }
 }
 return 0;
}





/*
 * Given the list of all the servers, this function pings the servers and
 * updates the bandwidth of the client links.
 *
 * @param (inout) server_list -- list of servers received from agent.
 * @param (in) nb_servers -- number of servers in the server list.
*/

int
gs_smart_netpm_measure_client_links_perf(gs_server_t ** server_list, int nb_servers)
{

  int i, len, ns, nping;
  char *msg;
  int gs_max_ping=5;
  int grpc_measure_comm_num_servers = gs_max_ping;
  double elapsed_time, *comm_bw;
  char *srv_cid;
  char *cid_dup;
  srv_cid=(char *)calloc(2*CID_LEN+1, sizeof(char));
  icl_hash_t *grpc_comm_cache = NULL;
  ns = MIN(nb_servers, grpc_measure_comm_num_servers);
  len = 32 * 1024;

  msg = (char *)calloc(len, 1);

  if(!msg)
    return -1;

  nping = 0;

  for(i=0;i<nb_servers;i++) {

    proxy_cid_to_str(srv_cid, server_list[i]->componentid);

    if(grpc_comm_cache)
      comm_bw = (double *) icl_hash_find(grpc_comm_cache, srv_cid);
    else
      comm_bw = NULL;

    if(comm_bw) {
      server_list[i]->comm_bw = *comm_bw;
    }
    else {
      if(gs_do_ping(server_list[i], msg, len, &elapsed_time) < 0) {
        server_list[i]->comm_bw = 0.0;
      }
      else {
        server_list[i]->comm_bw = len / elapsed_time;

        if(grpc_comm_cache) {
          comm_bw = (double *)malloc(sizeof(double));

          if(comm_bw) {

            cid_dup = strdup(srv_cid);

            if(cid_dup) {
              free(comm_bw);
            }
            else {
              *comm_bw = server_list[i]->comm_bw;

              icl_hash_insert(grpc_comm_cache, cid_dup, comm_bw);
            }
          }
        }
      }

      nping++;
    }

    if(nping == ns)
      break;
  }

  if(grpc_comm_cache) free(grpc_comm_cache);
  if(msg) free(msg);
  if(srv_cid) free(srv_cid);
  return 0;

}   













/*
 * This function prints the network performance model out to file
 * in the graphviz format. This function can be used to create
 * jpgs of network performance model.
 *
 * @param (in) netpm -- the network performance model
 * @param (out) filename -- output file.
*/


int 
gs_smart_net_pm_print_to_dotgraph(gs_smart_netpm * netpm, char * filename){
  gs_smart_netpm_node * netpm_node;
  gs_server_t * server=NULL;
  FILE *fp;
  int i, j;


  char *gridsolve_root;
  if((!netpm) || (!filename)) return -1;

  if((gridsolve_root = getenv("GRIDSOLVE_ROOT")) == NULL)
    gridsolve_root = GRIDSOLVE_TOP_BUILD_DIR;
  if(!gridsolve_root) {
    ERRPRINTF("Warning: GRIDSOLVE_ROOT unknown, assuming cwd.\n");
    gridsolve_root = strdup(".");
  }

  fp = fopen(filename, "w");
  if(fp<0){
    ERRPRINTF("SMART: Error opening file\n");
    return -1;
  } 

  fprintf(fp, "digraph Net_PM {\n");
  fprintf(fp, " client [shape=\"diamond\", style=\"filled\", fillcolor=\"#ADD8E6\" fontsize=\"15.0\"];\n");
  for(i=0; i<netpm->nb_nodes;i++){

    if(!netpm->netpm_nodes[i]) return -1;
    netpm_node=netpm->netpm_nodes[i];

    if(!netpm_node->server) return -1;

    server=netpm_node->server;
    if(server->smart==1){
      fprintf(fp, " \"%s_%d\" [shape=box,label=\"%s\\nport=%d\\nkflops=%d\\nworkload=%d\\nsmart=%d\",  style=filled,fillcolor=\"firebrick1\",fontsize=15.0];\n", server->hostname, server->port, server->hostname, server->port, server->kflops, server->workload,  server->smart);
        
      for(j=1;j<netpm_node->nb_links;j++){

        if(!netpm_node->netpm_links[j]) return -1;
        if((!netpm_node->netpm_links[j]->linked_nodeB) ||
           (!netpm_node->netpm_links[j]->linked_nodeA)) return -1;
        
        if(!netpm_node->netpm_links[j]->linked_nodeA->server) return -1;

        if(!netpm_node->netpm_links[j]->linked_nodeA->server->hostname) return -1;
      
           
 
        if( netpm_node->netpm_links[j]->linked_nodeB->id== netpm_node->id){
          fprintf(fp, " \"%s_%d\" -> \"%s_%d\" [label=\"bw=%f\", color=\"red\"];\n", server->hostname, server->port, netpm_node->netpm_links[j]->linked_nodeA->server->hostname, netpm_node->netpm_links[j]->linked_nodeA->server->port,  netpm_node->netpm_links[j]->bw); 
        }
      }
    }
    else{
        if(!server->hostname) return -1; 
        fprintf(fp, " \"%s_%d\" [shape=box,label=\"%s\\nport=%d\\nkflops=%d\\nworkload=%d\\nsmart=%d\",  style=filled,fillcolor=\"deepskyblue\",fontsize=15.0];\n", server->hostname, server->port, server->hostname, server->port, server->kflops, server->workload,  server->smart);


    }
    if(!server->hostname) return -1;
    if(!netpm_node->netpm_links[0]) return -1;
 
    fprintf(fp, " client -> \"%s_%d\" [label=\"bw=%f\"];\n", server->hostname, server->port, netpm_node->netpm_links[0]->bw);
 
  }

  fprintf(fp, "\n}\n");
  fclose(fp);
  return 0;
}















/*
 * This function frees the network performance model
 *
 * @param (in) netpm -- the network performance model
*/


int gs_smart_netpm_free(gs_smart_netpm * netpm){
  int i, j;
  gs_smart_netpm_node * netpm_node;
  if(!netpm) return -1;

 /*
  * Free client netpm node
  */ 
  if(netpm->netpm_clnt_node){
    netpm_node=netpm->netpm_clnt_node;
 
    if(!netpm_node->netpm_links) return -1;
    free(netpm_node->netpm_links); 
    free(netpm_node);
  }


 /*
  * Free netpm server node
  */ 

  for(i=0; i<netpm->nb_nodes;i++){
    if(!netpm->netpm_nodes[i]) return -1;
    netpm_node=netpm->netpm_nodes[i];
    for(j=0; j<netpm_node->nb_links;j++){
      if(!netpm_node->netpm_links[j]) return -1;
      free(netpm_node->netpm_links[j]); 
    }
    if(!netpm_node->netpm_links) return -1;
    free(netpm_node->netpm_links); 
    free(netpm_node);
  }  

  free(netpm);


  return 0;
}




#endif