106 lines
3.2 KiB
Python
106 lines
3.2 KiB
Python
import datetime
|
|
import random
|
|
|
|
import connexion
|
|
import peewee
|
|
import six
|
|
import asyncio
|
|
|
|
from swagger_server.models.job_response import JobResponse # noqa: E501
|
|
from swagger_server import util
|
|
from swagger_server.requets_db.models.vrsta import (Job, JobManager)
|
|
from threading import Semaphore, Thread
|
|
from swagger_server.classla import cl_utils
|
|
|
|
CLASSLA_CONCURANCE_LIMIT = 4
|
|
classla_sem = asyncio.Semaphore(CLASSLA_CONCURANCE_LIMIT)
|
|
|
|
|
|
def get_job_status(job_id, show_estimated_completion=None): # noqa: E501
|
|
"""Vrne status
|
|
|
|
# noqa: E501
|
|
|
|
:param job_id:
|
|
:type job_id: int
|
|
:param show_estimated_completion: Calculate estimate time remaining based on various factors (could be inaccurate)
|
|
:type show_estimated_completion: bool
|
|
|
|
:rtype: JobResponse
|
|
"""
|
|
try:
|
|
job = Job.get_by_id(job_id)
|
|
if not job.finished_on:
|
|
est_com = None
|
|
# todo: if estimate completion: calculate it and set it to est_com
|
|
return JobResponse(finished_job=False, estimated_completion=est_com), 200
|
|
return JobResponse(finished_job=True, completed_at=job.finished_on, job_result=job.job_output), 200
|
|
except peewee.DoesNotExist as e:
|
|
return "Job with this ID does not exist", 404
|
|
|
|
|
|
def clear_up_unfinished_jobs():
|
|
"""
|
|
In case server crashed while jobs were in queue...
|
|
"""
|
|
Job.update(started_on=None).where(Job.started_on != None, Job.finished_on == None).execute()
|
|
|
|
|
|
### Job looping
|
|
async def try_do_jobs():
|
|
await asyncio.sleep(15) # wait for tokenizers to load for classla ...
|
|
while True:
|
|
try:
|
|
if cl_utils.nlp_loaded:
|
|
# print(cl_utils.raw_text_to_conllu("Danes je lep soncen dan. Res je!"))
|
|
if classla_sem._value > 0:
|
|
# classla
|
|
unfinished_jobs = Job.select() \
|
|
.where(Job.finished_on == None, Job.started_on == None, Job.job_type == 1) \
|
|
.limit(classla_sem._value)
|
|
tasks = [
|
|
asyncio.ensure_future(execute_classla_job(job))
|
|
for job in unfinished_jobs
|
|
]
|
|
asyncio.get_event_loop().create_task(prep_classla_jobs(tasks))
|
|
# await asyncio.gather(*tasks)
|
|
|
|
|
|
|
|
else:
|
|
pass
|
|
except Exception as e:
|
|
print(f"Exception in ... {e}")
|
|
finally:
|
|
await asyncio.sleep(3)
|
|
|
|
|
|
async def prep_classla_jobs(tasks):
|
|
await asyncio.gather(*tasks)
|
|
|
|
|
|
async def execute_classla_job(job: Job):
|
|
async with classla_sem:
|
|
job.started_on = datetime.datetime.utcnow()
|
|
job.save()
|
|
conllu, _ = cl_utils.raw_text_to_conllu(job.job_input)
|
|
job.job_output = conllu
|
|
job.finished_on = datetime.datetime.utcnow()
|
|
job.save()
|
|
|
|
|
|
clear_up_unfinished_jobs()
|
|
loop = asyncio.get_event_loop()
|
|
def loop_in_thread(loop):
|
|
asyncio.set_event_loop(loop)
|
|
loop.run_until_complete(try_do_jobs())
|
|
|
|
|
|
t = Thread(target=loop_in_thread, args=(loop,))
|
|
t.start()
|
|
|
|
|
|
# clear_up_unfinished_jobs()
|
|
# loop = asyncio.get_event_loop()
|
|
# loop.run_until_complete(try_do_jobs()) this version seems more at home, but it blocks the thread, fix that?
|