ribbit
ribbit
PostsComments
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
will try to do so
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
the normal streaming works well, now we have 2 separate endpoints one to handle all non-streaming and one to handle all the streaming hahah
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
{
  "delayTime": 9289,
  "executionTime": 54,
  "id": "bc6aaadd-0970-4100-99a0-f4230d18be4f-e1",
  "output": [],
  "status": "COMPLETED"
}
{
  "delayTime": 9289,
  "executionTime": 54,
  "id": "bc6aaadd-0970-4100-99a0-f4230d18be4f-e1",
  "output": [],
  "status": "COMPLETED"
}
returned a generator still, same code ran on serverless. no log to prove that it's a generator tho but it has the same behavior as before when it casted all my outputs to be a generator, it returns a [] and empty stream when streamed.
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
thank you all
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
def dummy(job):
    something = False
    if something:
        yield 1
    else:
        return 1


if __name__ == "__main__":
    runpod.serverless.start(
        {
            "handler": dummy,
            "return_aggregate_stream": True
        }
    )
def dummy(job):
    something = False
    if something:
        yield 1
    else:
        return 1


if __name__ == "__main__":
    runpod.serverless.start(
        {
            "handler": dummy,
            "return_aggregate_stream": True
        }
    )
this code also returns a generator 
INFO   | local_test | Started.
DEBUG  | local_test | Handler output: <generator object dummy at 0x7376fcbdedc0>
DEBUG  | local_test | run_job return: {'output': <generator object dummy at 0x7376fcbdedc0>}
INFO   | Job local_test completed successfully.
INFO   | Job result: {'output': <generator object dummy at 0x7376fcbdedc0>}
INFO   | Local testing complete, exiting.
INFO   | local_test | Started.
DEBUG  | local_test | Handler output: <generator object dummy at 0x7376fcbdedc0>
DEBUG  | local_test | run_job return: {'output': <generator object dummy at 0x7376fcbdedc0>}
INFO   | Job local_test completed successfully.
INFO   | Job result: {'output': <generator object dummy at 0x7376fcbdedc0>}
INFO   | Local testing complete, exiting.
I think it's just the way it is
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
because the change of the response type would disrupt other running services, i am to avoid that
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
i am very sure that this is not the case, tried and checked every other function, this stream function is a new addition to the code and before that (all the function above of the stream_func is the original code) no function ever coded to return a generator
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
def invoke(
        endpoint: str,
        payload: dict
) -> Any:
    if endpoint == "...":
        response = some_func() # this does not return a generator
    elif endpoint == "...":
        response = some_func() # this does not return a generator
    elif endpoint == "answer_stream":
        response = stream_func() # this returns a generator
    
    return response
def invoke(
        endpoint: str,
        payload: dict
) -> Any:
    if endpoint == "...":
        response = some_func() # this does not return a generator
    elif endpoint == "...":
        response = some_func() # this does not return a generator
    elif endpoint == "answer_stream":
        response = stream_func() # this returns a generator
    
    return response
this is how the invoke method is, i omitted irrelevant codes but that's generally how the function is. 
endpoint, validated_payload = validate_payload(job)
endpoint, validated_payload = validate_payload(job)
I got endpoint from this line in the handler function, which will pass job into a validator function that simply check if my payload schema is ok. basically the inputted payload should look like this: 
{
  "input": {
    "api": {
      "endpoint": "...",
    },
    ...rest of the json
  }
}
{
  "input": {
    "api": {
      "endpoint": "...",
    },
    ...rest of the json
  }
}
basically that function validates and extract the endpoint variable from my input payload
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
the endpoint's value is not stream_answer, yet it always returns a generator
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
STREAM_ENDPOINTS = {"stream_answer"}

def handler(job):
    validated_input = validate_input(job)
    if "errors" in validated_input:
        return {
            "error": "\n".join(validated_input["errors"])
        }

    validated_api = validate_api(job)
    if "errors" in validated_api:
        return {
            "error": "\n".join(validated_api["errors"])
        }

    endpoint, validated_payload = validate_payload(job)
    if "errors" in validated_payload:
        return {
            "error": "\n".join(validated_payload["errors"])
        }

    if "validated_input" in validated_payload:
        payload = validated_payload["validated_input"]
    else:
        payload = validated_payload

    if payload.get("emu_error", False):
        return {
            "error": "Error emulated. If this is unintended, please check your environment variables"
        }

    LOGGER.info(f"[INVOKING] {endpoint}", job["id"])
    LOGGER.info(f"[SUCCESS] Returning")

    response = invoke(endpoint, payload)

    if endpoint in STREAM_ENDPOINTS:
        for token in response:
            yield token
    else:
        return response
