Laboratoire de l'Informatique et du Parallélisme » Benchmark Grid5000

#!/usr/bin/python

import sys

import tempfile

import time

from execo_g5k import *

from execo import *

OUTDIR="/home/dponsard/DFS_EXECO/RESULT"

def usage():

  print "Usage %s job site dfs_type" % sys.argv[0]

  sys.exit(1)

def start_benchmark(client_nodes,bench_size=4096):

  prepare=Remote("sync && echo 3 > /proc/sys/vm/drop_caches && echo 3 > /proc/sys/vm/drop_caches", client_nodes, connexion_params={'user': 'root'})

  bench=Remote("TMP_DIR=`mktemp -d /data/BENCH.XXXX` ;dd if=/dev/zero of=${TMP_DIR}/write.dd bs=1M count=%d; sync" % bench_size, client_nodes, connexion_params={'user': 'root'})

  prepare.run()

  bench.run()

  cumul_speed = 0

  for p in bench.processes():

    cumul_speed += bench_size/float(p.end_date()-p.start_date())

  print "Benchmark speed is "+str(cumul_speed)

  print "Benchmark output"

  for p in bench.processes():

    print p.stdout() 

  return cumul_speed

def create_dfs5k_conf(dfs_type,site,job,master_node,client_nodes,data_nodes):

  try:

    f=tempfile.NamedTemporaryFile(prefix="dfs5k_%s_%s_%s-" % (dfs_type,site,job), delete=False)

  except Exception as e: 

    print "Failed to open /tmp/dfs5k.conf"

    print e

    sys.exit(1)   

  if dfs_type == "ceph" or dfs_type=="lustre":

    f.write("""name: DFS5k

options:

dataDir: /tmp

master: root@%s

clients:

""" % master_node.address)

    for client_node in client_nodes:

      f.write("  root@%s\n" % client_node.address)

    f.write("dataNodes:\n")

    for data_node in data_nodes:

      f.write("  root@%s\n" % data_node.address)

  elif dfs_type == "glusterfs":

    f.write("""name: DFS5k

options: stripe %d

master: root@%s

clients:

""" % (len(data_nodes),master_node.address))

    for client_node in client_nodes:

      f.write("  root@%s\n" % client_node.address)

    f.write("dataNodes:\n")

    for data_node in data_nodes:

      f.write("  root@%s:/tmp\n" % data_node.address)

  else:

    print "Error: Unsupported dfs type %s" % dfs_type

    sys.exit(1)

  f.close() 

  return f

def start_localfs_bench(site,job,node,bench_size=4096):

  prepare=SshProcess("sync && echo 3 > /proc/sys/vm/drop_caches && echo 3 > /proc/sys/vm/drop_caches", node, connexion_params={'user': 'root'})

  bench=SshProcess("dd if=/dev/zero of=/tmp/write.dd bs=1M count=%d && sync" % bench_size, node, connexion_params={'user': 'root'})

  prepare.run()

  start=time.time()

  bench.run()

  end=time.time()

  print "Benchmark output"

  print bench.stdout()

  print "Cleanning /tmp"

  clean=SshProcess("rm -rf /tmp/write.dd",node, connexion_params={'user': 'root'})

  clean.run()

  speed = bench_size/float(end-start)

  print "Benchmark speed is "+str(speed)

  return speed

def clean_nodes_before(dfs_type,site,dfs_conf,nodes):

  print "Cleaning /data"

  Remote("rm -rf /data/* 2&>/dev/null", nodes, connexion_params={'user': 'root'}).run()

  print "Umounting DFS5k"

  print "Umounting DFS5k"

  SshProcess("dfs5k -d -a umount -s %s -f %s" % (dfs_type,dfs_conf.name) , Host(site)).run()

  if dfs_type != "lustre":

        print "UnDeploying DFS5k" 

        SshProcess("dfs5k -d -a undeploy -s %s -f %s" % (dfs_type,dfs_conf.name) , Host(site)).run()

  else:

        print "UnDeploying DFS5k" 

        p=SshProcess("dfs5k -d -a undeploy -s %s -f %s" % (dfs_type,dfs_conf.name) , Host(site), close_stdin=False).start()

        p._process.stdin.write("y\n")

        p._process.stdin.flush()

        p.wait()

  print "Cleaning /tmp and /data"

  Remote("rm -rf /tmp/* /data/* 2&>/dev/null", nodes, connexion_params={'user': 'root'}).run()

  return

