add single process app() (#212)

* add single_app()

Author: ShenYuhan

paddlehub/commands/serving.py

@@ -18,13 +18,12 @@
 from __future__ import print_function
 
 import argparse
-import subprocess
-import shlex
 import os
+import platform
+import socket
 import json
 import paddlehub as hub
 from paddlehub.commands.base_command import BaseCommand, ENTRY
-from paddlehub.serving import app
 
 
 class ServingCommand(BaseCommand):
@@ -41,33 +40,56 @@ def __init__(self, name):
             usage='%(prog)s',
             add_help=True)
         self.parser.add_argument("command")
+        self.parser.add_argument("sub_command")
         self.sub_parse = self.parser.add_mutually_exclusive_group(
             required=False)
-        self.sub_parse.add_argument("--start", action="store_true")
         self.parser.add_argument(
             "--use_gpu", action="store_true", default=False)
+        self.parser.add_argument(
+            "--use_multiprocess", action="store_true", default=False)
         self.parser.add_argument("--modules", "-m", nargs="+")
         self.parser.add_argument("--config", "-c", nargs="+")
-        self.parser.add_argument("--port", "-p", nargs="+", default=[8888])
+        self.parser.add_argument("--port", "-p", nargs="+", default=[8866])
+
+    @staticmethod
+    def is_port_occupied(ip, port):
+        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        try:
+            s.connect((ip, int(port)))
+            s.shutdown(2)
+            return True
+        except:
+            return False
 
     @staticmethod
     def preinstall_modules(modules):
         configs = []
+        module_exist = {}
         if modules is not None:
             for module in modules:
                 module_name = module if "==" not in module else \
                 module.split("==")[0]
                 module_version = None if "==" not in module else \
                 module.split("==")[1]
+                if module_exist.get(module_name, "") != "":
+                    print(module_name, "==", module_exist.get(module_name),
+                          " will be ignored cause new version is specified.")
+                    configs.pop()
+                module_exist.update({module_name: module_version})
                 try:
                     m = hub.Module(name=module_name, version=module_version)
+                    method_name = m.desc.attr.map.data['default_signature'].s
+                    if method_name == "":
+                        raise RuntimeError("{} cannot be use for "
+                                           "predicting".format(module_name))
                     configs.append({
                         "module": module_name,
                         "version": m.version,
                         "category": str(m.type).split("/")[0].upper()
                     })
                 except Exception as err:
-                    pass
+                    print(err, ", start Hub-Serving unsuccessfully.")
+                    exit(1)
             return configs
 
     @staticmethod
@@ -78,8 +100,24 @@ def start_serving(args):
             if os.path.exists(config_file):
                 with open(config_file, "r") as fp:
                     configs = json.load(fp)
+                    use_multiprocess = configs.get("use_multiprocess", False)
+                    if use_multiprocess is True:
+                        if platform.system() == "Windows":
+                            print(
+                                "Warning: Windows cannot use multiprocess working "
+                                "mode, Hub-Serving will switch to single process mode"
+                            )
+                            from paddlehub.serving import app_single as app
+                        else:
+                            from paddlehub.serving import app
+                    else:
+                        from paddlehub.serving import app_single as app
                     use_gpu = configs.get("use_gpu", False)
