webapps/term_extractor
18
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Working fix pre merge

Browse Source
This commit is contained in:
marko.ferme
2023-01-10 11:20:14 +01:00
parent 1cd7663d49
commit 69a92a3420
32 changed files with 1457 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files
swagger_server/controllers/jobs_controller.py
+233 -57
View File
@@ -6,28 +6,54 @@ import traceback
import peewee
import asyncio
import concurrent.futures as cf
from flask import Response
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
from swagger_server.utils import cl_utils
from swagger_server.utils import cl_utils, db_utils
from swagger_server.utils import txt_utils
from werkzeug.datastructures import FileStorage
import threading
import time
CLASSLA_CONCURANCE_LIMIT = 3
DOC2TEXT_CONCURANCE_LIMIT = 3
ATEAPI_CONCURANCE_LIMIT = 2
CLASSLA_SMALL_SIZE_LIMIT = 500 * 1e3 # 500 KB, aka.: 500 * 10^3
classla_sem = threading.Semaphore(CLASSLA_CONCURANCE_LIMIT)
doc2text_sem = threading.Semaphore(DOC2TEXT_CONCURANCE_LIMIT)
ateapi_sem = threading.Semaphore(ATEAPI_CONCURANCE_LIMIT)
DOC2TEXT_SMALL_SIZE_LIMIT = 10 * 1e6 # 10 MB
ATEAPI_SMALL_SIZE_LIMIT = 2 * 1e6 # 2 MB
########################################
CLASSLA_CONCURANCE_LIMIT_BIG = 2
CLASSLA_CONCURANCE_LIMIT_SMALL = 2
DOC2TEXT_CONCURANCE_LIMIT_BIG = 2
DOC2TEXT_CONCURANCE_LIMIT_SMALL = 2
ATEAPI_CONCURANCE_LIMIT_BIG = 2
ATEAPI_CONCURANCE_LIMIT_SMALL = 2
IZLUSCI_PO_ISKANJU_CONCURANCE_LIMIT = 4
########################################
classla_sem_big = threading.Semaphore(CLASSLA_CONCURANCE_LIMIT_BIG)
classla_sem_small = threading.Semaphore(CLASSLA_CONCURANCE_LIMIT_SMALL)
doc2text_sem_big = threading.Semaphore(DOC2TEXT_CONCURANCE_LIMIT_BIG)
doc2text_sem_small = threading.Semaphore(DOC2TEXT_CONCURANCE_LIMIT_SMALL)
ateapi_sem_big = threading.Semaphore(ATEAPI_CONCURANCE_LIMIT_BIG)
ateapi_sem_small = threading.Semaphore(ATEAPI_CONCURANCE_LIMIT_SMALL)
izluscipoiskanju_sem = threading.Semaphore(IZLUSCI_PO_ISKANJU_CONCURANCE_LIMIT)
running_threads = {} # <--- dict currently not used, was trying to figure out how to cancel workers mid execution,
# no luck with that yet
def delete_job(job_id): # noqa: E501
"""Izbriše job
"""Izbriše job
# noqa: E501
@@ -36,7 +62,14 @@ def delete_job(job_id): # noqa: E501
:rtype: str
"""
return 'Endpoint currently disabled'
try:
job = Job.get_by_id(job_id)
if job.started_on is not None and job.finished_on is None:
return Response("Cancelling ongoing jobs currently not implemented.", 400)
job.delete_instance()
return f"Job with the ID {job_id} was removed."
except peewee.DoesNotExist:
return Response("Job with this ID does not exist", 404)
def get_job_status(job_id): # noqa: E501
@@ -56,8 +89,19 @@ def get_job_status(job_id): # noqa: E501
if job.started_on is not None and job.finished_on is None:
return JobResponse(job_status="currently processing", created_on=job.created_on,
started_on=job.started_on), 200
if job.started_on is not None and job.finished_on is not None:
return JobResponse(job_status="finished processing", created_on=job.created_on, started_on=job.started_on,
if job.started_on is not None and job.finished_on is not None and not job.job_output.startswith("ERROR -"):
res = job.job_output
if job.job_type in [4, 5]:
try:
res = json.loads(job.job_output)
except:
pass
return JobResponse(job_status="finished processing (OK)", created_on=job.created_on,
started_on=job.started_on,
finished_on=job.finished_on, job_result=res), 200
if job.started_on is not None and job.finished_on is not None and job.job_output.startswith("ERROR -"):
return JobResponse(job_status="finished processing (ERROR)", created_on=job.created_on,
started_on=job.started_on,
finished_on=job.finished_on, job_result=job.job_output), 200
except peewee.DoesNotExist:
return "Job with this ID does not exist", 404
@@ -75,22 +119,65 @@ def clear_up_unfinished_jobs():
async def try_do_jobs():
with cf.ThreadPoolExecutor(max_workers=3) as ex:
with cf.ThreadPoolExecutor(max_workers=4) as ex:
ex.submit(try_do_jobs_classla)
ex.submit(try_do_jobs_doc2text)
ex.submit(try_do_jobs_ateapi)
ex.submit(try_do_jobs_izluscipoiskanju)
### Job looping
# to pe je pod ex.submit
# sub = ex.submit(execute_ateapi_job, job)
# running_threads[job.id] = sub
# time.sleep(1)
# running_threads[job.id]
# preveri ce je done: ```running_threads[job.id].done()``` (vrne true false)
# was_canceled = running_threads[job.id].cancel()
# Todo: Mogoce kaksna druga opcija? Ampak verjetno ne, ne vidim (še?) kak prekicat ONGOING job
# To zgoraj preklice samo job, ki se se ni zacel, kar pa ni za ta use case uporabno.
### Picking jobs for looping
def try_do_jobs_izluscipoiskanju():
while True:
try:
if izluscipoiskanju_sem._value > 0:
unfinished_jobs = Job.select() \
.where(Job.finished_on.is_null(), Job.started_on.is_null(), Job.job_type == 5) \
.limit(izluscipoiskanju_sem._value)
with cf.ThreadPoolExecutor(max_workers=IZLUSCI_PO_ISKANJU_CONCURANCE_LIMIT) as ex:
[ex.submit(execute_izluscipoiskanju_job, job, izluscipoiskanju_sem) for job in unfinished_jobs]
except Exception as e:
print(f"Exception in try_do_jobs_izluscipoiskanju")
traceback.print_exc()
finally:
time.sleep(3)
### Picking jobs for 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]
unfinished_jobs_big = []
unfinished_jobs_small = []
if ateapi_sem_big._value > 0:
unfinished_jobs_big.extend(Job.select().where(Job.finished_on.is_null(), Job.started_on.is_null(),
Job.job_type == 4,
Job.input_size > ATEAPI_SMALL_SIZE_LIMIT) \
.limit(ateapi_sem_big._value))
if ateapi_sem_small._value > 0:
unfinished_jobs_small.extend(Job.select().where(Job.finished_on.is_null(), Job.started_on.is_null(),
Job.job_type == 4,
Job.input_size <= ATEAPI_SMALL_SIZE_LIMIT) \
.limit(ateapi_sem_small._value))
if len(unfinished_jobs_big) > 0:
with cf.ThreadPoolExecutor(max_workers=ATEAPI_CONCURANCE_LIMIT_BIG) as ex:
[ex.submit(execute_ateapi_job, job, ateapi_sem_big) for job in unfinished_jobs_big]
if len(unfinished_jobs_small) > 0:
with cf.ThreadPoolExecutor(max_workers=ATEAPI_CONCURANCE_LIMIT_SMALL) as ex:
[ex.submit(execute_ateapi_job, job, ateapi_sem_small) for job in unfinished_jobs_small]
except Exception as e:
print(f"Exception in try_do_jobs_ateapi")
traceback.print_exc()
@@ -98,27 +185,51 @@ def try_do_jobs_ateapi():
time.sleep(3)
### Job looping
### Picking jobs for looping
def try_do_jobs_classla():
time.sleep(15) # wait for tokenizers to load for classla ...
while True:
try:
if cl_utils.nlp_loaded:
if classla_sem._value > 0:
unfinished_jobs_txt = Job.select() \
.where(Job.finished_on.is_null(), Job.job_type == 2,
Job.input_file.is_null(False)) \
.limit(classla_sem._value)
if not cl_utils.nlp_loaded:
raise Exception("NLP utils not loaded yet.")
unfinished_jobs_no_txt = Job.select() \
.where(Job.finished_on.is_null(), Job.started_on.is_null(), Job.job_type == 2,
Job.input_file.is_null()) \
.limit(classla_sem._value)
unfinished_jobs_big = []
unfinished_jobs_small = []
unfinished_jobs = [j for j in unfinished_jobs_txt] + [j for j in unfinished_jobs_no_txt]
unfinished_jobs = unfinished_jobs[:classla_sem._value]
with cf.ThreadPoolExecutor(max_workers=CLASSLA_CONCURANCE_LIMIT) as ex:
[ex.submit(execute_classla_job, job) for job in unfinished_jobs]
if classla_sem_big._value > 0:
unfinished_jobs_txt = Job.select() \
.where(Job.finished_on.is_null(), Job.job_type == 2,
Job.input_file.is_null(False), Job.input_size > CLASSLA_SMALL_SIZE_LIMIT) \
.limit(classla_sem_big._value)
unfinished_jobs_no_txt = Job.select() \
.where(Job.finished_on.is_null(), Job.started_on.is_null(), Job.job_type == 2,
Job.input_file.is_null(), Job.input_size > CLASSLA_SMALL_SIZE_LIMIT) \
.limit(classla_sem_big._value)
unfinished_jobs_big = [j for j in unfinished_jobs_txt] + [j for j in unfinished_jobs_no_txt]
unfinished_jobs_big = unfinished_jobs_big[:classla_sem_big._value]
if classla_sem_small._value > 0:
unfinished_jobs_txt = Job.select() \
.where(Job.finished_on.is_null(), Job.job_type == 2,
Job.input_file.is_null(False), Job.input_size <= CLASSLA_SMALL_SIZE_LIMIT) \
.limit(classla_sem_small._value)
unfinished_jobs_no_txt = Job.select() \
.where(Job.finished_on.is_null(), Job.started_on.is_null(), Job.job_type == 2,
Job.input_file.is_null(), Job.input_size <= CLASSLA_SMALL_SIZE_LIMIT) \
.limit(classla_sem_small._value)
unfinished_jobs_small = [j for j in unfinished_jobs_txt] + [j for j in unfinished_jobs_no_txt]
unfinished_jobs_small = unfinished_jobs_small[:classla_sem_small._value]
if len(unfinished_jobs_big) > 0:
with cf.ThreadPoolExecutor(max_workers=CLASSLA_CONCURANCE_LIMIT_BIG) as ex:
[ex.submit(execute_classla_job, job, doc2text_sem_big) for job in unfinished_jobs_big]
if len(unfinished_jobs_small) > 0:
with cf.ThreadPoolExecutor(max_workers=CLASSLA_CONCURANCE_LIMIT_SMALL) as ex:
[ex.submit(execute_classla_job, job, doc2text_sem_small) for job in unfinished_jobs_small]
except Exception as e:
print(f"Exception in try_do_jobs_classla")
@@ -127,16 +238,29 @@ def try_do_jobs_classla():
time.sleep(3)
### Job looping
### Picking jobs for looping
def try_do_jobs_doc2text():
while True:
try:
if doc2text_sem._value > 0:
unfinished_jobs = Job.select() \
.where(Job.finished_on.is_null(), Job.started_on.is_null(), Job.job_type << [1, 12, 3, 32]) \
.limit(doc2text_sem._value)
with cf.ThreadPoolExecutor(max_workers=DOC2TEXT_CONCURANCE_LIMIT) as ex:
[ex.submit(execute_doc2text_job, job) for job in unfinished_jobs]
unfinished_jobs_big = []
unfinished_jobs_small = []
if doc2text_sem_big._value > 0:
unfinished_jobs_big.extend(Job.select().where(Job.finished_on.is_null(), Job.started_on.is_null(),
Job.job_type << [1, 12, 3, 32],
Job.input_size > DOC2TEXT_SMALL_SIZE_LIMIT) \
.limit(doc2text_sem_big._value))
if doc2text_sem_small._value > 0:
unfinished_jobs_small.extend(Job.select().where(Job.finished_on.is_null(), Job.started_on.is_null(),
Job.job_type << [1, 12, 3, 32],
Job.input_size <= DOC2TEXT_SMALL_SIZE_LIMIT) \
.limit(doc2text_sem_small._value))
if len(unfinished_jobs_big) > 0:
with cf.ThreadPoolExecutor(max_workers=DOC2TEXT_CONCURANCE_LIMIT_BIG) as ex:
[ex.submit(execute_doc2text_job, job, doc2text_sem_big) for job in unfinished_jobs_big]
if len(unfinished_jobs_small) > 0:
with cf.ThreadPoolExecutor(max_workers=DOC2TEXT_CONCURANCE_LIMIT_SMALL) as ex:
[ex.submit(execute_doc2text_job, job, doc2text_sem_small) for job in unfinished_jobs_small]
except Exception as e:
print(f"Exception in try_do_jobs_doc2text")
traceback.print_exc()
@@ -144,13 +268,14 @@ def try_do_jobs_doc2text():
time.sleep(3)
async def prep_jobs(tasks):
await asyncio.gather(*tasks)
# async def prep_jobs(tasks):
# await asyncio.gather(*tasks)
def execute_doc2text_job(job: Job):
####### JOB EXECUTION LOGIC
def execute_doc2text_job(job: Job, sem: threading.Semaphore):
try:
doc2text_sem.acquire()
sem.acquire()
del_file = False
job.started_on = datetime.datetime.utcnow()
job.save()
@@ -158,7 +283,7 @@ def execute_doc2text_job(job: Job):
tmp_file_path = job.input_file
if not os.path.exists(tmp_file_path):
job.finished_on = datetime.datetime.utcnow()
job.job_output = "ERROR - Temporary file went missing, couldn't properly finish job"
job.job_output = "ERROR - Temporary file went missing, couldn't properly finish job. Please try executing the job again."
job.save()
return
@@ -190,34 +315,85 @@ def execute_doc2text_job(job: Job):
job.started_on = None
job.save()
finally:
doc2text_sem.release()
sem.release()
def execute_classla_job(job: Job):
####### JOB EXECUTION LOGIC
def execute_classla_job(job: Job, sem: threading.Semaphore):
try:
classla_sem.acquire()
sem.acquire()
job.started_on = datetime.datetime.utcnow()
job.save()
conllu, _ = cl_utils.raw_text_to_conllu(job.job_input)
conllu, status = cl_utils.raw_text_to_conllu(job.job_input)
if status != 200:
conllu = f'ERROR - {conllu}'
job.job_output = conllu
job.finished_on = datetime.datetime.utcnow()
job.save()
except:
job.job_output = "ERROR - Something unexpected went wrong. Logs have been saved. Please contact the api admin if the problem persists."
job.finished_on = datetime.datetime.utcnow()
job.save()
print(f"Unexpected error at job {job.id}")
finally:
classla_sem.release()
sem.release()
def execute_ateapi_job(job: Job):
####### JOB EXECUTION LOGIC
def execute_ateapi_job(job: Job, sem: threading.Semaphore):
try:
ateapi_sem.acquire()
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
_res = do_izlusci(info['conllus'], info['prepovedane_besede'],info['definicije'])
if type(_res) is tuple:
ret_json = _res[0]
else:
try:
if type(_res.response) is dict:
ret_json = str(_res.response)
else:
try:
ret_json = _res.response[0].decode('utf-8')
except:
ret_json = "ERROR - Unknown exception."
if _res.status_code != 200:
ret_json = f'ERROR - {ret_json}'
except:
ret_json = "ERROR - Unknown exception."
job.job_output = json.dumps(ret_json, ensure_ascii=False)
job.finished_on = datetime.datetime.utcnow()
job.save()
except:
job.job_output = "ERROR - Something unexpected went wrong. Logs have been saved. Please contact the api admin if the problem persists."
job.finished_on = datetime.datetime.utcnow()
job.save()
print(f"Unexpected error at job {job.id}")
finally:
ateapi_sem.release()
sem.release()
####### JOB EXECUTION LOGIC
def execute_izluscipoiskanju_job(job: Job, sem: threading.Semaphore):
try:
sem.acquire()
job.started_on = datetime.datetime.utcnow()
job.save()
info = json.loads(job.job_input)
terKand = db_utils.vrni_oss_terminoloske_kandidate(info['leta'], info['vrste'], info['kljucne_besede'],
info['prepovedane_besede'], info['udk'],info['definicije'])
job.job_output = json.dumps(terKand, ensure_ascii=False)
job.finished_on = datetime.datetime.utcnow()
job.save()
except:
job.job_output = "ERROR - Something unexpected went wrong. Logs have been saved. Please contact the api admin if the problem persists."
job.finished_on = datetime.datetime.utcnow()
job.save()
print(f"Unexpected error at job {job.id}")
finally:
sem.release()
clear_up_unfinished_jobs()