main.cpp

mysql-5.0.22.tar.gz源码包

  
 end:
  fclose(f);
  return result;
}

bool
connect_hosts(atrt_config& config){
  for(size_t i = 0; i<config.m_hosts.size(); i++){
    if(config.m_hosts[i].m_cpcd->connect() != 0){
      g_logger.error("Unable to connect to cpc %s:%d",
		     config.m_hosts[i].m_cpcd->getHost(),
		     config.m_hosts[i].m_cpcd->getPort());
      return false;
    }
    g_logger.debug("Connected to %s:%d",
		   config.m_hosts[i].m_cpcd->getHost(),
		   config.m_hosts[i].m_cpcd->getPort());
  }
  
  return true;
}

bool
connect_ndb_mgm(atrt_process & proc){
  NdbMgmHandle handle = ndb_mgm_create_handle();
  if(handle == 0){
    g_logger.critical("Unable to create mgm handle");
    return false;
  }
  BaseString tmp = proc.m_hostname;
  tmp.appfmt(":%d", proc.m_ndb_mgm_port);

  if (ndb_mgm_set_connectstring(handle,tmp.c_str()))
  {
    g_logger.critical("Unable to create parse connectstring");
    return false;
  }

  if(ndb_mgm_connect(handle, 30, 1, 0) != -1)
  {
    proc.m_ndb_mgm_handle = handle;
    return true;
  }

  g_logger.critical("Unable to connect to ndb mgm %s", tmp.c_str());
  return false;
}

bool
connect_ndb_mgm(atrt_config& config){
  for(size_t i = 0; i<config.m_processes.size(); i++){
    atrt_process & proc = config.m_processes[i];
    if((proc.m_type & atrt_process::NDB_MGM) != 0){
      if(!connect_ndb_mgm(proc)){
	return false;
      }
    }
  }
  
  return true;
}

static int remap(int i){
  if(i == NDB_MGM_NODE_STATUS_NO_CONTACT) return NDB_MGM_NODE_STATUS_UNKNOWN;
  if(i == NDB_MGM_NODE_STATUS_UNKNOWN) return NDB_MGM_NODE_STATUS_NO_CONTACT;
  return i;
}

bool
wait_ndb(atrt_config& config, int goal){

  goal = remap(goal);


  /**
   * Get mgm handle for cluster
   */
  NdbMgmHandle handle = 0;
  for(size_t i = 0; i<config.m_processes.size(); i++){
    atrt_process & proc = config.m_processes[i];
    if((proc.m_type & atrt_process::NDB_MGM) != 0){
      handle = proc.m_ndb_mgm_handle;
      break;
    }
  }
  if(handle == 0){
    g_logger.critical("Unable to find mgm handle");
    return false;
  }

  if(goal == NDB_MGM_NODE_STATUS_STARTED){
    /**
     * 1) wait NOT_STARTED
     * 2) send start
     * 3) wait STARTED
     */
    if(!wait_ndb(config, NDB_MGM_NODE_STATUS_NOT_STARTED))
      return false;
    
    ndb_mgm_start(handle, 0, 0);
  }

  struct ndb_mgm_cluster_state * state;

  time_t now = time(0);
  time_t end = now + 360;
  int min = remap(NDB_MGM_NODE_STATUS_NO_CONTACT);
  int min2 = goal;

  while(now < end){
    /**
     * 1) retreive current state
     */
    state = 0;
    do {
      state = ndb_mgm_get_status(handle);
      if(state == 0){
	const int err = ndb_mgm_get_latest_error(handle);
	g_logger.error("Unable to poll db state: %d %s %s",
		       ndb_mgm_get_latest_error(handle),
		       ndb_mgm_get_latest_error_msg(handle),
		       ndb_mgm_get_latest_error_desc(handle));
	if(err == NDB_MGM_SERVER_NOT_CONNECTED && connect_ndb_mgm(config)){
	  g_logger.error("Reconnected...");
	  continue;
	}
	return false;
      }
    } while(state == 0);
    NdbAutoPtr<void> tmp(state);
    
    min2 = goal;
    for(int i = 0; i<state->no_of_nodes; i++){
      if(state->node_states[i].node_type == NDB_MGM_NODE_TYPE_NDB){
	const int s = remap(state->node_states[i].node_status);
	min2 = (min2 < s ? min2 : s );
	
	if(s < remap(NDB_MGM_NODE_STATUS_NO_CONTACT) || 
	   s > NDB_MGM_NODE_STATUS_STARTED){
	  g_logger.critical("Strange DB status during start: %d %d", i, min2);
	  return false;
	}

	if(min2 < min){
	  g_logger.critical("wait ndb failed node: %d %d %d %d", 
			    state->node_states[i].node_id, min, min2, goal);
	}
      }
    }
    
    if(min2 < min){
      g_logger.critical("wait ndb failed %d %d %d", min, min2, goal);
      return false;
    }
    
    if(min2 == goal){
      return true;
      break;
    }
    
    min = min2;
    now = time(0);
  }
  
  g_logger.critical("wait ndb timed out %d %d %d", min, min2, goal);
  
  return false;
}

