0
\$\begingroup\$

Last week I wrote a simple server on FastApi to run ML model to classify a text on Chinese (split text on separate words). When I had clarified work of this model with API, it became time to confirm it correctness.

Here is a code:

/model.py

""" Chinese text classifier
@ref https://ckip-transformers.readthedocs.io/en/stable/main/readme.html#git
"""
import asyncio
from ckip_transformers.nlp import (
    CkipWordSegmenter, 
)


class ChineseTextClassifier:

    def __init__(self):
        self.ws_driver  = CkipWordSegmenter(model="bert-base")
        self.lock = asyncio.Lock()

    async def run_single_word_segmentation(self, text):
        async with self.lock: 
            assert(isinstance(text, list) == True)
            ws = self.ws_driver(text, use_delim=True)
            return ws

/app.py

from typing import Union

from typing import Optional, Any
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from model import ChineseTextClassifier

app = FastAPI()

model = ChineseTextClassifier()

class TextForClassification(BaseModel):
    text: str

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.post("/classify")
async def classify(payload: TextForClassification):
    result = await model.run_single_word_segmentation([payload.text])
    return get_response(True, result)

# ref. https://pypi.org/project/fastapi-queue/
def get_response(success_status: bool, result: Any) -> JSONResponse | dict:
    if success_status:
        return {"status": 200, "data": result}
    if result == -1:
        return JSONResponse(status_code=503, content="Service Temporarily Unavailable")
    else:
        return JSONResponse(status_code=500, content="Internal Server Error")

The question is how to synchronise access to a model.run_single_word_segmentation()?

I could fade-out concurrency issue here by creating a new instance of model on each request. However, the model underneath use 400 mb pytorch model which will be downloaded each time on new model instance which forced me to keep just one instance of this model and share it among requests.

In Java/Kotlin I could use synchronisation keyword to sync access to a critical section or variety of Locks or any other primitives. Surprisingly, I don't have any thought on how to do this in kotlin coroutines - seems I haven't met such task during past 3-4 years of active kotlin coroutines usage.

Further research shown it asyncio Lock is also supported in Python, but they are not thread safe. However, since I moved my API request to fully async / await mechanism all code is run in the single thread and all async work is done by python coroutines.

Although, docs says CPU intensive operations should be run within loop.run_in_executor(). Despite on model classify test sample of text quite fast, I still have to think about better alternatives. run_in_executor() will move all benefits of coroutines to a plain threading.

Other options to consider are threading.Lock vs threading.RLock vs asyncio.Lock

\$\endgroup\$

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.