Display realtime output

[AbbeKamalov]

I have a text-generator language model that is compressed and is in .bin format, and it can accessed by command line terminal. It generates about one word per second and prints out every word in realtime when accessed through command line. I would like to deploy my model using cortex but I'm struggling to get the output in realtime word-by-word. Right now my code prints out only one line a time.

import subprocess
def run_command(text):
    command = ['./mycommand', text]
    process = subprocess.Popen(command, stdout=subprocess.PIPE, universal_newlines=True, bufsize=-1)
    while True:
        output = process.stdout.readline()
        if output == '' and process.poll() is not None:
            break
        if output:
            print(output.strip())

run_command('TEXT')

One line may include about 20 words, so that means 1 line will be displayed every 20 seconds (since my model outputs roughly 1 word/ second). I would really like the output to be more dynamic and output one word at a time (as it does through command line) instead of just one line. Is there a way this can be achieved?

[vishalbollu]

Is it possible for your model to generate a single word at a time rather than a sentence? If it is, rather than having your API respond with 20 words per request, set up your API to respond with a single word based on the provided prompt. In your client, you can iteratively query your API for the next word until you have a text with the desired length.

[AbbeKamalov]

Is it possible for your model to generate a single word at a time rather than a sentence? If it is, rather than having your API respond with 20 words per request, set up your API to respond with a single word based on the provided prompt. In your client, you can iteratively query your API for the next word until you have a text with the desired length.

Correct me if I'm wrong but if I would query my API iteratively that would mean that I would have to use subprocess.Popen multiple times, which would be costly. I would like to keep the subprocess open and just print out the output in realtime. Opening new subrocess takes about 3 to 4 seconds + 1 second to actually generate output makes for about 5 seconds/word which is quite slow

[vishalbollu]

Running subprocess.Popen multiple times is a valid concern. Using sockets could be a way to stream partial responses. Unfortunately, Cortex doesn't support sockets at the moment.

Rather than running subprocess.Popen in the predict function, you can run subprocess.Popen in the constructor (__init__) to initialize the process. Then you can use process.communicate to send text to your process in your predict function.

Out of curiosity, have you considered loading your model in Python rather than using subprocess.Popen to run a binary?

[AbbeKamalov]

Running subprocess.Popen multiple times is a valid concern. Using sockets could be a way to stream partial responses. Unfortunately, Cortex doesn't support sockets at the moment.

Rather than running subprocess.Popen in the predict function, you can run subprocess.Popen in the constructor (__init__) to initialize the process. Then you can use process.communicate to send text to your process in your predict function.

Out of curiosity, have you considered loading your model in Python rather than using subprocess.Popen to run a binary?

I run binary, since it is compressed version of a large model which otherwise require a lot more memory to run. You mentioned that cortex doesn’t yet support sockets. But cortex spins up instances that run Linux, and those instances could run websocket servers in normal circumstances. I’m thinking that I just need to somehow communicate to EC2 AWS instance and launch a websocket which then could display partial output in realtime instead of using subprocess. What do you think?

[RobertLucian]

@MUTAL to avoid having to run subprocess.Popen multiple times, you can just run it once in the constructor of the Python predictor. This way, you don't have to wait for those "3 to 4 seconds" every single time.

Once the subprocess.Popen is run in the constructor, you can then use subprocess.Popen.communicate to interact with your process. You can read from sdout/stderr, you can write to stdin and so on. The constructor only runs once. The predict method runs for each request.

With this in, it means that for a request that asks for 5 words, it will take 5 seconds to process that request - not 5 + 3/4 seconds = 8/9 seconds.

Does this answer your question?

[AbbeKamalov]

It still not as quick as running the command from command line though (1 word/sec). I’m looking into setting up a websocket on each instance using the predictor’s init function and then streaming the output to the client side. Cortex could then spin down the instance when client side sends signal to it. What do you think?

…

On Thu, 29 Oct 2020 at 16:18, Robert Lucian Chiriac < ***@***.***> wrote: @MUTAL <https://github.com/mutal> to avoid having to run subprocess.Popen multiple times, you can just run it once in the constructor of the Python predictor. This way, you don't have to wait for those "3 to 4 seconds" every single time. Once the subprocess.Popen is run in the constructor, you can then use subprocess.Popen.communicate <https://docs.python.org/3/library/subprocess.html#subprocess.Popen.communicate> to interact with your process. You can read from sdout/stderr, you can write to stdin and so on. The constructor only runs once. The predict method runs for each request. With this in, it means that for a request that asks for 5 words, it will take 5 seconds to process that request - not 5 + 3/4 seconds = 8/9 seconds. Does this answer your question? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1500 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ADFOXIUVIMVXJ5KKG3OMOYTSNGBUPANCNFSM4TCIV5QA> .

[deliahu]

@MUTAL I think I have discovered an approach that will work for you.

This example that works for me:

cortex.yaml

- name: test
  kind: RealtimeAPI
  predictor:
    type: python
    image: quay.io/cortexlabsdev/python-predictor-cpu-async:0.23.0
    path: predictor.py

