Cuda 12.4.1, pipe name randomization.

README.md

@@ -1,22 +1,3 @@
 # stt:git
 
 [![pypi_status](https://img.shields.io/badge/BentoML-1.1.10-informational)](https://pypi.org/project/BentoML)
-[![documentation_status](https://readthedocs.org/projects/bentoml/badge/?version=latest)](https://docs.bentoml.com/)
-[![join_slack](https://badgen.net/badge/Join/BentoML%20Slack/cyan?icon=slack)](https://l.bentoml.com/join-slack-swagger)
-[![BentoML GitHub Repo](https://img.shields.io/github/stars/bentoml/bentoml?style=social)](https://github.com/bentoml/BentoML)
-[![Twitter Follow](https://img.shields.io/twitter/follow/bentomlai?label=Follow%20BentoML&style=social)](https://twitter.com/bentomlai)
-
-This is a Machine Learning Service created with BentoML.
-| InferenceAPI | Input | Output |
-| ------------ | ----- | ------ |
-| POST [`/process_audio`](#operations-Service_APIs-stt__process_audio) | BytesIOFile | JSON |
-
-
-
-
-## Help
-
-* [📖 Documentation](https://docs.bentoml.com/en/latest/): Learn how to use BentoML.
-* [💬 Community](https://l.bentoml.com/join-slack-swagger): Join the BentoML Slack community.
-* [🐛 GitHub Issues](https://github.com/bentoml/BentoML/issues): Report bugs and feature requests.
-* Tip: you can also [customize this README](https://docs.bentoml.com/en/latest/concepts/bento.html#description).

apis/openapi.yaml

@@ -46,31 +46,11 @@ info:
   contact:
     email: [email protected]
     name: BentoML Team
-  description: "# stt:None\n\n[![pypi_status](https://img.shields.io/badge/BentoML-1.1.10-informational)](https://pypi.org/project/BentoML)\n\
-    [![documentation_status](https://readthedocs.org/projects/bentoml/badge/?version=latest)](https://docs.bentoml.com/)\n\
-    [![join_slack](https://badgen.net/badge/Join/BentoML%20Slack/cyan?icon=slack)](https://l.bentoml.com/join-slack-swagger)\n\
-    [![BentoML GitHub Repo](https://img.shields.io/github/stars/bentoml/bentoml?style=social)](https://github.com/bentoml/BentoML)\n\
-    [![Twitter Follow](https://img.shields.io/twitter/follow/bentomlai?label=Follow%20BentoML&style=social)](https://twitter.com/bentomlai)\n\
-    \nThis is a Machine Learning Service created with BentoML.\n| InferenceAPI | Input\
-    \ | Output |\n| ------------ | ----- | ------ |\n| POST [`/process_audio`](#operations-Service_APIs-stt__process_audio)\
-    \ | BytesIOFile | JSON |\n\n\n\n\n## Help\n\n* [\U0001F4D6 Documentation](https://docs.bentoml.com/en/latest/):\
-    \ Learn how to use BentoML.\n* [\U0001F4AC Community](https://l.bentoml.com/join-slack-swagger):\
-    \ Join the BentoML Slack community.\n* [\U0001F41B GitHub Issues](https://github.com/bentoml/BentoML/issues):\
-    \ Report bugs and feature requests.\n* Tip: you can also [customize this README](https://docs.bentoml.com/en/latest/concepts/bento.html#description).\n"
+  description: "# stt:None\n\n[![pypi_status](https://img.shields.io/badge/BentoML-1.1.10-informational)](https://pypi.org/project/BentoML)\n
   title: stt
   version: None
 openapi: 3.0.2
 paths:
-  /healthz:
-    get:
-      description: Health check endpoint. Expecting an empty response with status
-        code <code>200</code> when the service is in health state. The <code>/healthz</code>
-        endpoint is <b>deprecated</b>. (since Kubernetes v1.16)
-      responses:
-        '200':
-          description: Successful Response
-      tags:
-      - Infrastructure
   /livez:
     get:
       description: Health check endpoint for Kubernetes. Healthy endpoint responses

bento.yaml

@@ -18,7 +18,7 @@ apis:
 docker:
   distro: debian
   python_version: '3.11'
-  cuda_version: 11.6.2
+  cuda_version: 12.4.1
   env:
     BENTOML_CONFIG: src/configuration.yaml
   system_packages: null

bentofile.yaml

@@ -11,6 +11,6 @@ include:
 python:
   requirements_txt: "requirements.txt"
 docker: 
-  cuda_version: "11.6.2"
+  cuda_version: "12.4.1"
   env:
       BENTOML_CONFIG: "src/configuration.yaml"

env/docker/Dockerfile

@@ -5,7 +5,7 @@
 # ===========================================
 
 # Block SETUP_BENTO_BASE_IMAGE
-FROM nvidia/cuda:11.2.2-cudnn8-runtime-ubuntu20.04 as base-container
+FROM nvidia/cuda:12.4.1-devel-ubuntu22.04 as base-container as base-container
 
 ENV LANG=C.UTF-8
 
@@ -27,22 +27,6 @@ RUN set -eux && \
     apt-get install -q -y --no-install-recommends --allow-remove-essential \
         ca-certificates gnupg2 bash build-essential 
 
-RUN \
-    set -eux && \
-    apt-get install -y --no-install-recommends --allow-remove-essential software-properties-common && \
-    # add deadsnakes ppa to install python
-    add-apt-repository ppa:deadsnakes/ppa && \
-    apt-get update -y && \
-    apt-get install -y --no-install-recommends --allow-remove-essential curl python3.11 python3.11-dev python3.11-distutils
-
-RUN ln -sf /usr/bin/python3.11 /usr/bin/python3 && \
-    ln -sf /usr/bin/pip3.11 /usr/bin/pip3
-
-RUN curl -O https://bootstrap.pypa.io/get-pip.py && \
-    python3 get-pip.py && \
-    rm -rf get-pip.py
-
-
 # Block SETUP_BENTO_USER
 ARG BENTO_USER=bentoml
 ARG BENTO_USER_UID=1034
@@ -53,12 +37,11 @@ ENV BENTOML_CONFIG=src/configuration.yaml
 ARG BENTO_PATH=/home/bentoml/bento
 ENV BENTO_PATH=$BENTO_PATH
 ENV BENTOML_HOME=/home/bentoml/
+ENV BENTOML_DO_NOT_TRACK=True
 
 RUN mkdir $BENTO_PATH && chown bentoml:bentoml $BENTO_PATH -R
 WORKDIR $BENTO_PATH
 
-
-
 # Block SETUP_BENTO_COMPONENTS
 COPY --chown=bentoml:bentoml ./env/python ./env/python/
 # install python packages with install.sh

src/requirements.txt

@@ -1,5 +1,5 @@
-ctranslate2==3.24.0
-faster_whisper==0.10.1
+ctranslate2
+faster_whisper
 fastapi
 aiofiles
 asyncio

src/stt_api.py

@@ -3,7 +3,7 @@
 from fastapi.encoders import jsonable_encoder
 import utils.google_streaming.google_streaming_api_pb2 as speech
 import bentoml
-from utils.npipe import AsyncNamedPipe
+from utils.npipe import AsyncChannelWriter, AsyncChannelReader
 import io
 from  runners.audio_transcriber import AudioTranscriber
 import json
@@ -30,14 +30,11 @@ def to_bytes(self):
 
 app = FastAPI()
 
-clients = {}
-
 @app.post("/up")
 async def handleUpstream(pair: str, request: Request):
     try:
         mic_data = bytes()
-
-        async with await AsyncNamedPipe.create(pair) as pipe:
+        async with await AsyncChannelWriter.open(pair) as pipe:
                 async for chunk in request.stream():
                     if len(chunk) == 0:
                         break
@@ -51,10 +48,10 @@ async def handleUpstream(pair: str, request: Request):
     return JSONResponse(content = jsonable_encoder({ "status" : "ok" }))
 
 @app.get("/down")
-async def handleDownstream(pair: str, output: str = "pb"):
+async def handleDownstream(pair: str, request: Request, output: str = "pb"):
     async def handleStream(pair):
         try:
-            async with await AsyncNamedPipe.open(pair) as pipe:
+            async with await AsyncChannelReader.open(pair) as pipe:
                 while True:
                     text = await pipe.readline()
                     if not text:

src/utils/npipe/__init__.py

@@ -1,6 +1 @@
-from os import name as _os_name
-
-if _os_name == "nt":
-    from ._win32 import AsyncNamedPipe
-else:
-    from ._posix import AsyncNamedPipe
+from ._posix import AsyncChannelWriter, AsyncChannelReader

src/utils/npipe/_posix.py

@@ -1,71 +1,92 @@
 import os
+import tempfile
 import asyncio
 import aiofiles
 import aiofiles.os
-from contextlib import contextmanager
+from pathlib import Path
 
-class AsyncNamedPipe:
-    def __init__(self, pipe, path,  creator = False):
-        self._pipe = pipe
-        self._path = path
-        self._creator = creator
 
-    async def create(path: str, timeout: int = 10000):
+class AsyncChannelWriter:
+    def __init__(self, fd, dir: Path):
+        self._fd = fd
+        self._dir = dir
+
+    async def open(pair: str, timeout: int = 10000):
         loop = asyncio.get_running_loop()
-        path = f"/tmp/{path}"
-        try:
-            await loop.run_in_executor(None, os.mkfifo, path)
-        except:
-            pass
-
+
         try:
+            dir = await loop.run_in_executor(None, tempfile.mkdtemp, None, pair + "-", Path.home() / "tmp" / "channels")
+            dir = Path(dir)
+            await loop.run_in_executor(None, os.mkfifo, dir / "pipe")
             async with asyncio.timeout(timeout / 1000.0):
-                fd = await aiofiles.open(path, "w")
-                return AsyncNamedPipe(fd, path, True)
+                fd = await aiofiles.open(dir / "pipe", "w")
+                return AsyncChannelWriter(fd, dir)
 
         except asyncio.TimeoutError:
-            await aiofiles.os.unlink(path)
+            await aiofiles.os.unlink(dir / "pipe")
+            await aiofiles.os.rmdir(dir)
             raise Exception("No consumer")
 
-    async def open(path: str, timeout: int = 1000):
-        pipe = None
-        path = f"/tmp/{path}"
-        try:
-            async with asyncio.timeout(timeout / 1000.0):
-                while not await aiofiles.ospath.exists(path):
-                    await asyncio.sleep(0.1)
-                pipe = await aiofiles.open(path, "r")
-        except asyncio.TimeoutError:
-            raise Exception("No producer")
-
-        return AsyncNamedPipe(pipe, path)
+    async def close(self):
+        await self._fd.close()
+        await aiofiles.os.unlink(self._dir / "pipe")
+        await aiofiles.os.rmdir(self._dir)
 
     async def __aenter__(self):
         return self
 
     async def __aexit__(self, *_):
-        return await self.close()
+        await self.close()
 
     def __bool__(self):
-        return self._pipe is not None
+        return self._fd is not None
 
     async def write(self, data):
-        await self._pipe.write(data)
-        await self._pipe.flush()
+        await self._fd.write(data)
+        await self._fd.flush()
 
     async def writeline(self, data):
-        await self._pipe.writelines([data])
-        await self._pipe.flush()
+        await self._fd.writelines([data])
+        await self._fd.flush()
 
-    async def read(self, size = -1):
-        return await self._pipe.read(size)
 
-    async def readline(self):
-        return await self._pipe.readline()
+class AsyncChannelReader:
+    def __init__(self, fd):
+        self._fd = fd
 
+    async def open(pair: str, timeout: int = 10000):
+        fd = None
+        try:
+            async with asyncio.timeout(timeout / 1000.0):
+                channels = Path.home() / "tmp" / "channels"
+                dirs = None
+                while not dirs:
+                    dirs = list(filter(lambda d: d.startswith(f"{pair}-"),  await aiofiles.os.listdir(channels)))
+                    if not dir:
+                        await asyncio.sleep(0.1)
+
+                path = channels / dirs[0] / "pipe"
+                while not await aiofiles.ospath.exists(path):
+                    await asyncio.sleep(0.1)
+
+                fd = await aiofiles.open(path, "r")
+        except asyncio.TimeoutError:
+            raise Exception("No producer")
+
+        return AsyncChannelReader(fd)
 
     async def close(self):
-        if self._creator:
-            await aiofiles.os.unlink(self._path)
+        await self._fd.close()
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_):
+        await self.close()
+
+    async def read(self, size = -1):
+        return await self._fd.read(size)
+
+    async def readline(self):
+        return await self._fd.readline()
 
-        return await self._pipe.close()