How to manage dirty workers

[marcinkuzminski]

Hey,

Thanks for creating this feature! Looks awesome, i'm experimenting with it and run into an issue when i run 2 workers, with 2 dirty workers, and they perform a very long tasks, e.g 10 minutes token generation stream.

This works nice when handling 2 requests, but when 3rd comes there are no more dirty workers to handle extra requests.

I can't figure out how it's possible to monitor that and potentially kill or scale-up dirty workers ? Could you provide some examples how to do this ?

[benoitc]

Can you share a snippet of the code handling it?

If you are doing it asynchronously then the dirty worker is able to start a request in ytour app , when the result comes, your app send it back to the webapp . Let me know

[marcinkuzminski]

I'm not using async, i had hope to have it now sync, here's the code:

the gunicorn_test.dirty_app:ComputeApp streams bytes and can take 5-10min to process. But i'd love if i could somehow manage dirty workers and kill them or spawn more if needed.

def app(environ, start_response):
    """WSGI application that demonstrates dirty worker integration."""
    path = environ.get('PATH_INFO', '/')
    method = environ.get('REQUEST_METHOD', 'GET')

    # Parse query string
    query = parse_qs(environ.get('QUERY_STRING', ''))

    # Get dirty client
    client = get_dirty_client()

    try:
        if path == '/':
            result = {
                "message": "Dirty Workers Demo",
                "dirty_enabled": client is not None,
                "pid": os.getpid(),
                "endpoints": {
                    "/command": "Run command",
                    "/stats": "Get stats",
                }
            }
            start_response('200 OK', [('Content-Type', 'application/json')])
            return [json.dumps(result, indent=2).encode()]

        elif path == '/command':
            cmd = query.get('cmd')
            cmd = cmd[0] if cmd else ""
            if not cmd:
                start_response('400 Bad Request', [('Content-Type', 'application/json')])
                return [json.dumps({"error": "Missing cmd parameter"}).encode()]
            
            if client is None:
                start_response('500 Internal Server Error', [('Content-Type', 'application/json')])
                return [json.dumps({"error": "Dirty workers not enabled"}).encode()]

            print(f"Executing cmd: {cmd}")

            resp = client.execute("gunicorn_test.dirty_app:ComputeApp", "stats")
            print(resp)

            def generate():
                try:
                    for chunk in client.stream("gunicorn_test.dirty_app:ComputeApp", "cmd", cmd):
                        yield chunk.encode('utf-8') if isinstance(chunk, str) else chunk
                except DirtyTimeoutError:
                    print("closing client")
                    client.close()
                    yield b""

            # Stream response headers
            headers = [
                ('Content-Type', 'text/plain; charset=utf-8'),
                ('X-Accel-Buffering', 'no'),
                ('Cache-Control', 'no-cache'),
            ]
            start_response('200 OK', headers)
            return generate()

        elif path == '/stats':
            if client is None:
                start_response('500 Internal Server Error', [('Content-Type', 'application/json')])
                return [json.dumps({"error": "Dirty workers not enabled"}).encode()]

            compute_stats = client.execute(
                "my_app.dirty_app:ComputeApp",
                "stats"
            )
            result = {
                "compute_app": compute_stats,
                "http_worker_pid": os.getpid(),
            }
            start_response('200 OK', [('Content-Type', 'application/json')])
            return [json.dumps(result, indent=2).encode()]

        else:
            # 404 for unknown paths
            start_response('404 Not Found', [('Content-Type', 'application/json')])
            return [json.dumps({"error": "Not found"}).encode()]

    except Exception as e:
        start_response('500 Internal Server Error', [('Content-Type', 'application/json')])
        return [json.dumps({
            "error": str(e),
            "type": type(e).__name__
        }).encode()]

[benoitc]

@marcinkuzminski you means beeing able to increase the number of workers like TTIN/TTOU on current web arbiter or something else,?

[benoitc]

@marcinkuzminski You can now do it with the 25.1.0 release.

2 ways:

• using the new gunicornc command line: https://gunicorn.org/guides/gunicornc
• using signaling: https://gunicorn.org/dirty/#dynamic-scaling-with-ttinttou

[marcinkuzminski]

Sorry i somehow missed your question previously, Yes this looks great, so within a code i can search for the dirty arbiter PID (would be lovely if client have that ?) and spawn / reduce ad-hoc.

One thing i'm not clear about is how i'd track all dirty workers are busy, because afair when calling client one all busy workers i get no response...

My use case here would be (based on my code)

# before calling it, i'd like to understand the pool of dirty workers that are pending works vs these that are busy
resp = client.execute("gunicorn_test.dirty_app:ComputeApp", "stats")

[benoitc]

ah for now the only way to do it is by catching the exceptions. There is no direct support of metrics to do it internally yet.

https://gunicorn.org/dirty/#error-handling_2

I'm wondering though if your issue couldn't be fixed by having ssome kind of internal task system , so you can spawn a work from your app and place the result back in the queue. Would it work ?

Also gunicorrnc can return results as json, so you can write a monitor that spawn a new worker when needed it using gunicornc to query the state.

[marcinkuzminski]

hmm not sure, what i wanted here is the stream support, so basically how i imagine this to work is i use dirty workers to handle heavy operations and stream back the work items to the client (that happens right away), usually these outputs are very fast, but i have few that takes a long time, so i was looking for a way to spawn more dirty workers ad-hoc.

I guess catching exception on timeout here would allow me to handle that case, i'll test it out and let you know

[benoitc]

@marcinkuzminski note that using a gthread worker for web app doesn't stop you to use asyncio in the dirty app. I think I will play we the idea to delay some work in a function so you don't have to block at all there. That should be doable. Will test it aside.

[marcinkuzminski]

hmm, interesting i'll play with this idea too