in src/functions_framework/_http/gunicorn.py [0:0]
def __init__(self, app, host, port, debug, **options):
threads = int(os.environ.get("THREADS", (os.cpu_count() or 1) * 4))
global TIMEOUT_SECONDS
TIMEOUT_SECONDS = int(os.environ.get("CLOUD_RUN_TIMEOUT_SECONDS", 0))
self.options = {
"bind": "%s:%s" % (host, port),
"workers": int(os.environ.get("WORKERS", 1)),
"threads": threads,
"loglevel": os.environ.get("GUNICORN_LOG_LEVEL", "error"),
"limit_request_line": 0,
}
if (
TIMEOUT_SECONDS > 0
and threads > 1
and (os.environ.get("THREADED_TIMEOUT_ENABLED", "False").lower() == "true")
): # pragma: no cover
self.options["worker_class"] = (
"functions_framework._http.gunicorn.GThreadWorkerWithTimeoutSupport"
)
else:
self.options["timeout"] = TIMEOUT_SECONDS
self.options.update(options)
self.app = app
super().__init__()