Implement host and solver function/GUIs (a1c46b41) · Commits · adc2019 / adc2019-system

adc2019system.py

+1 −0

Changes for adc2019system.py: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -33,6 +33,7 @@ def set_role(role_name, config_data):

		def call_api(method, cmd, params):

		print(f'I: API Received: {cmd}')
		if role is not None:
		return role.call_api(method, cmd, params)
		else:

main.py

+13 −2

Changes for main.py: 13 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -24,6 +24,14 @@ def webui_template_questions():
		else:
		return abort(404)

		@webui.route('/template/system-summary')
		def webui_template_workers():
		if (adc2019system.role is not None) and (adc2019system.role.type == 'host'):
		workers = adc2019system.role.get_workers()
		return render_template('part_system_summary.html', workers=workers)
		else:
		return abort(404)

		@webui.route('/template/question/<name>')
		def webui_template_question_status(name=None):

		@@ -31,8 +39,11 @@ def webui_template_question_status(name=None):
		return abort(404)

		if (adc2019system.role is not None) and (adc2019system.role.type == 'host'):
		questions = adc2019system.role.get_questions()
		return render_template('part_question_status.html', questions=questions)
		question = adc2019system.role.get_question(name)
		if question is None:
		return abort(404)
		else:
		return render_template('part_question_status.html', question=question)
		else:
		return abort(404)

roles/host.py

+175 −10

Changes for roles/host.py: 175 added lines, 10 removed lines.
Original line number	Diff line number	Diff line
		import glob
		import json
		import sys
		import threading
		import time
		import requests

		from utils import Question
		@@ -10,11 +13,13 @@ class Host(object):
		self.type = 'host'

		self.config = config
		self.questions = []
		self._load_questions(config['question_path'])
		self.questions = dict()
		self.worker_manager = None
		self.request = dict()
		self.processing_request = None

		self.worker = dict()
		self.setup_workers()
		self._load_questions(config['question_path'])
		self._setup_workers()

		def __repr__(self):
		return "Host"
		@@ -28,39 +33,199 @@ class Host(object):
		for v in questions_path:

		question = Question(v)
		_key = question.name

		self.questions.append(question)
		self.questions[_key] = question

		def setup_workers(self):
		def _setup_workers(self):

		self.worker_manager = WorkerManager(self.config['address'])
		for v in self.config['worker']:
		worker_name = v['name']
		self.worker[worker_name] = Worker(v)
		self.worker_manager.add_worker(v)

		def get_workers(self):
		return self.worker_manager.get_workers()

		def distribute_question(self, question_key):

		question = self.get_question(question_key)
		if question is None:
		return {'status': 'key error'}
		else:
		request = Request(self.worker_manager, question.get_dict())
		request_id = request.get_id()
		self.request[request_id] = request

		request.broadcast()
		return {
		'status': 'processed',
		'request_id': request_id,
		'timeout': request.timeout
		}

		def get_questions(self):
		return self.questions

		def get_question(self, _key):
		if _key in self.questions:
		return self.questions[_key]
		else:
		return None

		def store_solution(self, solution):
		# request_idをチェックする機能もつくておく
		request_id = solution['request_id']
		if request_id in self.request:
		self.request[request_id].store_response(solution)
		else:
		print(f'W: Unknown request_id: {request_id}')

		question_key = solution['question']

		if question_key in self.questions:
		self.questions[question_key].put_solution(solution)
		return {'status': 'registered'}
		else:
		return {'status': 'error'}

		def get_request_status(self, request_id):
		if request_id in self.request:
		return self.request[request_id].get_status()
		else:
		return {'status': 'unknown request'}

		def call_api(self, method, cmd, params):
		if cmd == 'role':
		# サーバの役割確認
		return {'role': self.type}
		elif cmd == 'question/solve':
		# params['question']に指定された問題をworkerに配信
		question_key = params['question']
		return self.distribute_question(question_key)
		elif cmd == 'question/solution':
		self.store_solution(params)
		return {'status': 'received'}
		elif cmd == 'request/status':
		request_id = float(params['request_id'])
		return self.get_request_status(request_id)
		else:
		return None

		class WorkerManager(object):

		def __init__(self, host_address):
		self.workers = dict()
		self.host = host_address

		def add_worker(self, conf):

		worker_conf = dict()
		worker_conf.update(conf)
		worker_conf['host'] = self.host

		worker_address = worker_conf['address']
		self.workers[worker_address] = Worker(worker_conf)

		def get_workers(self):
		return self.workers

		def broadcast(self, cmd, params):

		threads = []

		def _sender(_worker, _cmd, _params):
		_worker.post(_cmd, _params)

		for k, v in self.workers.items():
		_th = threading.Thread(name=v.address, target=_sender, args=(v, cmd, params), daemon=True)
		_th.start()
		threads.append(_th)

		class Worker(object):

		def __init__(self, params):
		self.address = params['address']
		self.name = params['name']
		self.host = params['host']
		self.role = None
		self.params = params
		self.status = 'Setting up'

		self.set_role(params['role'])

		def post(self, path, data):
		try:
		response = requests.post(
		f'http://{self.address}/api/{path}',
		json.dumps(data),
		headers={'Content-Type': 'application/json'})
		headers={'Content-Type': 'application/json'},
		timeout=2)
		print(f"I: Post to {self.address}, API Cmd: {path}")
		except Exception as e:
		# sys.stderr.write(str(e) + "\n")
		print(f"W: Failed to connect {self.address}")
		self.status = 'Not connected'
		response = None
		return response

		def set_role(self, role):
		r = self.post('role', {'role': role})
		r = self.post('role', self.params)

		class Request(object):

		def __init__(self, worker_manager, data, timeout=10):
		self.worker_manager = worker_manager

		self.data = data
		self.timeout = timeout

		self.request_id = time.time()

		self.broadcast_time = None

		self.response = dict()

		@property
		def request_data(self):
		data = self.data
		data['request_id'] = self.request_id
		data['timeout'] = self.timeout
		return data

		def get_id(self):
		return self.request_id

		def get_dict(self):
		return self.request_data

		def store_response(self, data):
		# TODO: 1つのrequest_idに対し同一のworkerから2つ以上答えが返ってきた場合の例外処理
		worker = data['worker']
		self.response[worker] = data

		def get_status(self):
		all_workers = self.worker_manager.get_workers().keys()
		worker_count = 0
		response_count = 0
		for v in all_workers:
		worker_count += 1
		if v in self.response:
		response_count += 1

		status = ''
		if worker_count == response_count:
		status = 'done'
		elif time.time() - self.broadcast_time > self.timeout:
		status = 'timeout'
		else:
		status = 'processing'

		return {
		'status': status,
		'workers': worker_count,
		'solutions': response_count
		}

		def broadcast(self):
		self.worker_manager.broadcast('solve', self.request_data)
		self.broadcast_time = time.time()