bool
start_process(atrt_process & proc){
  if(proc.m_proc.m_id != -1){
    g_logger.critical("starting already started process: %d", proc.m_index);
    return false;
  }
  
  BaseString path = proc.m_proc.m_cwd.substr(proc.m_host->m_base_dir.length()+BaseString("/run").length());
  
  BaseString tmp = g_setup_progname;
  tmp.appfmt(" %s %s/%s/ %s",
	     proc.m_host->m_hostname.c_str(),
	     g_setup_path,
	     path.c_str(),
	     proc.m_proc.m_cwd.c_str());

  const int r1 = system(tmp.c_str());
  if(r1 != 0){
    g_logger.critical("Failed to setup process");
    return false;
  }

  {
    Properties reply;
    if(proc.m_host->m_cpcd->define_process(proc.m_proc, reply) != 0){
      BaseString msg;
      reply.get("errormessage", msg);
      g_logger.error("Unable to define process: %s", msg.c_str());      
      return false;
    }
  }
  {
    Properties reply;
    if(proc.m_host->m_cpcd->start_process(proc.m_proc.m_id, reply) != 0){
      BaseString msg;
      reply.get("errormessage", msg);
      g_logger.error("Unable to start process: %s", msg.c_str());
      return false;
    }
  }
  return true;
}

bool
start_processes(atrt_config& config, int types){
  for(size_t i = 0; i<config.m_processes.size(); i++){
    atrt_process & proc = config.m_processes[i];
    if((types & proc.m_type) != 0 && proc.m_proc.m_path != ""){
      if(!start_process(proc)){
	return false;
      }
    }
  }
  return true;
}

bool
stop_process(atrt_process & proc){
  if(proc.m_proc.m_id == -1){
    return true;
  }

  {
    Properties reply;
    if(proc.m_host->m_cpcd->stop_process(proc.m_proc.m_id, reply) != 0){
      Uint32 status;
      reply.get("status", &status);
      if(status != 4){
	BaseString msg;
	reply.get("errormessage", msg);
	g_logger.error("Unable to stop process: %s(%d)", msg.c_str(), status);
	return false;
      }
    }
  }
  {
    Properties reply;
    if(proc.m_host->m_cpcd->undefine_process(proc.m_proc.m_id, reply) != 0){
      BaseString msg;
      reply.get("errormessage", msg);
      g_logger.error("Unable to undefine process: %s", msg.c_str());      
      return false;
    }
    proc.m_proc.m_id = -1;
  }
  return true;
}

bool
stop_processes(atrt_config& config, int types){
  for(size_t i = 0; i<config.m_processes.size(); i++){
    atrt_process & proc = config.m_processes[i];
    if((types & proc.m_type) != 0){
      if(!stop_process(proc)){
	return false;
      }
    }
  }
  return true;
}

bool
update_status(atrt_config& config, int){
  
  Vector<Vector<SimpleCpcClient::Process> > m_procs;
  
  Vector<SimpleCpcClient::Process> dummy;
  m_procs.fill(config.m_hosts.size(), dummy);
  for(size_t i = 0; i<config.m_hosts.size(); i++){
    Properties p;
    config.m_hosts[i].m_cpcd->list_processes(m_procs[i], p);
  }

  for(size_t i = 0; i<config.m_processes.size(); i++){
    atrt_process & proc = config.m_processes[i];
    if(proc.m_proc.m_id != -1){
      Vector<SimpleCpcClient::Process> &h_procs= m_procs[proc.m_host->m_index];
      bool found = false;
      for(size_t j = 0; j<h_procs.size(); j++){
	if(proc.m_proc.m_id == h_procs[j].m_id){
	  found = true;
	  proc.m_proc.m_status = h_procs[j].m_status;
	  break;
	}
      }
      if(!found){
	g_logger.error("update_status: not found");
	g_logger.error("id: %d host: %s cmd: %s", 
		       proc.m_proc.m_id,
		       proc.m_hostname.c_str(),
		       proc.m_proc.m_path.c_str());
	for(size_t j = 0; j<h_procs.size(); j++){
	  g_logger.error("found: %d %s", h_procs[j].m_id, 
			 h_procs[j].m_path.c_str());
	}
	return false;
      }
    }
  }
  return true;
}

int
is_running(atrt_config& config, int types){
  int found = 0, running = 0;
  for(size_t i = 0; i<config.m_processes.size(); i++){
    atrt_process & proc = config.m_processes[i]; 
    if((types & proc.m_type) != 0){
      found++;
      if(proc.m_proc.m_status == "running")
	running++;
    }
  }
  
  if(found == running)
    return 2;
  if(running == 0)
    return 0;
  return 1;
}