predictor.py

import asyncio
from sse_starlette.sse import EventSourceResponse


async def run_command():
    command = "echo test1 && sleep 2 && echo test2 && sleep 2 && echo test3 && echo done"

    proc = await asyncio.create_subprocess_shell(
        command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )

    while True:
        output = await proc.stdout.readline()
        if not output:
            return
        yield output.decode().strip()


class PythonPredictor:
    def __init__(self, config):
        pass

    async def predict(self, payload):
        return EventSourceResponse(run_command())

requirements.txt

sse-starlette

Note that I built a new version of the predictor image (specified in the image API configuration field), because it required a small change to one of our source files. It should be straightforward for you to build a custom image with this change for yourself, depending on the version of cortex you want do use. The easiest way would be to use the official Cortex image as a base image in your Dockerfile, and then replace the file that needs to be updated using a COPY command. Here are instructions for using a custom docker image.

Also note that with this approach, threads_per_process will not be enforced.

We are not ready to merge #1565, because there are some known consequences of this that we would need to address (e.g. threads_per_process not being enforced).

Let me know if that works for you!

[AbbeKamalov]

@MUTAL I think I have discovered an approach that will work for you.

This example that works for me:

cortex.yaml

- name: test
  kind: RealtimeAPI
  predictor:
    type: python
    image: quay.io/cortexlabsdev/python-predictor-cpu-async:0.23.0
    path: predictor.py

predictor.py

import asyncio
from sse_starlette.sse import EventSourceResponse


async def run_command():
    command = "echo test1 && sleep 2 && echo test2 && sleep 2 && echo test3 && echo done"

    proc = await asyncio.create_subprocess_shell(
        command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )

    while True:
        output = await proc.stdout.readline()
        if not output:
            return
        yield output.decode().strip()


class PythonPredictor:
    def __init__(self, config):
        pass

    async def predict(self, payload):
        return EventSourceResponse(run_command())

requirements.txt

sse-starlette

Note that I built a new version of the predictor image (specified in the image API configuration field), because it required a small change to one of our source files. It should be straightforward for you to build a custom image with this change for yourself, depending on the version of cortex you want do use. The easiest way would be to use the official Cortex image as a base image in your Dockerfile, and then replace the file that needs to be updated using a COPY command. Here are instructions for using a custom docker image.

Also note that with this approach, threads_per_process will not be enforced.

We are not ready to merge #1565, because there are some known consequences of this that we would need to address (e.g. threads_per_process not being enforced).

Let me know if that works for you!

@MUTAL I think I have discovered an approach that will work for you.

This example that works for me:

cortex.yaml

- name: test
  kind: RealtimeAPI
  predictor:
    type: python
    image: quay.io/cortexlabsdev/python-predictor-cpu-async:0.23.0
    path: predictor.py

predictor.py

import asyncio
from sse_starlette.sse import EventSourceResponse


async def run_command():
    command = "echo test1 && sleep 2 && echo test2 && sleep 2 && echo test3 && echo done"

    proc = await asyncio.create_subprocess_shell(
        command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )

    while True:
        output = await proc.stdout.readline()
        if not output:
            return
        yield output.decode().strip()


class PythonPredictor:
    def __init__(self, config):
        pass

    async def predict(self, payload):
        return EventSourceResponse(run_command())

requirements.txt

sse-starlette

Note that I built a new version of the predictor image (specified in the image API configuration field), because it required a small change to one of our source files. It should be straightforward for you to build a custom image with this change for yourself, depending on the version of cortex you want do use. The easiest way would be to use the official Cortex image as a base image in your Dockerfile, and then replace the file that needs to be updated using a COPY command. Here are instructions for using a custom docker image.

Also note that with this approach, threads_per_process will not be enforced.

We are not ready to merge #1565, because there are some known consequences of this that we would need to address (e.g. threads_per_process not being enforced).

Let me know if that works for you!

I'm getting this message in my logs when running my command:

Successfully built sse-starlette
Installing collected packages: sse-starlette
Successfully installed sse-starlette-0.6.1
[cont-init.d] bootloader.sh: exited 0.
[cont-init.d] done.
[services.d] starting services
[services.d] done.
2020-11-27 19:43:22.076150:cortex:pid-241:INFO:loading the predictor from predictor.py
2020-11-27 19:43:22.080023:cortex:pid-241:INFO:Started server process [241]
2020-11-27 19:43:22.080230:cortex:pid-241:INFO:Waiting for application startup.
2020-11-27 19:43:22.080905:cortex:pid-241:INFO:Application startup complete.
2020-11-27 19:43:22.081202:cortex:pid-241:INFO:Uvicorn running on unix socket /run/uvicorn/proc-0.sock (Press CTRL+C to quit)
[cont-finish.d] executing container finish scripts...
[cont-finish.d] done.
[s6-finish] waiting for services.
2020-11-27 19:43:29.310160:cortex:pid-237:INFO:Shutting down
2020-11-27 19:43:29.410600:cortex:pid-237:INFO:Waiting for application shutdown.
2020-11-27 19:43:29.410999:cortex:pid-237:INFO:Application shutdown complete.
2020-11-27 19:43:29.411105:cortex:pid-237:INFO:Finished server process [237]
s6-svwait: fatal: timed out
[s6-finish] sending all processes the TERM signal.
s6-svscanctl: fatal: unable to control /var/run/s6/services: supervisor not listening
[s6-finish] sending all processes the KILL signal and exiting.
2020-11-27 19:47:09.893313:cortex:pid-241:INFO:200 OK POST /predict

