From f89b84dd310c372e873a6bf98ae17bab0ed9f10f Mon Sep 17 00:00:00 2001 From: Kikimanox Date: Tue, 18 Oct 2022 19:24:38 +0200 Subject: [PATCH] extract method (async) --- .../controllers/extract_controller.py | 35 +++++++++------- swagger_server/controllers/jobs_controller.py | 40 ++++++++++++++++++- swagger_server/requets_db/models/vrsta.py | 4 +- 3 files changed, 60 insertions(+), 19 deletions(-) diff --git a/swagger_server/controllers/extract_controller.py b/swagger_server/controllers/extract_controller.py index fdd0e47..f9efece 100644 --- a/swagger_server/controllers/extract_controller.py +++ b/swagger_server/controllers/extract_controller.py @@ -6,6 +6,7 @@ import json from pathlib import Path from swagger_server.models.izlusci_async_body import IzlusciAsyncBody # noqa: E501 from swagger_server.models.izlusci_sync_body import IzlusciSyncBody # noqa: E501 +from swagger_server.requets_db.models.vrsta import JobManager from swagger_server.utils import cl_utils from swagger_server.util import get_random_filename, create_random_file_in_tmp_folder import requests @@ -17,21 +18,6 @@ ATEapi_endpoint = "http://localhost:5000/predict" # ATEapi_endpoint = "http://ate-api:5000/predict" -def get_candidates_async(body): # noqa: E501 - """Izlusci terminološke kandidate iz seznama besedil v conllu obliki [asinhrono, ustvari novi job] - - # noqa: E501 - - :param body: - :type body: dict | bytes - - :rtype: str - """ - if connexion.request.is_json: - body = IzlusciAsyncBody.from_dict(connexion.request.get_json()) # noqa: E501 - return 'do some magic!' - - def do_izlusci(conllus, prepovedane_besede): tmp_file_path = "" try: @@ -67,6 +53,25 @@ def do_izlusci(conllus, prepovedane_besede): return str(e), 500 +def get_candidates_async(body): # noqa: E501 + """Izlusci terminološke kandidate iz seznama besedil v conllu obliki [asinhrono, ustvari novi job] + + # noqa: E501 + + :param body: + :type body: dict | bytes + + :rtype: str + """ + if connexion.request.is_json: + body = IzlusciAsyncBody.from_dict(connexion.request.get_json()) # noqa: E501 + job, is_old_job = JobManager.create_job(4, json.dumps(body.to_dict())) + if job is None: + return "Something went wrong", 500 + ret = {'check_job_url': f'{connexion.request.url_root}/job/{job.id}'} + return ret, 200 + + def get_candidates_sync(body): # noqa: E501 """Izlusci terminološke kandidate iz seznama besedil v conllu obliki [sihrono, rezultat v sami zahtevi] diff --git a/swagger_server/controllers/jobs_controller.py b/swagger_server/controllers/jobs_controller.py index 6c5518b..2607fa4 100644 --- a/swagger_server/controllers/jobs_controller.py +++ b/swagger_server/controllers/jobs_controller.py @@ -1,9 +1,12 @@ import datetime +import json import os.path import peewee import asyncio import concurrent.futures as cf + +from swagger_server.controllers.extract_controller import do_izlusci from swagger_server.models.job_response import JobResponse # noqa: E501 from swagger_server.requets_db.models.vrsta import (Job) from threading import Thread @@ -14,10 +17,12 @@ import threading import time CLASSLA_CONCURANCE_LIMIT = 3 -DOC2TEXT_CONCURANCE_LIMIT = 4 +DOC2TEXT_CONCURANCE_LIMIT = 3 +ATEAPI_CONCURANCE_LIMIT = 2 classla_sem = threading.Semaphore(CLASSLA_CONCURANCE_LIMIT) doc2text_sem = threading.Semaphore(DOC2TEXT_CONCURANCE_LIMIT) +ateapi_sem = threading.Semaphore(ATEAPI_CONCURANCE_LIMIT) def delete_job(job_id): # noqa: E501 @@ -69,9 +74,26 @@ def clear_up_unfinished_jobs(): async def try_do_jobs(): - with cf.ThreadPoolExecutor(max_workers=2) as ex: + with cf.ThreadPoolExecutor(max_workers=3) as ex: ex.submit(try_do_jobs_classla) ex.submit(try_do_jobs_doc2text) + ex.submit(try_do_jobs_ateapi) + + +### Job looping +def try_do_jobs_ateapi(): + while True: + try: + if ateapi_sem._value > 0: + unfinished_jobs = Job.select() \ + .where(Job.finished_on.is_null(), Job.started_on.is_null(), Job.job_type == 4) \ + .limit(ateapi_sem._value) + with cf.ThreadPoolExecutor(max_workers=ATEAPI_CONCURANCE_LIMIT) as ex: + [ex.submit(execute_ateapi_job, job) for job in unfinished_jobs] + except Exception as e: + print(f"Exception in try_do_jobs_ateapi") + finally: + time.sleep(3) ### Job looping @@ -180,6 +202,20 @@ def execute_classla_job(job: Job): classla_sem.release() +def execute_ateapi_job(job: Job): + try: + ateapi_sem.acquire() + job.started_on = datetime.datetime.utcnow() + job.save() + info = json.loads(job.job_input) + ret_json, _ = do_izlusci(info['conllus'], info['prepovedane_besede']) + job.job_output = ret_json + job.finished_on = datetime.datetime.utcnow() + job.save() + finally: + ateapi_sem.release() + + clear_up_unfinished_jobs() loop = asyncio.get_event_loop() diff --git a/swagger_server/requets_db/models/vrsta.py b/swagger_server/requets_db/models/vrsta.py index 4d75031..dbfd5f9 100644 --- a/swagger_server/requets_db/models/vrsta.py +++ b/swagger_server/requets_db/models/vrsta.py @@ -43,7 +43,7 @@ class Job(BaseModel): input_file = TextField(index=True, null=True) -db.drop_tables([Job]) # TODO: After pushing this, comment it and push again +# db.drop_tables([Job]) # TODO: After pushing this, comment it and push again db.create_tables([Job]) @@ -60,7 +60,7 @@ class JobManager: :return: Job object, Did already exist boolean """ try: - if job_type == 2: + if job_type in [2, 4]: job, is_new = Job.get_or_create(job_type=job_type, job_input=job_input, input_size=len(job_input)) elif job_type in [1, 3, 12, 32]: tmp_file = ""