We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Make default asyncio workers proportional to cpu-limit
/kind feature

Currently a custom transformer spawns threads promotional to the number of cores on the machine/node. This might be due to python (and tornatdo, asyncio) not being container aware. On nodes with high core count, more threads cause throttling when cpu-limit is comparatively low

Making the number of threads spawned proportional to cpu-limit would help with throttling.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/kfserving/kfserving/utils/utils.py`
Content:
```
1 # Copyright 2020 kubeflow.org.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #    http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import os
16 
17 
18 def is_running_in_k8s():
19     return os.path.isdir('/var/run/secrets/kubernetes.io/')
20 
21 
22 def get_current_k8s_namespace():
23     with open('/var/run/secrets/kubernetes.io/serviceaccount/namespace', 'r') as f:
24         return f.readline()
25 
26 
27 def get_default_target_namespace():
28     if not is_running_in_k8s():
29         return 'default'
30     return get_current_k8s_namespace()
31 
32 
33 def set_isvc_namespace(inferenceservice):
34     isvc_namespace = inferenceservice.metadata.namespace
35     namespace = isvc_namespace or get_default_target_namespace()
36     return namespace
37
```

Path: `python/kfserving/kfserving/kfserver.py`
Content:
```
1 # Copyright 2020 kubeflow.org.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #    http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import argparse
16 import logging
17 import json
18 import inspect
19 import sys
20 from typing import List, Optional
21 import tornado.ioloop
22 import tornado.web
23 import tornado.httpserver
24 import tornado.log
25 
26 from kfserving.handlers.http import PredictHandler, ExplainHandler
27 from kfserving import KFModel
28 from kfserving.kfmodel_repository import KFModelRepository
29 
30 DEFAULT_HTTP_PORT = 8080
31 DEFAULT_GRPC_PORT = 8081
32 DEFAULT_MAX_BUFFER_SIZE = 104857600
33 
34 parser = argparse.ArgumentParser(add_help=False)
35 parser.add_argument('--http_port', default=DEFAULT_HTTP_PORT, type=int,
36                     help='The HTTP Port listened to by the model server.')
37 parser.add_argument('--grpc_port', default=DEFAULT_GRPC_PORT, type=int,
38                     help='The GRPC Port listened to by the model server.')
39 parser.add_argument('--max_buffer_size', default=DEFAULT_MAX_BUFFER_SIZE, type=int,
40                     help='The max buffer size for tornado.')
41 parser.add_argument('--workers', default=1, type=int,
42                     help='The number of works to fork')
43 args, _ = parser.parse_known_args()
44 
45 tornado.log.enable_pretty_logging()
46 
47 
48 class KFServer:
49     def __init__(self, http_port: int = args.http_port,
50                  grpc_port: int = args.grpc_port,
51                  max_buffer_size: int = args.max_buffer_size,
52                  workers: int = args.workers,
53                  registered_models: KFModelRepository = KFModelRepository()):
54         self.registered_models = registered_models
55         self.http_port = http_port
56         self.grpc_port = grpc_port
57         self.max_buffer_size = max_buffer_size
58         self.workers = workers
59         self._http_server: Optional[tornado.httpserver.HTTPServer] = None
60 
61     def create_application(self):
62         return tornado.web.Application([
63             # Server Liveness API returns 200 if server is alive.
64             (r"/", LivenessHandler),
65             (r"/v2/health/live", LivenessHandler),
66             (r"/v1/models",
67              ListHandler, dict(models=self.registered_models)),
68             (r"/v2/models",
69              ListHandler, dict(models=self.registered_models)),
70             # Model Health API returns 200 if model is ready to serve.
71             (r"/v1/models/([a-zA-Z0-9_-]+)",
72              HealthHandler, dict(models=self.registered_models)),
73             (r"/v2/models/([a-zA-Z0-9_-]+)/status",
74              HealthHandler, dict(models=self.registered_models)),
75             (r"/v1/models/([a-zA-Z0-9_-]+):predict",
76              PredictHandler, dict(models=self.registered_models)),
77             (r"/v2/models/([a-zA-Z0-9_-]+)/infer",
78              PredictHandler, dict(models=self.registered_models)),
79             (r"/v1/models/([a-zA-Z0-9_-]+):explain",
80              ExplainHandler, dict(models=self.registered_models)),
81             (r"/v2/models/([a-zA-Z0-9_-]+)/explain",
82              ExplainHandler, dict(models=self.registered_models)),
83             (r"/v2/repository/models/([a-zA-Z0-9_-]+)/load",
84              LoadHandler, dict(models=self.registered_models)),
85             (r"/v2/repository/models/([a-zA-Z0-9_-]+)/unload",
86              UnloadHandler, dict(models=self.registered_models)),
87         ])
88 
89     def start(self, models: List[KFModel], nest_asyncio: bool = False):
90         for model in models:
91             self.register_model(model)
92 
93         self._http_server = tornado.httpserver.HTTPServer(
94             self.create_application(), max_buffer_size=self.max_buffer_size)
95 
96         logging.info("Listening on port %s", self.http_port)
97         self._http_server.bind(self.http_port)
98         logging.info("Will fork %d workers", self.workers)
99         self._http_server.start(self.workers)
100 
101         # Need to start the IOLoop after workers have been started
102         # https://github.com/tornadoweb/tornado/issues/2426
103         # The nest_asyncio package needs to be installed by the downstream module
104         if nest_asyncio:
105             import nest_asyncio
106             nest_asyncio.apply()
107 
108         tornado.ioloop.IOLoop.current().start()
109 
110     def register_model(self, model: KFModel):
111         if not model.name:
112             raise Exception(
113                 "Failed to register model, model.name must be provided.")
114         self.registered_models.update(model)
115         logging.info("Registering model: %s", model.name)
116 
117 
118 class LivenessHandler(tornado.web.RequestHandler):  # pylint:disable=too-few-public-methods
119     def get(self):
120         self.write("Alive")
121 
122 
123 class HealthHandler(tornado.web.RequestHandler):
124     def initialize(self, models: KFModelRepository):
125         self.models = models  # pylint:disable=attribute-defined-outside-init
126 
127     def get(self, name: str):
128         model = self.models.get_model(name)
129         if model is None:
130             raise tornado.web.HTTPError(
131                 status_code=404,
132                 reason="Model with name %s does not exist." % name
133             )
134 
135         if not model.ready:
136             raise tornado.web.HTTPError(
137                 status_code=503,
138                 reason="Model with name %s is not ready." % name
139             )
140 
141         self.write(json.dumps({
142             "name": model.name,
143             "ready": model.ready
144         }))
145 
146 
147 class ListHandler(tornado.web.RequestHandler):
148     def initialize(self, models: KFModelRepository):
149         self.models = models  # pylint:disable=attribute-defined-outside-init
150 
151     def get(self):
152         self.write(json.dumps([ob.name for ob in self.models.get_models()]))
153 
154 
155 class LoadHandler(tornado.web.RequestHandler):
156     def initialize(self, models: KFModelRepository):  # pylint:disable=attribute-defined-outside-init
157         self.models = models
158 
159     async def post(self, name: str):
160         try:
161             (await self.models.load(name)) if inspect.iscoroutinefunction(self.models.load) else self.models.load(name)
162         except Exception:
163             ex_type, ex_value, ex_traceback = sys.exc_info()
164             raise tornado.web.HTTPError(
165                 status_code=500,
166                 reason=f"Model with name {name} is not ready. "
167                        f"Error type: {ex_type} error msg: {ex_value}"
168             )
169 
170         if not self.models.is_model_ready(name):
171             raise tornado.web.HTTPError(
172                 status_code=503,
173                 reason=f"Model with name {name} is not ready."
174             )
175         self.write(json.dumps({
176             "name": name,
177             "load": True
178         }))
179 
180 
181 class UnloadHandler(tornado.web.RequestHandler):
182     def initialize(self, models: KFModelRepository):  # pylint:disable=attribute-defined-outside-init
183         self.models = models
184 
185     def post(self, name: str):
186         try:
187             self.models.unload(name)
188         except KeyError:
189             raise tornado.web.HTTPError(
190                 status_code=404,
191                 reason="Model with name %s does not exist." % name
192             )
193         self.write(json.dumps({
194             "name": name,
195             "unload": True
196         }))
197
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/kfserving/kfserving/kfserver.py b/python/kfserving/kfserving/kfserver.py
--- a/python/kfserving/kfserving/kfserver.py
+++ b/python/kfserving/kfserving/kfserver.py
@@ -22,6 +22,9 @@
 import tornado.web
 import tornado.httpserver
 import tornado.log
+import asyncio
+from tornado import concurrent
+from .utils import utils
 
 from kfserving.handlers.http import PredictHandler, ExplainHandler
 from kfserving import KFModel
@@ -40,6 +43,8 @@
                     help='The max buffer size for tornado.')
 parser.add_argument('--workers', default=1, type=int,
                     help='The number of works to fork')
+parser.add_argument('--max_asyncio_workers', default=None, type=int,
+                    help='Max number of asyncio workers to spawn')
 args, _ = parser.parse_known_args()
 
 tornado.log.enable_pretty_logging()
@@ -50,12 +55,14 @@
                  grpc_port: int = args.grpc_port,
                  max_buffer_size: int = args.max_buffer_size,
                  workers: int = args.workers,
+                 max_asyncio_workers: int = args.max_asyncio_workers,
                  registered_models: KFModelRepository = KFModelRepository()):
         self.registered_models = registered_models
         self.http_port = http_port
         self.grpc_port = grpc_port
         self.max_buffer_size = max_buffer_size
         self.workers = workers
+        self.max_asyncio_workers = max_asyncio_workers
         self._http_server: Optional[tornado.httpserver.HTTPServer] = None
 
     def create_application(self):
@@ -87,6 +94,14 @@
         ])
 
     def start(self, models: List[KFModel], nest_asyncio: bool = False):
+        if self.max_asyncio_workers is None:
+            # formula as suggest in https://bugs.python.org/issue35279
+            self.max_asyncio_workers = min(32, utils.cpu_count()+4)
+
+        logging.info(f"Setting asyncio max_workers as {self.max_asyncio_workers}")
+        asyncio.get_event_loop().set_default_executor(
+            concurrent.futures.ThreadPoolExecutor(max_workers=self.max_asyncio_workers))
+
         for model in models:
             self.register_model(model)
 
diff --git a/python/kfserving/kfserving/utils/utils.py b/python/kfserving/kfserving/utils/utils.py
--- a/python/kfserving/kfserving/utils/utils.py
+++ b/python/kfserving/kfserving/utils/utils.py
@@ -13,6 +13,8 @@
 # limitations under the License.
 
 import os
+import sys
+import psutil
 
 
 def is_running_in_k8s():
@@ -34,3 +36,36 @@
     isvc_namespace = inferenceservice.metadata.namespace
     namespace = isvc_namespace or get_default_target_namespace()
     return namespace
+
+
+def cpu_count():
+    """Get the available CPU count for this system.
+    Takes the minimum value from the following locations:
+    - Total system cpus available on the host.
+    - CPU Affinity (if set)
+    - Cgroups limit (if set)
+    """
+    count = os.cpu_count()
+
+    # Check CPU affinity if available
+    try:
+        affinity_count = len(psutil.Process().cpu_affinity())
+        if affinity_count > 0:
+            count = min(count, affinity_count)
+    except Exception:
+        pass
+
+    # Check cgroups if available
+    if sys.platform == "linux":
+        try:
+            with open("/sys/fs/cgroup/cpu,cpuacct/cpu.cfs_quota_us") as f:
+                quota = int(f.read())
+            with open("/sys/fs/cgroup/cpu,cpuacct/cpu.cfs_period_us") as f:
+                period = int(f.read())
+            cgroups_count = int(quota / period)
+            if cgroups_count > 0:
+                count = min(count, cgroups_count)
+        except Exception:
+            pass
+
+    return count

{"golden_diff": "diff --git a/python/kfserving/kfserving/kfserver.py b/python/kfserving/kfserving/kfserver.py\n--- a/python/kfserving/kfserving/kfserver.py\n+++ b/python/kfserving/kfserving/kfserver.py\n@@ -22,6 +22,9 @@\n import tornado.web\n import tornado.httpserver\n import tornado.log\n+import asyncio\n+from tornado import concurrent\n+from .utils import utils\n \n from kfserving.handlers.http import PredictHandler, ExplainHandler\n from kfserving import KFModel\n@@ -40,6 +43,8 @@\n                     help='The max buffer size for tornado.')\n parser.add_argument('--workers', default=1, type=int,\n                     help='The number of works to fork')\n+parser.add_argument('--max_asyncio_workers', default=None, type=int,\n+                    help='Max number of asyncio workers to spawn')\n args, _ = parser.parse_known_args()\n \n tornado.log.enable_pretty_logging()\n@@ -50,12 +55,14 @@\n                  grpc_port: int = args.grpc_port,\n                  max_buffer_size: int = args.max_buffer_size,\n                  workers: int = args.workers,\n+                 max_asyncio_workers: int = args.max_asyncio_workers,\n                  registered_models: KFModelRepository = KFModelRepository()):\n         self.registered_models = registered_models\n         self.http_port = http_port\n         self.grpc_port = grpc_port\n         self.max_buffer_size = max_buffer_size\n         self.workers = workers\n+        self.max_asyncio_workers = max_asyncio_workers\n         self._http_server: Optional[tornado.httpserver.HTTPServer] = None\n \n     def create_application(self):\n@@ -87,6 +94,14 @@\n         ])\n \n     def start(self, models: List[KFModel], nest_asyncio: bool = False):\n+        if self.max_asyncio_workers is None:\n+            # formula as suggest in https://bugs.python.org/issue35279\n+            self.max_asyncio_workers = min(32, utils.cpu_count()+4)\n+\n+        logging.info(f\"Setting asyncio max_workers as {self.max_asyncio_workers}\")\n+        asyncio.get_event_loop().set_default_executor(\n+            concurrent.futures.ThreadPoolExecutor(max_workers=self.max_asyncio_workers))\n+\n         for model in models:\n             self.register_model(model)\n \ndiff --git a/python/kfserving/kfserving/utils/utils.py b/python/kfserving/kfserving/utils/utils.py\n--- a/python/kfserving/kfserving/utils/utils.py\n+++ b/python/kfserving/kfserving/utils/utils.py\n@@ -13,6 +13,8 @@\n # limitations under the License.\n \n import os\n+import sys\n+import psutil\n \n \n def is_running_in_k8s():\n@@ -34,3 +36,36 @@\n     isvc_namespace = inferenceservice.metadata.namespace\n     namespace = isvc_namespace or get_default_target_namespace()\n     return namespace\n+\n+\n+def cpu_count():\n+    \"\"\"Get the available CPU count for this system.\n+    Takes the minimum value from the following locations:\n+    - Total system cpus available on the host.\n+    - CPU Affinity (if set)\n+    - Cgroups limit (if set)\n+    \"\"\"\n+    count = os.cpu_count()\n+\n+    # Check CPU affinity if available\n+    try:\n+        affinity_count = len(psutil.Process().cpu_affinity())\n+        if affinity_count > 0:\n+            count = min(count, affinity_count)\n+    except Exception:\n+        pass\n+\n+    # Check cgroups if available\n+    if sys.platform == \"linux\":\n+        try:\n+            with open(\"/sys/fs/cgroup/cpu,cpuacct/cpu.cfs_quota_us\") as f:\n+                quota = int(f.read())\n+            with open(\"/sys/fs/cgroup/cpu,cpuacct/cpu.cfs_period_us\") as f:\n+                period = int(f.read())\n+            cgroups_count = int(quota / period)\n+            if cgroups_count > 0:\n+                count = min(count, cgroups_count)\n+        except Exception:\n+            pass\n+\n+    return count\n", "issue": "Make default asyncio workers proportional to cpu-limit\n/kind feature\r\n\r\nCurrently a custom transformer spawns threads promotional to the number of cores on the machine/node. This might be due to python (and tornatdo, asyncio) not being container aware. On nodes with high core count, more threads cause throttling when cpu-limit is comparatively low\r\n\r\nMaking the number of threads spawned proportional to cpu-limit would help with throttling.\r\n\n", "before_files": [{"content": "# Copyright 2020 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport os\n\n\ndef is_running_in_k8s():\n    return os.path.isdir('/var/run/secrets/kubernetes.io/')\n\n\ndef get_current_k8s_namespace():\n    with open('/var/run/secrets/kubernetes.io/serviceaccount/namespace', 'r') as f:\n        return f.readline()\n\n\ndef get_default_target_namespace():\n    if not is_running_in_k8s():\n        return 'default'\n    return get_current_k8s_namespace()\n\n\ndef set_isvc_namespace(inferenceservice):\n    isvc_namespace = inferenceservice.metadata.namespace\n    namespace = isvc_namespace or get_default_target_namespace()\n    return namespace\n", "path": "python/kfserving/kfserving/utils/utils.py"}, {"content": "# Copyright 2020 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport argparse\nimport logging\nimport json\nimport inspect\nimport sys\nfrom typing import List, Optional\nimport tornado.ioloop\nimport tornado.web\nimport tornado.httpserver\nimport tornado.log\n\nfrom kfserving.handlers.http import PredictHandler, ExplainHandler\nfrom kfserving import KFModel\nfrom kfserving.kfmodel_repository import KFModelRepository\n\nDEFAULT_HTTP_PORT = 8080\nDEFAULT_GRPC_PORT = 8081\nDEFAULT_MAX_BUFFER_SIZE = 104857600\n\nparser = argparse.ArgumentParser(add_help=False)\nparser.add_argument('--http_port', default=DEFAULT_HTTP_PORT, type=int,\n                    help='The HTTP Port listened to by the model server.')\nparser.add_argument('--grpc_port', default=DEFAULT_GRPC_PORT, type=int,\n                    help='The GRPC Port listened to by the model server.')\nparser.add_argument('--max_buffer_size', default=DEFAULT_MAX_BUFFER_SIZE, type=int,\n                    help='The max buffer size for tornado.')\nparser.add_argument('--workers', default=1, type=int,\n                    help='The number of works to fork')\nargs, _ = parser.parse_known_args()\n\ntornado.log.enable_pretty_logging()\n\n\nclass KFServer:\n    def __init__(self, http_port: int = args.http_port,\n                 grpc_port: int = args.grpc_port,\n                 max_buffer_size: int = args.max_buffer_size,\n                 workers: int = args.workers,\n                 registered_models: KFModelRepository = KFModelRepository()):\n        self.registered_models = registered_models\n        self.http_port = http_port\n        self.grpc_port = grpc_port\n        self.max_buffer_size = max_buffer_size\n        self.workers = workers\n        self._http_server: Optional[tornado.httpserver.HTTPServer] = None\n\n    def create_application(self):\n        return tornado.web.Application([\n            # Server Liveness API returns 200 if server is alive.\n            (r\"/\", LivenessHandler),\n            (r\"/v2/health/live\", LivenessHandler),\n            (r\"/v1/models\",\n             ListHandler, dict(models=self.registered_models)),\n            (r\"/v2/models\",\n             ListHandler, dict(models=self.registered_models)),\n            # Model Health API returns 200 if model is ready to serve.\n            (r\"/v1/models/([a-zA-Z0-9_-]+)\",\n             HealthHandler, dict(models=self.registered_models)),\n            (r\"/v2/models/([a-zA-Z0-9_-]+)/status\",\n             HealthHandler, dict(models=self.registered_models)),\n            (r\"/v1/models/([a-zA-Z0-9_-]+):predict\",\n             PredictHandler, dict(models=self.registered_models)),\n            (r\"/v2/models/([a-zA-Z0-9_-]+)/infer\",\n             PredictHandler, dict(models=self.registered_models)),\n            (r\"/v1/models/([a-zA-Z0-9_-]+):explain\",\n             ExplainHandler, dict(models=self.registered_models)),\n            (r\"/v2/models/([a-zA-Z0-9_-]+)/explain\",\n             ExplainHandler, dict(models=self.registered_models)),\n            (r\"/v2/repository/models/([a-zA-Z0-9_-]+)/load\",\n             LoadHandler, dict(models=self.registered_models)),\n            (r\"/v2/repository/models/([a-zA-Z0-9_-]+)/unload\",\n             UnloadHandler, dict(models=self.registered_models)),\n        ])\n\n    def start(self, models: List[KFModel], nest_asyncio: bool = False):\n        for model in models:\n            self.register_model(model)\n\n        self._http_server = tornado.httpserver.HTTPServer(\n            self.create_application(), max_buffer_size=self.max_buffer_size)\n\n        logging.info(\"Listening on port %s\", self.http_port)\n        self._http_server.bind(self.http_port)\n        logging.info(\"Will fork %d workers\", self.workers)\n        self._http_server.start(self.workers)\n\n        # Need to start the IOLoop after workers have been started\n        # https://github.com/tornadoweb/tornado/issues/2426\n        # The nest_asyncio package needs to be installed by the downstream module\n        if nest_asyncio:\n            import nest_asyncio\n            nest_asyncio.apply()\n\n        tornado.ioloop.IOLoop.current().start()\n\n    def register_model(self, model: KFModel):\n        if not model.name:\n            raise Exception(\n                \"Failed to register model, model.name must be provided.\")\n        self.registered_models.update(model)\n        logging.info(\"Registering model: %s\", model.name)\n\n\nclass LivenessHandler(tornado.web.RequestHandler):  # pylint:disable=too-few-public-methods\n    def get(self):\n        self.write(\"Alive\")\n\n\nclass HealthHandler(tornado.web.RequestHandler):\n    def initialize(self, models: KFModelRepository):\n        self.models = models  # pylint:disable=attribute-defined-outside-init\n\n    def get(self, name: str):\n        model = self.models.get_model(name)\n        if model is None:\n            raise tornado.web.HTTPError(\n                status_code=404,\n                reason=\"Model with name %s does not exist.\" % name\n            )\n\n        if not model.ready:\n            raise tornado.web.HTTPError(\n                status_code=503,\n                reason=\"Model with name %s is not ready.\" % name\n            )\n\n        self.write(json.dumps({\n            \"name\": model.name,\n            \"ready\": model.ready\n        }))\n\n\nclass ListHandler(tornado.web.RequestHandler):\n    def initialize(self, models: KFModelRepository):\n        self.models = models  # pylint:disable=attribute-defined-outside-init\n\n    def get(self):\n        self.write(json.dumps([ob.name for ob in self.models.get_models()]))\n\n\nclass LoadHandler(tornado.web.RequestHandler):\n    def initialize(self, models: KFModelRepository):  # pylint:disable=attribute-defined-outside-init\n        self.models = models\n\n    async def post(self, name: str):\n        try:\n            (await self.models.load(name)) if inspect.iscoroutinefunction(self.models.load) else self.models.load(name)\n        except Exception:\n            ex_type, ex_value, ex_traceback = sys.exc_info()\n            raise tornado.web.HTTPError(\n                status_code=500,\n                reason=f\"Model with name {name} is not ready. \"\n                       f\"Error type: {ex_type} error msg: {ex_value}\"\n            )\n\n        if not self.models.is_model_ready(name):\n            raise tornado.web.HTTPError(\n                status_code=503,\n                reason=f\"Model with name {name} is not ready.\"\n            )\n        self.write(json.dumps({\n            \"name\": name,\n            \"load\": True\n        }))\n\n\nclass UnloadHandler(tornado.web.RequestHandler):\n    def initialize(self, models: KFModelRepository):  # pylint:disable=attribute-defined-outside-init\n        self.models = models\n\n    def post(self, name: str):\n        try:\n            self.models.unload(name)\n        except KeyError:\n            raise tornado.web.HTTPError(\n                status_code=404,\n                reason=\"Model with name %s does not exist.\" % name\n            )\n        self.write(json.dumps({\n            \"name\": name,\n            \"unload\": True\n        }))\n", "path": "python/kfserving/kfserving/kfserver.py"}], "after_files": [{"content": "# Copyright 2020 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport os\nimport sys\nimport psutil\n\n\ndef is_running_in_k8s():\n    return os.path.isdir('/var/run/secrets/kubernetes.io/')\n\n\ndef get_current_k8s_namespace():\n    with open('/var/run/secrets/kubernetes.io/serviceaccount/namespace', 'r') as f:\n        return f.readline()\n\n\ndef get_default_target_namespace():\n    if not is_running_in_k8s():\n        return 'default'\n    return get_current_k8s_namespace()\n\n\ndef set_isvc_namespace(inferenceservice):\n    isvc_namespace = inferenceservice.metadata.namespace\n    namespace = isvc_namespace or get_default_target_namespace()\n    return namespace\n\n\ndef cpu_count():\n    \"\"\"Get the available CPU count for this system.\n    Takes the minimum value from the following locations:\n    - Total system cpus available on the host.\n    - CPU Affinity (if set)\n    - Cgroups limit (if set)\n    \"\"\"\n    count = os.cpu_count()\n\n    # Check CPU affinity if available\n    try:\n        affinity_count = len(psutil.Process().cpu_affinity())\n        if affinity_count > 0:\n            count = min(count, affinity_count)\n    except Exception:\n        pass\n\n    # Check cgroups if available\n    if sys.platform == \"linux\":\n        try:\n            with open(\"/sys/fs/cgroup/cpu,cpuacct/cpu.cfs_quota_us\") as f:\n                quota = int(f.read())\n            with open(\"/sys/fs/cgroup/cpu,cpuacct/cpu.cfs_period_us\") as f:\n                period = int(f.read())\n            cgroups_count = int(quota / period)\n            if cgroups_count > 0:\n                count = min(count, cgroups_count)\n        except Exception:\n            pass\n\n    return count\n", "path": "python/kfserving/kfserving/utils/utils.py"}, {"content": "# Copyright 2020 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport argparse\nimport logging\nimport json\nimport inspect\nimport sys\nfrom typing import List, Optional\nimport tornado.ioloop\nimport tornado.web\nimport tornado.httpserver\nimport tornado.log\nimport asyncio\nfrom tornado import concurrent\nfrom .utils import utils\n\nfrom kfserving.handlers.http import PredictHandler, ExplainHandler\nfrom kfserving import KFModel\nfrom kfserving.kfmodel_repository import KFModelRepository\n\nDEFAULT_HTTP_PORT = 8080\nDEFAULT_GRPC_PORT = 8081\nDEFAULT_MAX_BUFFER_SIZE = 104857600\n\nparser = argparse.ArgumentParser(add_help=False)\nparser.add_argument('--http_port', default=DEFAULT_HTTP_PORT, type=int,\n                    help='The HTTP Port listened to by the model server.')\nparser.add_argument('--grpc_port', default=DEFAULT_GRPC_PORT, type=int,\n                    help='The GRPC Port listened to by the model server.')\nparser.add_argument('--max_buffer_size', default=DEFAULT_MAX_BUFFER_SIZE, type=int,\n                    help='The max buffer size for tornado.')\nparser.add_argument('--workers', default=1, type=int,\n                    help='The number of works to fork')\nparser.add_argument('--max_asyncio_workers', default=None, type=int,\n                    help='Max number of asyncio workers to spawn')\nargs, _ = parser.parse_known_args()\n\ntornado.log.enable_pretty_logging()\n\n\nclass KFServer:\n    def __init__(self, http_port: int = args.http_port,\n                 grpc_port: int = args.grpc_port,\n                 max_buffer_size: int = args.max_buffer_size,\n                 workers: int = args.workers,\n                 max_asyncio_workers: int = args.max_asyncio_workers,\n                 registered_models: KFModelRepository = KFModelRepository()):\n        self.registered_models = registered_models\n        self.http_port = http_port\n        self.grpc_port = grpc_port\n        self.max_buffer_size = max_buffer_size\n        self.workers = workers\n        self.max_asyncio_workers = max_asyncio_workers\n        self._http_server: Optional[tornado.httpserver.HTTPServer] = None\n\n    def create_application(self):\n        return tornado.web.Application([\n            # Server Liveness API returns 200 if server is alive.\n            (r\"/\", LivenessHandler),\n            (r\"/v2/health/live\", LivenessHandler),\n            (r\"/v1/models\",\n             ListHandler, dict(models=self.registered_models)),\n            (r\"/v2/models\",\n             ListHandler, dict(models=self.registered_models)),\n            # Model Health API returns 200 if model is ready to serve.\n            (r\"/v1/models/([a-zA-Z0-9_-]+)\",\n             HealthHandler, dict(models=self.registered_models)),\n            (r\"/v2/models/([a-zA-Z0-9_-]+)/status\",\n             HealthHandler, dict(models=self.registered_models)),\n            (r\"/v1/models/([a-zA-Z0-9_-]+):predict\",\n             PredictHandler, dict(models=self.registered_models)),\n            (r\"/v2/models/([a-zA-Z0-9_-]+)/infer\",\n             PredictHandler, dict(models=self.registered_models)),\n            (r\"/v1/models/([a-zA-Z0-9_-]+):explain\",\n             ExplainHandler, dict(models=self.registered_models)),\n            (r\"/v2/models/([a-zA-Z0-9_-]+)/explain\",\n             ExplainHandler, dict(models=self.registered_models)),\n            (r\"/v2/repository/models/([a-zA-Z0-9_-]+)/load\",\n             LoadHandler, dict(models=self.registered_models)),\n            (r\"/v2/repository/models/([a-zA-Z0-9_-]+)/unload\",\n             UnloadHandler, dict(models=self.registered_models)),\n        ])\n\n    def start(self, models: List[KFModel], nest_asyncio: bool = False):\n        if self.max_asyncio_workers is None:\n            # formula as suggest in https://bugs.python.org/issue35279\n            self.max_asyncio_workers = min(32, utils.cpu_count()+4)\n\n        logging.info(f\"Setting asyncio max_workers as {self.max_asyncio_workers}\")\n        asyncio.get_event_loop().set_default_executor(\n            concurrent.futures.ThreadPoolExecutor(max_workers=self.max_asyncio_workers))\n\n        for model in models:\n            self.register_model(model)\n\n        self._http_server = tornado.httpserver.HTTPServer(\n            self.create_application(), max_buffer_size=self.max_buffer_size)\n\n        logging.info(\"Listening on port %s\", self.http_port)\n        self._http_server.bind(self.http_port)\n        logging.info(\"Will fork %d workers\", self.workers)\n        self._http_server.start(self.workers)\n\n        # Need to start the IOLoop after workers have been started\n        # https://github.com/tornadoweb/tornado/issues/2426\n        # The nest_asyncio package needs to be installed by the downstream module\n        if nest_asyncio:\n            import nest_asyncio\n            nest_asyncio.apply()\n\n        tornado.ioloop.IOLoop.current().start()\n\n    def register_model(self, model: KFModel):\n        if not model.name:\n            raise Exception(\n                \"Failed to register model, model.name must be provided.\")\n        self.registered_models.update(model)\n        logging.info(\"Registering model: %s\", model.name)\n\n\nclass LivenessHandler(tornado.web.RequestHandler):  # pylint:disable=too-few-public-methods\n    def get(self):\n        self.write(\"Alive\")\n\n\nclass HealthHandler(tornado.web.RequestHandler):\n    def initialize(self, models: KFModelRepository):\n        self.models = models  # pylint:disable=attribute-defined-outside-init\n\n    def get(self, name: str):\n        model = self.models.get_model(name)\n        if model is None:\n            raise tornado.web.HTTPError(\n                status_code=404,\n                reason=\"Model with name %s does not exist.\" % name\n            )\n\n        if not model.ready:\n            raise tornado.web.HTTPError(\n                status_code=503,\n                reason=\"Model with name %s is not ready.\" % name\n            )\n\n        self.write(json.dumps({\n            \"name\": model.name,\n            \"ready\": model.ready\n        }))\n\n\nclass ListHandler(tornado.web.RequestHandler):\n    def initialize(self, models: KFModelRepository):\n        self.models = models  # pylint:disable=attribute-defined-outside-init\n\n    def get(self):\n        self.write(json.dumps([ob.name for ob in self.models.get_models()]))\n\n\nclass LoadHandler(tornado.web.RequestHandler):\n    def initialize(self, models: KFModelRepository):  # pylint:disable=attribute-defined-outside-init\n        self.models = models\n\n    async def post(self, name: str):\n        try:\n            (await self.models.load(name)) if inspect.iscoroutinefunction(self.models.load) else self.models.load(name)\n        except Exception:\n            ex_type, ex_value, ex_traceback = sys.exc_info()\n            raise tornado.web.HTTPError(\n                status_code=500,\n                reason=f\"Model with name {name} is not ready. \"\n                       f\"Error type: {ex_type} error msg: {ex_value}\"\n            )\n\n        if not self.models.is_model_ready(name):\n            raise tornado.web.HTTPError(\n                status_code=503,\n                reason=f\"Model with name {name} is not ready.\"\n            )\n        self.write(json.dumps({\n            \"name\": name,\n            \"load\": True\n        }))\n\n\nclass UnloadHandler(tornado.web.RequestHandler):\n    def initialize(self, models: KFModelRepository):  # pylint:disable=attribute-defined-outside-init\n        self.models = models\n\n    def post(self, name: str):\n        try:\n            self.models.unload(name)\n        except KeyError:\n            raise tornado.web.HTTPError(\n                status_code=404,\n                reason=\"Model with name %s does not exist.\" % name\n            )\n        self.write(json.dumps({\n            \"name\": name,\n            \"unload\": True\n        }))\n", "path": "python/kfserving/kfserving/kfserver.py"}]}