Your test command seems to be working though

[RobertLucian]

@MUTAL judging by what I see, that might be caused by an OOM (out-of-memory) error. You can check that by running cortex get <api-name> and see if it reports this error. The reason I'm leaning towards thinking this is an OOM error is that an OOM error usually kills a process right away, not giving it any chance of logging anything else. You can fix that by increasing the predictor.mem's value.

That being said, there's a possibility that something else is causing this. Could you also show us the implementation of your predictor?

[AbbeKamalov]

@MUTAL judging by what I see, that might be caused by an OOM (out-of-memory) error. You can check that by running cortex get <api-name> and see if it reports this error. The reason I'm leaning towards thinking this is an OOM error is that an OOM error usually kills a process right away, not giving it any chance of logging anything else. You can fix that by increasing the predictor.mem's value.

That being said, there's a possibility that something else is causing this. Could you also show us the implementation of your predictor?

I've changed predictor.mem's value accordingly and Cortex gives expected output, but not really as a stream? It seems that after the whole output is processed and it is only then it gets displayed, just like when using the usual api method.

Of course, here's my predictor:

import asyncio
from sse_starlette.sse import EventSourceResponse
import os
import boto3
import sys
import stat


async def run_command():
    st = os.stat('./gpt2tc')
    os.chmod('./gpt2tc', st.st_mode | stat.S_IEXEC ) 
    print('done')
    command = "./gpt2tc -m 117M -l 20 g hi"

    proc = await asyncio.create_subprocess_shell(
        command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )

    while True:
        output = await proc.stdout.readline()
        if not output:
            return
        yield output.decode().strip()


class PythonPredictor:
    def __init__(self, config):
        s3 = boto3.client('s3', aws_access_key_id='key',
        aws_secret_access_key='s_key')
        s3.download_file(config['bucket'], config['key'], "./gpt2_117M.bin")

    async def predict(self, payload):
        return EventSourceResponse(run_command())

If it might be helpful, I'm using this program while prototyping this.

[AbbeKamalov]

@MUTAL I think I have discovered an approach that will work for you.

This example that works for me:

cortex.yaml

- name: test
  kind: RealtimeAPI
  predictor:
    type: python
    image: quay.io/cortexlabsdev/python-predictor-cpu-async:0.23.0
    path: predictor.py

predictor.py

import asyncio
from sse_starlette.sse import EventSourceResponse


async def run_command():
    command = "echo test1 && sleep 2 && echo test2 && sleep 2 && echo test3 && echo done"

    proc = await asyncio.create_subprocess_shell(
        command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
    )

    while True:
        output = await proc.stdout.readline()
        if not output:
            return
        yield output.decode().strip()


class PythonPredictor:
    def __init__(self, config):
        pass

    async def predict(self, payload):
        return EventSourceResponse(run_command())

requirements.txt

sse-starlette

Note that I built a new version of the predictor image (specified in the image API configuration field), because it required a small change to one of our source files. It should be straightforward for you to build a custom image with this change for yourself, depending on the version of cortex you want do use. The easiest way would be to use the official Cortex image as a base image in your Dockerfile, and then replace the file that needs to be updated using a COPY command. Here are instructions for using a custom docker image.

Also note that with this approach, threads_per_process will not be enforced.

We are not ready to merge #1565, because there are some known consequences of this that we would need to address (e.g. threads_per_process not being enforced).

Let me know if that works for you!

@deliahu Your approach gives the desired output, but doesn't seem like it really streams it in realtime, but only after the whole output is processed, like when using the usual Cortex api method. Is this expected? Perhaps it is because Linux console can't show the stream, but only the finished output? I've provided my predictor.py if you would find it helpful.

[deliahu]

@MUTAL Are you using curl to make the request to your API? Also, if you change the command variable to the one I sent in my example, does it properly stream the output incrementally?

[AbbeKamalov]

Yes, I’m using curl to make the request and when using your command variable, the whole output (test1,test2,test3,done) is shown at once after a few seconds.

…

On Sat, 28 Nov 2020 at 20:57, David Eliahu ***@***.***> wrote: @MUTAL <https://github.com/mutal> Are you using curl to make the request to your API? Also, if you change the command variable to the one I sent in my example, does it properly stream the output incrementally? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#1500 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ADFOXIQ6LAHEKXRCIXXGTVDSSFI25ANCNFSM4TCIV5QA> .