int
insert(const char * pair, Properties & p){
  BaseString tmp(pair);
  
  tmp.trim(" \t\n\r");

  Vector<BaseString> split;
  tmp.split(split, ":=", 2);

  if(split.size() != 2)
    return -1;

  p.put(split[0].trim().c_str(), split[1].trim().c_str()); 

  return 0;
}

bool
read_test_case(FILE * file, atrt_testcase& tc, int& line){

  Properties p;
  int elements = 0;
  char buf[1024];
  while(!feof(file)){
    if(!fgets(buf, 1024, file))
      break;

    line++;
    BaseString tmp = buf;
    
    if(tmp.length() > 0 && tmp.c_str()[0] == '#')
      continue;
    
    if(insert(tmp.c_str(), p) != 0)
      break;
    
    elements++;
  }
  
  if(elements == 0){
    if(file == stdin){
      BaseString tmp(buf); 
      tmp.trim(" \t\n\r");
      Vector<BaseString> split;
      tmp.split(split, " ", 2);
      tc.m_command = split[0];
      if(split.size() == 2)
	tc.m_args = split[1];
      else
	tc.m_args = "";
      tc.m_max_time = 60000;
      return true;
    }
    return false;
  }

  if(!p.get("cmd", tc.m_command)){
    g_logger.critical("Invalid test file: cmd is missing near line: %d", line);
    return false;
  }
  
  if(!p.get("args", tc.m_args))
    tc.m_args = "";

  const char * mt = 0;
  if(!p.get("max-time", &mt))
    tc.m_max_time = 60000;
  else
    tc.m_max_time = atoi(mt);

  if(p.get("type", &mt) && strcmp(mt, "bench") == 0)
    tc.m_report= true;
  else
    tc.m_report= false;

  if(p.get("run-all", &mt) && strcmp(mt, "yes") == 0)
    tc.m_run_all= true;
  else
    tc.m_run_all= false;
  
  return true;
}

bool
setup_test_case(atrt_config& config, const atrt_testcase& tc){
  const int r1 = system(g_clear_progname);
  if(r1 != 0){
    g_logger.critical("Failed to clear result");
    return false;
  }
  
  size_t i = 0;
  for(; i<config.m_processes.size(); i++){
    atrt_process & proc = config.m_processes[i]; 
    if(proc.m_type == atrt_process::NDB_API || proc.m_type == atrt_process::MYSQL_CLIENT){
      proc.m_proc.m_path.assfmt("%s/bin/%s", proc.m_host->m_base_dir.c_str(),
				tc.m_command.c_str());
      proc.m_proc.m_args.assign(tc.m_args);
      if(!tc.m_run_all)
        break;
    }
  }
  for(i++; i<config.m_processes.size(); i++){
    atrt_process & proc = config.m_processes[i]; 
    if(proc.m_type == atrt_process::NDB_API || proc.m_type == atrt_process::MYSQL_CLIENT){
      proc.m_proc.m_path.assign("");
      proc.m_proc.m_args.assign("");
    }
  }
  return true;
}

bool
gather_result(atrt_config& config, int * result){
  BaseString tmp = g_gather_progname;
  for(size_t i = 0; i<config.m_processes.size(); i++){
    atrt_process & proc = config.m_processes[i]; 
    if(proc.m_proc.m_path != ""){
      tmp.appfmt(" %s:%s", 
		 proc.m_hostname.c_str(),
		 proc.m_proc.m_cwd.c_str());
    }
  }
  
  const int r1 = system(tmp.c_str());
  if(r1 != 0){
    g_logger.critical("Failed to gather result");
    return false;
  }

  const int r2 = system(g_analyze_progname);

  if(r2 == -1 || r2 == (127 << 8)){
    g_logger.critical("Failed to analyze results");
    return false;
  }
  
  * result = r2 ;
  return true;
}

bool
setup_hosts(atrt_config& config){
  const int r1 = system(g_clear_progname);
  if(r1 != 0){
    g_logger.critical("Failed to clear result");
    return false;
  }

  for(size_t i = 0; i<config.m_hosts.size(); i++){
    BaseString tmp = g_setup_progname;
    tmp.appfmt(" %s %s/ %s/run", 
	       config.m_hosts[i].m_hostname.c_str(),
	       g_setup_path,
	       config.m_hosts[i].m_base_dir.c_str());
    
    const int r1 = system(tmp.c_str());
    if(r1 != 0){
      g_logger.critical("Failed to setup %s",
			config.m_hosts[i].m_hostname.c_str());
      return false;
    }
  }
  return true;
}

template class Vector<Vector<SimpleCpcClient::Process> >;
template class Vector<atrt_host>;
template class Vector<atrt_process>;