roles/solver.py

+84 −0

Changes for roles/solver.py: 84 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		import importlib
		import json
		import os
		import requests
		import sys
		import time
		import threading

		class StoppableThread(threading.Thread):

		def __init__(self, target, args=()):
		super(StoppableThread, self).__init__(target=target, args=args)
		self._status = 'running'
		def stop(self):
		if self._status == 'running':
		self._status = 'stopping'
		def stopped(self):
		self._status = 'stopped'
		def is_running(self):
		return (self._status == 'running')
		def is_stopping(self):
		return (self._status == 'stopping')
		def is_stopped(self):
		return (self._status == 'stopped')

		class Solver(object):

		def __init__(self, config):
		self.type = 'solver'

		self.host = config['host']
		self.address = config['address']
		self.name = config['name']
		self.solver = importlib.import_module(f"solvers.{config['solver']}")
		self.thread = None

		def __repr__(self):
		return "Solver"

		def solve(self, params):

		start_time = time.time()
		solution = self.solver.solve(params)
		end_time = time.time()
		self.thread.stopped()

		elapsed_time = end_time - start_time

		if not 'elapsed_time' in solution:
		solution['elapsed_time'] = elapsed_time

		self.submit_solution(params, solution)
		self.thread = None
		return True

		def post(self, path, data):
		response = requests.post(
		f'http://{self.host}/api/{path}',
		json.dumps(data),
		headers={'Content-Type': 'application/json'})
		print(f"I: Post to {self.host}, API Cmd: {path}")
		return response

		def submit_solution(self, params, solution):
		data = {
		'request_id': params['request_id'],
		'question': params['name'],
		'worker': self.address
		}
		data.update(solution)
		self.post('question/solution', data)

		def start_solver(self, params):

		if self.thread is None:
		print("I: Solver started")
		self.thread = StoppableThread(target=self.solve, args=(params, ))
		self.thread.start()
		return {'status': 'started'}
		else:
		return {'status': 'busy'}

		def stop_solver():

		if self.thread is not None:
		if self.thread.is_running():
		self.thread.stop()
		self.thread.join()
		self.thread = None

		def call_api(self, method, cmd, params):
		if cmd == 'role':
		return {'role': self.type}
		elif cmd == 'solve':
		return self.start_solver(params)
		else:
		return None

solvers/SampleSolver/main.py

0 → 100644

+6 −0

Changes for solvers/SampleSolver/main.py: 6 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		def solve(params):
		print("This is a sample solver.")
		return {'solution': 'aaa!'}

		def main(params):
		solve(a)