STREAM_ENDPOINTS = {"stream_answer"}

def handler(job):
    validated_input = validate_input(job)
    if "errors" in validated_input:
        return {
            "error": "\n".join(validated_input["errors"])
        }

    validated_api = validate_api(job)
    if "errors" in validated_api:
        return {
            "error": "\n".join(validated_api["errors"])
        }

    endpoint, validated_payload = validate_payload(job)
    if "errors" in validated_payload:
        return {
            "error": "\n".join(validated_payload["errors"])
        }

    if "validated_input" in validated_payload:
        payload = validated_payload["validated_input"]
    else:
        payload = validated_payload

    if payload.get("emu_error", False):
        return {
            "error": "Error emulated. If this is unintended, please check your environment variables"
        }

    LOGGER.info(f"[INVOKING] {endpoint}", job["id"])
    LOGGER.info(f"[SUCCESS] Returning")

    response = invoke(endpoint, payload)

    if endpoint in STREAM_ENDPOINTS:
        for token in response:
            yield token
    else:
        return response
output 
INFO   | local_test | Started.
DEBUG  | local_test | Handler output: <generator object handler at 0x7d0f40748890>
DEBUG  | local_test | run_job return: {'output': <generator object handler at 0x7d0f40748890>}
INFO   | Job local_test completed successfully.
INFO   | Job result: {'output': <generator object handler at 0x7d0f40748890>}
INFO   | Local testing complete, exiting.
INFO   | local_test | Started.
DEBUG  | local_test | Handler output: <generator object handler at 0x7d0f40748890>
DEBUG  | local_test | run_job return: {'output': <generator object handler at 0x7d0f40748890>}
INFO   | Job local_test completed successfully.
INFO   | Job result: {'output': <generator object handler at 0x7d0f40748890>}
INFO   | Local testing complete, exiting.
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
that's from the runpod local dev server log, the one that shows the output of the handler function
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
haha sorry was dizzy, i meant /run
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
sorry i can't produce screenshot right now, but in the local testing library, the output is somewhat like this: 
...runpod logs...
output: <generator object ...>
...runpod logs...
...runpod logs...
output: <generator object ...>
...runpod logs...
assume that this is my code 
def handler():
    value = 1
    something = False
    if something:
        yield value # output is a generator
    else:
        return value # this also outputs a generator
def handler():
    value = 1
    something = False
    if something:
        yield value # output is a generator
    else:
        return value # this also outputs a generator
it is expected if something is True, but somehow when something is False, it returns a generator as well
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
i hit the /run endpoint first, then retrieve the stream by hitting /stream
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
yeah that's not how, i just wrote the example that way for convinience
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
sorry, I mean whenever a yield is present in the handler, the output becomes a generator regardless if I use yield or return. for example 
def handler():
    if something:
        yield value # output is a generator
    else:
        return value # this also outputs a generator
def handler():
    if something:
        yield value # output is a generator
    else:
        return value # this also outputs a generator
67 replies
RRunPod
Created by ribbit on 4/30/2024 in #⚡｜serverless
How do I handle both streaming and non-streaming request in a serverless pod?
no unfortunately it's not easy to test locally, but I tried deploying it anyway and turns out whenever there's yield in the handler function everything I return becomes a generator? I can't get it to work yet
67 replies
RRunPod
Created by ribbit on 4/17/2024 in #⚡｜serverless
Connection reset by peer
yeah hahahah, things are ok up till now thanks all
34 replies
RRunPod
Created by ribbit on 4/17/2024 in #⚡｜serverless
Connection reset by peer
now it's stable, i think they fixed it
34 replies
RRunPod
Created by ribbit on 4/17/2024 in #⚡｜serverless
Connection reset by peer
kk
34 replies