-                    port = configs.get("port", 8888)
+                    port = configs.get("port", 8866)
+                    if ServingCommand.is_port_occupied("127.0.0.1",
+                                                       port) is True:
+                        print("Port %s is occupied, please change it." % (port))
+                        return False
                     configs = configs.get("modules_info")
                     module = [
                         str(i["module"]) + "==" + str(i["version"])
@@ -92,10 +130,23 @@ def start_serving(args):
             else:
                 print("config_file ", config_file, "not exists.")
         else:
+            if args.use_multiprocess is True:
+                if platform.system() == "Windows":
+                    print(
+                        "Warning: Windows cannot use multiprocess working "
+                        "mode, Hub-Serving will switch to single process mode")
+                    from paddlehub.serving import app_single as app
+                else:
+                    from paddlehub.serving import app
+            else:
+                from paddlehub.serving import app_single as app
             module = args.modules
             if module is not None:
                 use_gpu = args.use_gpu
                 port = args.port[0]
+                if ServingCommand.is_port_occupied("127.0.0.1", port) is True:
+                    print("Port %s is occupied, please change it." % (port))
+                    return False
                 module_info = ServingCommand.preinstall_modules(module)
                 [
                     item.update({
@@ -111,9 +162,10 @@ def start_serving(args):
     def show_help():
         str = "serving <option>\n"
         str += "\tManage PaddleHub-Serving.\n"
-        str += "option:\n"
-        str += "--start\n"
+        str += "sub command:\n"
+        str += "start\n"
         str += "\tStart PaddleHub-Serving if specifies this parameter.\n"
+        str += "option:\n"
         str += "--modules/-m [module1==version, module2==version...]\n"
         str += "\tPre-install modules via this parameter list.\n"
         str += "--port/-p XXXX\n"
@@ -126,8 +178,13 @@ def show_help():
         print(str)
 
     def execute(self, argv):
-        args = self.parser.parse_args()
-        if args.start is True:
+        try:
+            args = self.parser.parse_args()
+        except:
+            print("Please refer to the instructions below.")
+            ServingCommand.show_help()
+            return False
+        if args.sub_command == "start":
             ServingCommand.start_serving(args)
         else:
             ServingCommand.show_help()

paddlehub/module/module.py

@@ -155,18 +155,20 @@ def _init_with_name(self, name, version=None):
             module_name=name, module_version=version, extra=extra)
         if not result:
             logger.error(tips)
-            exit(1)
-        logger.info(tips)
-        self._init_with_module_file(module_dir[0])
+            raise RuntimeError(tips)
+        else:
+            logger.info(tips)
+            self._init_with_module_file(module_dir[0])
 
     def _init_with_url(self, url):
         utils.check_url(url)
         result, tips, module_dir = default_downloader.download_file_and_uncompress(
             url, save_path=".")
         if not result:
             logger.error(tips)
-            exit(1)
-        self._init_with_module_file(module_dir)
+            raise RuntimeError(tips)
+        else:
+            self._init_with_module_file(module_dir)
 
     def _dump_processor(self):
         import inspect

paddlehub/serving/app.py

@@ -185,24 +185,51 @@ def index():
 
     @app_instance.before_request
     def before_request():
-        request.data = {"id": str(time.time())}
+        request.data = {"id": utils.md5(request.remote_addr + str(time.time()))}
+        print(request.remote_addr)
         pass
 
+    @app_instance.route("/get/modules", methods=["GET", "POST"])
+    def get_modules_info():
+        global nlp_module, cv_module
+        module_info = {}
+        if len(nlp_module) > 0:
+            module_info.update({"nlp_module": [{"Choose...": "Choose..."}]})
+            for item in nlp_module:
+                module_info["nlp_module"].append({item: item})
+        if len(cv_module) > 0:
+            module_info.update({"cv_module": [{"Choose...": "Choose..."}]})
+            for item in cv_module:
+                module_info["cv_module"].append({item: item})
+        module_info.update({"Choose...": [{"请先选择分类": "Choose..."}]})
+        return {"module_info": module_info}
+
     @app_instance.route("/predict/image/<module_name>", methods=["POST"])
     def predict_iamge(module_name):
         global results_dict
         req_id = request.data.get("id")
-        img_base64 = request.form.get("input_img", "")
-        received_file_name = request.form.get("input_file", "")
-        ext = received_file_name.split(".")[-1]
-        if ext == "":
-            return {"result": "Unrecognized file type"}
+
+        img_base64 = request.form.get("image", "")
+        if img_base64 != "":
+            img_base64 = request.form.get("image", "")
+            ext = img_base64.split(";")[0].split("/")[-1]
+            if ext not in ["jpeg", "jpg", "png"]:
+                return {"result": "Unrecognized file type"}
+            filename = utils.md5(str(time.time()) + str(img_base64)) + "." + ext
+            base64_head = img_base64.split(',')[0]
+            img_data = base64.b64decode(img_base64.split(',')[-1])
+            with open(filename, "wb") as fp:
+                fp.write(img_data)
+        else:
+            file = request.files["image"]
+            filename = file.filename
+            ext = file.filename.split(".")[-1]
+            if ext not in ["jpeg", "jpg", "png"]:
+                return {"result": "Unrecognized file type"}
+            base64_head = "data:image/" + ext + ";base64"
+            filename = utils.md5(filename) + '.' + ext
+            file.save(filename)
         score = time.time()
-        filename = utils.md5(str(time.time()) + str(img_base64)) + "." + ext
-        base64_head = img_base64.split(',')[0]
-        img_data = base64.b64decode(img_base64.split(',')[-1])
-        with open(filename, "wb") as fp:
-            fp.write(img_data)
         file_list = [filename]
         if queues_dict[module_name].qsize(
         ) + 1 > queues_dict[module_name].get_attribute("maxsize"):
@@ -211,9 +238,14 @@ def predict_iamge(module_name):
         data_num = len(file_list)
         results = []
         result_len = 0
+        start_time = time.time()
         while result_len != data_num:
             result_len = len(results_dict.get(req_id, []))
+            if time.time() - start_time > time_out:
+                results_dict.pop(req_id, None)
+                return {"result": "Request time out."}
         results = results_dict.get(req_id)
+        results_dict.pop(req_id, None)
         results = [i[1] for i in sorted(results, key=lambda k: k[0])]
         filename = results[0].get("path")
         ext = filename.split(".")[-1]
@@ -225,7 +257,7 @@ def predict_iamge(module_name):
                 os.remove(filename)
                 os.remove(output_file)
                 results = {
-                    "border":
+                    "desc":
                     str(results[0]["data"]),
                     "output_img":
                     base64_head + "," + str(output_img_base64).replace(
@@ -244,7 +276,7 @@ def data_2_item(data_list, req_id, score, module_name):
     def predict_text(module_name):
         global results_dict, queues_dict
         req_id = request.data.get("id")
-        data_list = request.form.get("input_text")
+        data_list = request.form.get("text")
         score = time.time()
         data_list = data_list.splitlines()
         data_temp = []
@@ -261,14 +293,17 @@ def predict_text(module_name):
         if data_num + queues_dict[module_name].qsize(
         ) > queues_dict[module_name].get_attribute("maxsize"):
             return {"result": "Too many visitors now, please come back later."}
-        start = time.time()
-
         data_2_item(data_list, req_id, score, module_name)
         results = []
         result_len = 0
+        start_time = time.time()
         while result_len != data_num:
             result_len = len(results_dict.get(req_id, []))
+            if time.time() - start_time > time_out:
+                results_dict.pop(req_id, None)
+                return {"result": "Request time out."}
         results = results_dict.get(req_id)
+        results_dict.pop(req_id, None)
         results = [i[1] for i in sorted(results, key=lambda k: k[0])]
         return {"result": results}
 
@@ -302,8 +337,9 @@ def config_with_file(configs):
         queue_name_list.append(item["module"])
 
 
-def run(is_use_gpu=False, configs=None, port=8888):
-    global use_gpu
+def run(is_use_gpu=False, configs=None, port=8866, timeout=60):
+    global use_gpu, time_out
+    time_out = timeout
     use_gpu = is_use_gpu
     if configs is not None:
         config_with_file(configs)