Skip to content

O
OpenXG-Erm
Commit a156dc47 authored by Suzhi Bi's avatar Suzhi Bi
Browse files
Options

Update code

parent 261ea31b
Hide whitespace changes
Inline Side-by-side
Showing with 258 additions and 0 deletions
code
View file @ a156dc47
import subprocess
import socket
import sys
import os
import struct
import numpy as np
import time
def allocation(workload, cpu_per_core, core_num, arrival, requirement, queueing_time):
gain = 1.6
print('A new task arrives.')
flag1 = 0
start.append(arrival + queueing_time)
deadline.append(arrival + requirement)
cpu_remain_0 = np.ones(core_num) * cpu_per_core
if len(finish) != 0:
for i in range(core_num):
cpu_remain_0[i] = cpu_per_core - sum(((np.array(finish) * (np.array(cpu_index) == i)) > start[-1]) * cpu)
index = np.argsort(cpu_remain_0)
if len(finish) != 0:
index = index[::-1]
flag2 = -1
for i in index:
flag2 = flag2 + 1
start[-1] = arrival + queueing_time
if flag2 > 0:
cpu.pop()
if len(finish) != 0:
cpu_remain = cpu_per_core - sum((( np.array(finish) * (np.array(cpu_index) == i) ) > start[-1]) * cpu)
cpu.append(max(gain * workload/requirement, 150 * 10 ** 6))
if cpu[-1] <= cpu_remain:
finish.append(start[-1] + workload/cpu[-1])
cpu_index.append(i)
print("Current task assignment %0.2f MHz frequency of cpu_%d." %(cpu[-1] * 10 ** -6, cpu_index[-1]))
print('--------------------------------------------')
break
else:
temp = np.array(finish) * (np.array(cpu_index) == i) - start[-1]
n = sum(temp > 0)
if n == 0:
if i == (core_num - 1):
cpu[-1] = 0
start[-1] = float('inf')
finish.append(0)
cpu_index.append(-1)
print("drop")
print('--------------------------------------------')
else:
temp[temp < 0] = 10 ** 8 # inf
min_index = np.argsort(temp)
for j in range(n):
cpu_remain = cpu[min_index[j]] + cpu_remain
start[-1] = finish[min_index[j]]
if deadline[-1] <= start[-1] :
if i == (core_num - 1):
cpu[-1] = 0
start[-1] = float('inf')
finish.append(0)
cpu_index.append(-1)
print("drop")
print('--------------------------------------------')
break
cpu[-1] = max(gain * workload/(deadline[-1] - start[-1]), 150 * 10 ** 6)
if cpu[-1] <= cpu_remain:
finish.append(start[-1] + workload/cpu[-1])
cpu_index.append(i)
print("Wait a while.")
print("Current task assignment %0.2f MHz frequency of cpu_%d." %(cpu[-1] * 10 ** -6, cpu_index[-1]))
print('--------------------------------------------')
flag1 = 1
break
if j == (n - 1) and i == (core_num - 1):
cpu[-1] = 0
start[-1] = float('inf')
finish.append(0)
cpu_index.append(-1)
print("drop")
print('--------------------------------------------')
if flag1 == 1:
break
else:
cpu.append(max(gain * workload/requirement, 150 * 10 ** 6))
if cpu[-1] <= cpu_per_core:
finish.append(start[-1] + workload/cpu[-1])
cpu_index.append(i)
print("Current task assignment %0.2f MHz frequency of cpu_%d." %(cpu[-1] * 10 ** -6, cpu_index[-1]))
print('--------------------------------------------')
break
else:
if i == (core_num - 1):
cpu[-1] = 0
start[-1] = float('inf')
finish.append(0)
cpu_index.append(-1)
print("drop")
print('--------------------------------------------')
def deal_data(conn, addr):
print ('Accept new connection from {0}'.format(addr))
fileinfo_size = struct.calcsize('128sq')
buf = conn.recv(fileinfo_size)
global flag
flag = 0
if buf:
if buf.decode('utf-8', 'ignore').isdigit() and len(buf.decode('utf-8', 'ignore')) < 5:
global time_max
time_max = int(buf.decode('utf-8'))
print('The maximum time (seconds) allowed for picture(s) is %d.' %time_max)
flag = 1
else:
global filesize
filename, filesize = struct.unpack('128sq', buf)
fn = filename.strip(str.encode('\00'))
global new_filename
new_filename = os.path.join(str.encode('./'), fn)
print ('file new name is {0}, filesize if {1}'.format(new_filename, filesize))
recvd_size = 0
fp = open(new_filename, 'wb')
print ("start receiving...")
while not recvd_size == filesize:
if filesize - recvd_size > 1024:
data = conn.recv(1024)
recvd_size += len(data)
else:
data = conn.recv(filesize - recvd_size)
recvd_size = filesize
fp.write(data)
fp.close()
print("end receive...")
else:
flag = 1
print("not received...")
conn.close()
ip = '192.168.191.4'
print ("Waiting...")
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind((ip, 8080))
s.listen(10)
except socket.error as msg:
print(msg)
sys.exit(1)
global cpu
cpu = []
global cpu_index
cpu_index = []
global start
start = []
global finish
finish = []
global deadline
deadline = []
data_size = []
workload = []
quota = []
name = []
result = []
image_location = []
python_code = []
core_num = 4
cpu_per_core = 15 * 10 ** 8
period = 10000
new_image = []
time_arrival = []
time_require = []
k = -1
t_start = time.time()
while 1:
conn, addr = s.accept()
deal_data(conn, addr)
if flag == 1:
continue
k = k + 1
t1 = time.time()
new_image.append(str(new_filename, encoding="utf-8"))
data_size.append(os.path.getsize(new_image[k]))
workload.append(data_size[k] * 10 ** 6)
time_arrival.append(t1 - t_start)
time_require.append(time_max)
python_code.append("/classifier_api/image_classification/classifier.py")
image_location.append("/classifier_api/" + str(k) + ".jpg")
name.append("class_" + str(k))
result.append("result_" + str(addr[0]) + "_" + str(os.path.split(new_image[k])[-1]).split('.')[0])
os.system("echo $(docker ps -aq) >> 1.txt")
os.system("echo $(wc 1.txt) >> 2.txt")
f = open("2.txt", "r")
for line in f.readlines():
line = line.strip()
num = int(line.split()[1])
os.system("rm -f 1.txt 2.txt")
queueing_time = 0
while num > 30:
time.sleep(1)
os.system("echo $(docker ps -aq) >> 1.txt")
os.system("echo $(wc 1.txt) >> 2.txt")
f = open("2.txt", "r")
for line in f.readlines():
line = line.strip()
num = int(line.split()[1])
os.system("rm -f 1.txt 2.txt")
queueing_time = queueing_time + 1
allocation(workload[k], cpu_per_core, core_num, time_arrival[k], time_require[k], queueing_time)
quota.append(int(cpu[k]/cpu_per_core * period))
if finish[k] == 0:
message = "The delay of " + str(os.path.split(new_image[k])[-1]) + " is not satisfied."
print(message)
cmd1 = "echo " + str(message) + " >> " + str(result[k]) + ".txt"
os.system(cmd1)
cmd2 = "python3 result_return.py " + str(addr[0]) + " " + str(os.path.split(new_image[k])[-1]).split('.')[0] + ".txt"
os.system(cmd2)
cmd3 = "rm -f result_" + str(addr[0]) + "_" + str(os.path.split(new_image[k])[-1]).split('.')[0] + ".txt"
os.system(cmd3)
os.system("echo Timeout. >> time.txt")
continue
if data_size[k] != filesize:
message = "Error receiving " + str(os.path.split(new_image[k])[-1])
print(message)
cmd1 = "echo " + str(message) + " >> " + str(result[k]) + ".txt"
os.system(cmd1)
cmd2 = "python3 result_return.py " + str(addr[0]) + " " + str(os.path.split(new_image[k])[-1]).split('.')[0] + ".txt"
os.system(cmd2)
cmd3 = "rm -f result_" + str(addr[0]) + "_" + str(os.path.split(new_image[k])[-1]).split('.')[0] + ".txt"
os.system(cmd3)
os.system("echo Error receiving. >> time.txt")
continue
waiting_time = start[k] - time_arrival[k]
time.sleep(waiting_time)
waiting_time = int(waiting_time)
print("start container...")
cmd1 = "docker run --name " + str(name[k]) + " --cpuset-cpus=" + str(cpu_index[k]) + " --cpu-period=" + str(period) + " --cpu-quota=" + str(quota[k]) + " -dt class_api/v1 /bin/bash"
cmd2 = "docker cp " + str(new_image[k]) + " " + str(name[k]) + ":/" + str(image_location[k])
cmd3 = "docker exec " + str(name[k]) + " python3 -W ignore::RuntimeWarning " + str(python_code[k]) + " --image_file " + str(image_location[k]) + " --num_top_predictions 1 >> " + str(result[k]) + ".txt"
cmd4 = "docker stop " + str(name[k])
cmd4_1 = "echo $(($(date --date=$(docker inspect --format='{{.State.FinishedAt}}' " + str(name[k]) + " ) +%s) - $(date --date=$(docker inspect --format='{{.State.StartedAt}}' " + str(name[k]) + " ) +%s)))\(" + str(waiting_time) + "\) seconds/" + str(time_max) + " seconds >> time.txt"
cmd5 = "docker rm " + str(name[k])
cmd6 = "python3 result_return.py " + str(addr[0]) + " " + str(os.path.split(new_image[k])[-1]).split('.')[0] + ".txt"
cmd7 = "rm -f result_" + str(addr[0]) + "_" + str(os.path.split(new_image[k])[-1]).split('.')[0] + ".txt"
cmd = cmd1 + " && " + cmd2 + " && " + cmd3 + " && " + cmd4 + " && " + cmd4_1 + " && " + cmd5 + " && " + cmd6 + " && " + cmd7 # + " && " + cmd8
p = subprocess.Popen(cmd, shell = True, stderr=subprocess.STDOUT, stdin=subprocess.PIPE)
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment