Use asyncio in ModelWrapper and adapt API

After moving to aiohttp it is easy to move the calls to de underlying
method to coroutines using asyncio. In order to do so we convert the
ModelWrapper to use async methods, and then we adapt the API to these
changes. As a side effect, we change the way we are defining the routes
and views, since we need to include a configuration option defining the
number of worker instances that we want to spawn.

We also implement a new warm method that can be defined by the user, so
that this method is called before the API is spawned.

Sem-Ver: api-break

Author: alvarolopez

deepaas/api/__init__.py

@@ -32,7 +32,7 @@
 CONF = cfg.CONF
 
 
-def get_app(doc="/docs"):
+async def get_app(doc="/docs"):
     """Get the main app."""
     global APP
 
@@ -66,6 +66,10 @@ def get_app(doc="/docs"):
 
     LOG.info("Serving loaded V2 models: %s", list(model.V2_MODELS.keys()))
 
+    for _, m in model.V2_MODELS.items():
+        LOG.debug("Warming models...")
+        await m.warm()
+
     if doc:
         # init docs with all parameters, usual for ApiSpec
         aiohttp_apispec.setup_aiohttp_apispec(

deepaas/api/v2/__init__.py

@@ -24,14 +24,10 @@
 from deepaas.api.v2 import predict as v2_predict
 from deepaas.api.v2 import responses
 from deepaas.api.v2 import train as v2_train
-from deepaas import model
 
 CONF = cfg.CONF
 LOG = log.getLogger("deepaas.api.v2")
 
-# Get the models (this is a singleton, so it is safe to call it multiple times
-model.register_v2_models()
-
 APP = None
 
 
@@ -51,10 +47,10 @@ def get_app():
     v2_debug.setup_debug()
 
     APP.router.add_get('/', get_version, name="v2")
-    APP.add_routes(v2_debug.routes)
-    APP.add_routes(v2_model.routes)
-    APP.add_routes(v2_train.routes)
-    APP.add_routes(v2_predict.routes)
+    v2_debug.setup_routes(APP)
+    v2_model.setup_routes(APP)
+    v2_train.setup_routes(APP)
+    v2_predict.setup_routes(APP)
 
     return APP
 

deepaas/api/v2/debug.py

@@ -25,13 +25,8 @@
 from oslo_log import log
 import six
 
-from deepaas import model
-
 CONF = cfg.CONF
 
-# Get the models (this is a singleton, so it is safe to call it multiple times
-model.register_v2_models()
-
 app = web.Application()
 routes = web.RouteTableDef()
 
@@ -85,10 +80,13 @@ def setup_debug():
     summary="""Return debug information if enabled by API.""",
     description="""Return debug information if enabled by API.""",
 )
-@routes.get('/debug')
-async def get(self):
+async def get(request):
     print("--- DEBUG MARKER %s ---" % datetime.datetime.now())
     resp = ""
     if DEBUG_STREAM is not None:
         resp = DEBUG_STREAM.getvalue()
     return web.Response(text=resp)
+
+
+def setup_routes(app):
+    app.router.add_get("/debug", get)

deepaas/api/v2/models.py

@@ -20,12 +20,6 @@
 from deepaas.api.v2 import responses
 from deepaas import model
 
-# Get the models (this is a singleton, so it is safe to call it multiple times
-model.register_v2_models()
-
-app = web.Application()
-routes = web.RouteTableDef()
-
 
 @aiohttp_apispec.docs(
     tags=["models"],
@@ -35,9 +29,8 @@
                 "will return the loaded models, as long as their basic "
                 "metadata.",
 )
-@routes.get('/models')
 @aiohttp_apispec.response_schema(responses.ModelMeta(), 200)
-async def get(request):
+async def index(request):
     """Return loaded models and its information.
 
     DEEPaaS can load several models and server them on the same endpoint,
@@ -61,30 +54,40 @@ async def get(request):
     return web.json_response({"models": models})
 
 
-# In the next lines we iterate over the loaded models and create the different
-# resources for each model. This way we can also load the expected parameters
-# if needed (as in the training method).
-for model_name, model_obj in model.V2_MODELS.items():
-    @routes.view('/models/%s' % model_name)
-    class BaseModel(web.View):
-        model_name = model_name
-        model_obj = model_obj
+class Handler(object):
+    model_name = None
+    model_obj = None
+
+    def __init__(self, model_name, model_obj):
+        self.model_name = model_name
+        self.model_obj = model_obj
+
+    @aiohttp_apispec.docs(
+        tags=["models"],
+        summary="Return model's metadata",
+    )
+    @aiohttp_apispec.response_schema(responses.ModelMeta(), 200)
+    async def get(self, request):
+        m = {
+            "id": self.model_name,
+            "name": self.model_name,
+            "links": [{
+                "rel": "self",
+                "href": "%s" % request.path,
+            }]
+        }
+        meta = self.model_obj.get_metadata()
+        m.update(meta)
+
+        return web.json_response(m)
+
 
-        @aiohttp_apispec.docs(
-            tags=["models"],
-            summary="Return '%s' model metadata" % model_name,
-        )
-        @aiohttp_apispec.response_schema(responses.ModelMeta(), 200)
-        async def get(self):
-            m = {
-                "id": self.model_name,
-                "name": self.model_name,
-                "links": [{
-                    "rel": "self",
-                    "href": "%s" % self.request.path,
-                }]
-            }
-            meta = self.model_obj.get_metadata()
-            m.update(meta)
+def setup_routes(app):
+    app.router.add_get("/models", index)
 
-            return web.json_response(m)
+    # In the next lines we iterate over the loaded models and create the
+    # different resources for each model. This way we can also load the
+    # expected parameters if needed (as in the training method).
+    for model_name, model_obj in model.V2_MODELS.items():
+        hdlr = Handler(model_name, model_obj)
+        app.router.add_get("/models/%s" % model_name, hdlr.get)

deepaas/api/v2/predict.py

@@ -22,12 +22,6 @@
 from deepaas.api.v2 import responses
 from deepaas import model
 
-# Get the models (this is a singleton, so it is safe to call it multiple times
-model.register_v2_models()
-
-app = web.Application()
-routes = web.RouteTableDef()
-
 
 def _get_model_response(model_name, model_obj):
     response_schema = model_obj.response_schema
@@ -38,33 +32,38 @@ def _get_model_response(model_name, model_obj):
     return responses.Prediction
 
 
-# In the next lines we iterate over the loaded models and create different
-# views for the different models. We take the corresponding arguments and
-# responses from the underlying model, adding them to the OpenAPI spec and
-# documentation.
-for model_name, model_obj in model.V2_MODELS.items():
-
-    args = webargs.core.dict2schema(model_obj.get_predict_args())
-    response = _get_model_response(model_name, model_obj)
-
-    @routes.view('/models/%s/predict' % model_name)
-    class ModelPredict(web.View):
-        model_name = model_name
-        model_obj = model_obj
-
-        @aiohttp_apispec.docs(
-            tags=["models"],
-            summary="Make a prediction given the input data"
-        )
-        @aiohttp_apispec.querystring_schema(args)
-        @aiohttp_apispec.response_schema(response(), 200)
-        @aiohttp_apispec.response_schema(responses.Failure(), 400)
-        @aiohttpparser.parser.use_args(args)
-        async def post(self, args):
-            ret = self.model_obj.predict(**args)
-
-            if self.model_obj.has_schema:
-                self.model_obj.validate_response(ret)
-                return web.json_response(ret)
-
-            return web.json_response({"status": "OK", "predictions": ret})
+def setup_routes(app):
+    # In the next lines we iterate over the loaded models and create the
+    # different resources for each model. This way we can also load the
+    # expected parameters if needed (as in the training method).
+    for model_name, model_obj in model.V2_MODELS.items():
+        args = webargs.core.dict2schema(model_obj.get_predict_args())
+        response = _get_model_response(model_name, model_obj)
+
+        class Handler(object):
+            model_name = None
+            model_obj = None
+
+            def __init__(self, model_name, model_obj):
+                self.model_name = model_name
+                self.model_obj = model_obj
+
+            @aiohttp_apispec.docs(
+                tags=["models"],
+                summary="Make a prediction given the input data"
+            )
+            @aiohttp_apispec.querystring_schema(args)
+            @aiohttp_apispec.response_schema(response(), 200)
+            @aiohttp_apispec.response_schema(responses.Failure(), 400)
+            @aiohttpparser.parser.use_args(args)
+            async def post(self, request, args):
+                ret = await self.model_obj.predict(**args)
+
+                if self.model_obj.has_schema:
+                    self.model_obj.validate_response(ret)
+                    return web.json_response(ret)
+
+                return web.json_response({"status": "OK", "predictions": ret})
+
+        hdlr = Handler(model_name, model_obj)
+        app.router.add_post("/models/%s/predict" % model_name, hdlr.post)

deepaas/api/v2/train.py

@@ -21,33 +21,33 @@
 
 from deepaas import model
 
-# Get the models (this is a singleton, so it is safe to call it multiple times
-model.register_v2_models()
-
-app = web.Application()
-routes = web.RouteTableDef()
-
-
-# In the next lines we iterate over the loaded models and create different
-# views for the different models. We take the corresponding arguments and
-# responses from the underlying model, adding them to the OpenAPI spec and
-# documentation.
-for model_name, model_obj in model.V2_MODELS.items():
-    args = webargs.core.dict2schema(model_obj.get_train_args())
-
-    @routes.view('/models/%s/train' % model_name)
-    class ModelTrain(web.View):
-        model_name = model_name
-        model_obj = model_obj
-
-        @aiohttp_apispec.docs(
-            tags=["models"],
-            summary="Retrain model with available data"
-        )
-        @aiohttp_apispec.querystring_schema(args)
-        @aiohttpparser.parser.use_args(args)
-        async def put(self, args):
-            ret = self.model_obj.train(**args)
-            # FIXME(aloga): what are we returning here? We need to take care
-            # of these responses as well.
-            return web.json_response(ret)
+
+def setup_routes(app):
+    # In the next lines we iterate over the loaded models and create the
+    # different resources for each model. This way we can also load the
+    # expected parameters if needed (as in the training method).
+    for model_name, model_obj in model.V2_MODELS.items():
+        args = webargs.core.dict2schema(model_obj.get_train_args())
+
+        class Handler(object):
+            model_name = None
+            model_obj = None
+
+            def __init__(self, model_name, model_obj):
+                self.model_name = model_name
+                self.model_obj = model_obj
+
+            @aiohttp_apispec.docs(
+                tags=["models"],
+                summary="Retrain model with available data"
+            )
+            @aiohttp_apispec.querystring_schema(args)
+            @aiohttpparser.parser.use_args(args)
+            async def post(self, request, args):
+                ret = await self.model_obj.train(**args)
+                # FIXME(aloga): what are we returning here? We need to take
+                # care of these responses as well.
+                return web.json_response(ret)
+
+        hdlr = Handler(model_name, model_obj)
+        app.router.add_post("/models/%s/train" % model_name, hdlr.post)

deepaas/config.py

@@ -41,6 +41,14 @@
 print to the standard output and error (i.e. stdout and stderr) through the
 "/debug" endpoint. Default is to not provide this information. This will not
 provide logging information about the API itself.
+"""),
+    cfg.IntOpt('model-workers',
+               short='n',
+               default=1,
+               help="""
+Specify the number of workers *per model* that we will initialize. If using a
+CPU you probably want to increase this number, if using a GPU probably you want
+to leave it to 1. (defaults to 1)
 """),
 ]
 

deepaas/model/v2/base.py

@@ -230,3 +230,15 @@ def get_train_args():
             application required arguments.
         """
         raise NotImplementedError()
+
+    @abc.abstractmethod
+    def warm(self):
+        """Warm (initialize, load) the model.
+
+        This is called when the model is loaded, before the API is spawned.
+
+        If implemented, it should prepare the model for execution. This is
+        useful for loading it into memory, perform any kind of preliminary
+        checks, etc.
+        """
+        raise NotImplementedError()

deepaas/model/v2/test.py

@@ -45,6 +45,9 @@ class TestModel(base.BaseModel):
         ),
     }
 
+    def warm(self):
+        LOG.debug("Test model is warming...")
+
     def predict(self, **kwargs):
         LOG.debug("Got the following kw arguments: %s", kwargs)
         return {