deep-scheduler/sim_wrelaunch_exp.py at master · schr476/deep-scheduler · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
from sim_objs_wrelaunch import *
from modeling import *

def sim(sinfo_m, mapping_m, sching_m):
  env = simpy.Environment()
  cl = Cluster_wrelaunch(env, scher=Scher_wrelaunch(mapping_m, sching_m), **sinfo_m)
  jg = JobGen_wrelaunch(env, out=cl, **sinfo_m)
  env.run(until=cl.wait_for_alljobs)

  avg_schedload_l = []
  for i, w in enumerate(cl.w_l):
    print("w._id= {}, w.avg_load= {}".format(w._id, w.avg_load() ) )
    avg_schedload_l.append(w.avg_load() )

  njobs, njobs_relaunched = 0, 0
  sl_l, serv_sl_l = [], []
  # for jid, info in cl.jid_info_m.items():
  #   if 'fate' in info:
  #     fate = info['fate']
  #     if fate == 'finished':
  #       serv_sl_l.append(info['run_time']/info['expected_run_time'] )
  #       sl_l.append(
  #         (info['wait_time'] + info['run_time'] )/info['expected_run_time'] )
  #   if 'relaunched' in info:
  #     njobs_relaunched += 1
  for jid in range(1, sinfo_m['njob']+1):
    info = cl.jid_info_m[jid]
    serv_sl_l.append(info['run_time']/info['expected_run_time'] )
    sl_l.append(
      (info['wait_time'] + info['run_time'] )/info['expected_run_time'] )
    if 'nrelaunched' in info:
      # print("k= {}, nrelaunched= {}".format(len(info['wid_l'] ), info['nrelaunched'] ) )
      njobs_relaunched += 1
    njobs += 1
  log(INFO, "", njobs=njobs, njobs_relaunched=njobs_relaunched)

  return {
    'sl_mean': np.mean(sl_l),
    'sl_std': np.std(sl_l),
    'serv_sl_mean': np.mean(serv_sl_l),
    'load_mean': np.mean(avg_schedload_l) }

def subopt_relaunch_time(j):
  return 4*j.lifetime

def opt_relaunch_time(j):
  ESl = ET_k_n_pareto(j.k, j.k, Sl.loc, Sl.a)
  if ESl > 4:
    return math.sqrt(ESl)*j.lifetime
  else:
    return None
  # return math.sqrt(ESl)*j.lifetime

def print_optimal_d():
  def alpha_gen(ro):
    return alpha
  r = 2
  # print("ESl2= {}".format(Sl.moment(2) ) )
  # print("ESl2_pareto= {}".format(ESl2_pareto(ro, N, Cap, k, r, b, beta, a, alpha_gen, d=0, red='Coding') ) )

  for ro0 in [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9]:
    d_opt = optimal_d_pareto(ro0, N, Cap, k, r, b, beta, a, alpha_gen, red='Coding')
    print("ro0= {}, d_opt= {}".format(ro0, d_opt) )

    # ET = ET_EW_Prqing_pareto_wMGc(ro0, N, Cap, k, r, b, beta, a, alpha_gen, d=0, red='Coding')[0]
    # print("ro0= {}, ET= {}".format(ro0, ET) )

def exp():
  print(">> OPT:")
  sching_m = {'relaunch_time': opt_relaunch_time}
  sim_m = sim(sinfo_m, mapping_m, sching_m)
  log(INFO, "", sching_m=sching_m, sim_m=sim_m)

  print(">> SUB-OPT:")
  sching_m = {'relaunch_time': subopt_relaunch_time}
  sim_m = sim(sinfo_m, mapping_m, sching_m)
  log(INFO, "", sching_m=sching_m, sim_m=sim_m)

  print(">> NONE:")
  sching_m = {'relaunch_time': lambda j: None}
  sim_m = sim(sinfo_m, mapping_m, sching_m)
  log(INFO, "", sching_m=sching_m, sim_m=sim_m)

if __name__ == '__main__':
  N, Cap = 20, 10
  k = BZipf(1, 10)
  R = Uniform(1, 1)
  b, beta = 10, 3
  L = Pareto(b, beta)
  a, alpha = 1, 2.05 # 3
  Sl = Pareto(a, alpha)
  ro = 0.3
  log(INFO, "ro= {}".format(ro) )
  mapping_m = {'type': 'spreading'}
  sinfo_m = {
    'njob': 2000*N, # 10*N
    'ar': ar_for_ro(ro, N, Cap, k, R, L, Sl),
    'nworker': N, 'wcap': Cap,
    'k_rv': k, 'reqed_rv': R, 'lifetime_rv': L,
    'straggle_m': {'slowdown': lambda load: Sl.sample() } }
  log(INFO, "", sinfo_m=sinfo_m, mapping_m=mapping_m)

  exp()