def clean_nodes_after(dfs_type,site,dfs_conf,nodes):

  print "Cleaning /data"

  Remote("rm -rf /data/* 2&>/dev/null", nodes, connexion_params={'user': 'root'}).run()

  print "Umounting DFS5k"

  SshProcess("dfs5k -d -a umount -s %s -f %s" % (dfs_type,dfs_conf.name) , Host(site)).run()

  print "UnDeploying DFS5k"

  SshProcess("dfs5k -d -a undeploy -s %s -f %s" % (dfs_type,dfs_conf.name) , Host(site)).run()

  print "Cleaning /tmp and /data"

  Remote("rm -rf /tmp/* /data/* 2&>/dev/null", nodes, connexion_params={'user': 'root'}).run()

  return

def start_dfs_bench(dfs_type,site,job,master_node,client_nodes,data_nodes):

  f = create_dfs5k_conf(dfs_type,site,job,master_node,client_nodes,data_nodes)

  print "Copying dfs5k configuration file" 

  Process("scp %s %s:/tmp/" % (f.name,site)).run()

  print "Copying SSH key to master node" 

  SshProcess("scp ~/.ssh/id_rsa root@%s:.ssh/" % master_node.address, Host(site)).run()

  clean_nodes_before(dfs_type, site, f, data_nodes+client_nodes+[master_node])

  print "Deploying DFS5k" 

  SshProcess("dfs5k -d -a deploy -s %s -f %s" % (dfs_type,f.name) , Host(site)).run()

  print "Mounting DFS5k" 

  print "Deploying DFS5k"

  SshProcess("dfs5k -d -a mount -s %s -f %s" % (dfs_type,f.name) , Host(site)).run()

  print "Performing benchmark %s" % dfs_type

  print "Master node is %s" % master_node

  print "## %d Data nodes are %s" % (len(data_nodes), data_nodes)

  print "## %d Client nodes are %s" % (len(client_nodes), client_nodes)

  speed = start_benchmark(client_nodes)

  clean_nodes_after(dfs_type, site, f, data_nodes+client_nodes+[master_node])

  return speed

def prepare_nodes(job,site,dfs_type):

  nodes = []

  try:

    if get_oar_job_info(job,site)["state"]!="Running":

      print "Cannot deploy on job %s @ %s, job not started" % (job,site)

    else: 

      nodes += get_oar_job_nodes(job,site)

  except Exception as e:

    print "Cannot get info on job %s @ %s" % (job,site)

    print e

  try:

    print "Deploying on %s" % nodes

    if dfs_type=="lustre":

      print "Deploying on %s  environment centos-x64-lustre for %s" % (nodes, dfs_type)

      (ok_nodes, failed_nodes) = deploy(Deployment(nodes,env_name="centos-x64-lustre"), node_connexion_params={'user': 'root'})

    else:

      print "Deploying on %s  environment squeeze-x64-nfs for %s" % (nodes, dfs_type)

      (ok_nodes, failed_nodes) = deploy(Deployment(nodes,env_name="squeeze-x64-nfs"))

    print "Deployed on %s " % ok_nodes

  except Exception as e:

    print "Failed to deploy"

    print e

    raise e

    sys.exit(1)

  else: 

    #print "Installing iozone"

    #Remote("apt-get update && apt-get install -y iozone3",ok_nodes, connexion_params={'user': 'root'}).run()

    return list(ok_nodes)

if len(sys.argv) < 4:

  usage()

job = int(sys.argv[1])

site = sys.argv[2]

dfs_type= sys.argv[3]

f_result = open(OUTDIR+"/results_%s_%s_%s.txt" % (job,site,dfs_type),"w")

nodes=prepare_nodes(job,site,dfs_type)

#nodes=["A","B","C","D","E","F","G","H","I","J","K","L","M","N","O","P"]

print nodes

master_node=nodes[0]

#Write first line of the table

line="X"

for i in range(2,len(nodes)-1):

  line+=" %d" % i

line+="\n"

f_result.write(line)

print "#### START LOCALFS BENCH #####"

speed = start_localfs_bench(site,job,master_node)

line="0"

for i in range(2,len(nodes)-1):

  line+=" %d" % int(speed)

line+="\n"

f_result.write(line)

for n_clients in [1,2,4,6,8,10,12,14,16,18,20]:

  f_result.write("%d" % n_clients)

  available_nodes = nodes[1:]

  client_nodes=available_nodes[:n_clients]

  available_nodes=available_nodes[n_clients:]

  for n_datanodes in range(2,len(available_nodes)+1):

    data_nodes=available_nodes[:n_datanodes]

    print "\n\n"

    print "#### START DFS BENCH #####"

    print master_node,client_nodes,data_nodes

    print "#### ############### #####"

    speed = start_dfs_bench(dfs_type,site,job,master_node,client_nodes,data_nodes)

    f_result.write(" %d" % int(speed))

    f_result.flush()

  f_result.write("\n")

  f_result.flush()

f_result.write("\n\n\n")

f_result.write(str(nodes))

f_result.close()