__init__.py 11.3 KB
Newer Older
G
guru4elephant 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14
#   Copyright (c) 2020 PaddlePaddle Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

G
guru4elephant 已提交
15
import os
16 17 18
from .proto import server_configure_pb2 as server_sdk
from .proto import general_model_config_pb2 as m_config
import google.protobuf.text_format
M
MRXLT 已提交
19
import tarfile
M
MRXLT 已提交
20
import socket
M
MRXLT 已提交
21 22
import paddle_serving_server as paddle_serving_server
from version import serving_server_version
M
MRXLT 已提交
23
from contextlib import closing
M
MRXLT 已提交
24

G
guru4elephant 已提交
25 26 27

class OpMaker(object):
    def __init__(self):
28
        self.op_dict = {
M
MRXLT 已提交
29 30 31 32 33 34
            "general_infer": "GeneralInferOp",
            "general_reader": "GeneralReaderOp",
            "general_response": "GeneralResponseOp",
            "general_text_reader": "GeneralTextReaderOp",
            "general_text_response": "GeneralTextResponseOp",
            "general_single_kv": "GeneralSingleKVOp",
G
guru4elephant 已提交
35 36
            "general_dist_kv": "GeneralDistKVOp",
            "general_copy": "GeneralCopyOp"
37
        }
G
guru4elephant 已提交
38

G
guru4elephant 已提交
39 40
    # currently, inputs and outputs are not used
    # when we have OpGraphMaker, inputs and outputs are necessary
G
guru4elephant 已提交
41 42
    def create(self, name, inputs=[], outputs=[]):
        if name not in self.op_dict:
M
MRXLT 已提交
43 44
            raise Exception("Op name {} is not supported right now".format(
                name))
G
guru4elephant 已提交
45 46 47 48 49
        node = server_sdk.DAGNode()
        node.name = "{}_op".format(name)
        node.type = self.op_dict[name]
        return node

M
MRXLT 已提交
50

G
guru4elephant 已提交
51 52 53 54 55 56 57
class OpSeqMaker(object):
    def __init__(self):
        self.workflow = server_sdk.Workflow()
        self.workflow.name = "workflow1"
        self.workflow.workflow_type = "Sequence"

    def add_op(self, node):
G
guru4elephant 已提交
58 59 60 61 62
        if len(self.workflow.nodes) >= 1:
            dep = server_sdk.DAGNodeDependency()
            dep.name = self.workflow.nodes[-1].name
            dep.mode = "RO"
            node.dependencies.extend([dep])
G
guru4elephant 已提交
63 64 65 66 67 68 69
        self.workflow.nodes.extend([node])

    def get_op_sequence(self):
        workflow_conf = server_sdk.WorkflowConf()
        workflow_conf.workflows.extend([self.workflow])
        return workflow_conf

M
MRXLT 已提交
70

G
guru4elephant 已提交
71 72 73 74 75
class Server(object):
    def __init__(self):
        self.server_handle_ = None
        self.infer_service_conf = None
        self.model_toolkit_conf = None
G
guru4elephant 已提交
76
        self.resource_conf = None
G
guru4elephant 已提交
77
        self.engine = None
78 79
        self.memory_optimization = False
        self.model_conf = None
G
guru4elephant 已提交
80 81 82 83
        self.workflow_fn = "workflow.prototxt"
        self.resource_fn = "resource.prototxt"
        self.infer_service_fn = "infer_service.prototxt"
        self.model_toolkit_fn = "model_toolkit.prototxt"
84
        self.general_model_config_fn = "general_model.prototxt"
G
guru4elephant 已提交
85 86
        self.workdir = ""
        self.max_concurrency = 0
M
MRXLT 已提交
87
        self.num_threads = 4
G
guru4elephant 已提交
88 89
        self.port = 8080
        self.reload_interval_s = 10
M
MRXLT 已提交
90 91
        self.module_path = os.path.dirname(paddle_serving_server.__file__)
        self.cur_path = os.getcwd()
M
MRXLT 已提交
92
        self.use_local_bin = False
M
MRXLT 已提交
93
        self.mkl_flag = False
G
guru4elephant 已提交
94 95 96 97 98 99 100 101 102 103 104 105

    def set_max_concurrency(self, concurrency):
        self.max_concurrency = concurrency

    def set_num_threads(self, threads):
        self.num_threads = threads

    def set_port(self, port):
        self.port = port

    def set_reload_interval(self, interval):
        self.reload_interval_s = interval
G
guru4elephant 已提交
106 107 108 109

    def set_op_sequence(self, op_seq):
        self.workflow_conf = op_seq

110 111 112
    def set_memory_optimize(self, flag=False):
        self.memory_optimization = flag

M
MRXLT 已提交
113 114 115 116
    def check_local_bin(self):
        if "SERVING_BIN" in os.environ:
            self.use_local_bin = True
            self.bin_path = os.environ["SERVING_BIN"]
M
MRXLT 已提交
117

G
guru4elephant 已提交
118 119 120
    def _prepare_engine(self, model_config_path, device):
        if self.model_toolkit_conf == None:
            self.model_toolkit_conf = server_sdk.ModelToolkitConf()
121

G
guru4elephant 已提交
122 123
        if self.engine == None:
            self.engine = server_sdk.EngineDesc()
124 125

        self.model_config_path = model_config_path
G
guru4elephant 已提交
126
        self.engine.name = "general_model"
G
guru4elephant 已提交
127
        self.engine.reloadable_meta = model_config_path + "/fluid_time_file"
128
        os.system("touch {}".format(self.engine.reloadable_meta))
G
guru4elephant 已提交
129
        self.engine.reloadable_type = "timestamp_ne"
G
guru4elephant 已提交
130 131 132
        self.engine.runtime_thread_num = 0
        self.engine.batch_infer_size = 0
        self.engine.enable_batch_align = 0
G
guru4elephant 已提交
133
        self.engine.model_data_path = model_config_path
134
        self.engine.enable_memory_optimization = self.memory_optimization
G
guru4elephant 已提交
135 136
        self.engine.static_optimization = False
        self.engine.force_update_static_cache = False
137

G
guru4elephant 已提交
138 139 140 141
        if device == "cpu":
            self.engine.type = "FLUID_CPU_ANALYSIS_DIR"
        elif device == "gpu":
            self.engine.type = "FLUID_GPU_ANALYSIS_DIR"
142

G
guru4elephant 已提交
143
        self.model_toolkit_conf.engines.extend([self.engine])
G
guru4elephant 已提交
144 145 146 147 148 149 150 151 152 153 154 155

    def _prepare_infer_service(self, port):
        if self.infer_service_conf == None:
            self.infer_service_conf = server_sdk.InferServiceConf()
            self.infer_service_conf.port = port
            infer_service = server_sdk.InferService()
            infer_service.name = "GeneralModelService"
            infer_service.workflows.extend(["workflow1"])
            self.infer_service_conf.services.extend([infer_service])

    def _prepare_resource(self, workdir):
        if self.resource_conf == None:
M
MRXLT 已提交
156 157
            with open("{}/{}".format(workdir, self.general_model_config_fn),
                      "w") as fout:
158
                fout.write(str(self.model_conf))
G
guru4elephant 已提交
159 160
            self.resource_conf = server_sdk.ResourceConf()
            self.resource_conf.model_toolkit_path = workdir
G
guru4elephant 已提交
161
            self.resource_conf.model_toolkit_file = self.model_toolkit_fn
162 163
            self.resource_conf.general_model_path = workdir
            self.resource_conf.general_model_file = self.general_model_config_fn
G
guru4elephant 已提交
164 165 166 167 168 169

    def _write_pb_str(self, filepath, pb_obj):
        with open(filepath, "w") as fout:
            fout.write(str(pb_obj))

    def load_model_config(self, path):
G
guru4elephant 已提交
170
        self.model_config_path = path
171 172 173 174
        self.model_conf = m_config.GeneralModelConfig()
        f = open("{}/serving_server_conf.prototxt".format(path), 'r')
        self.model_conf = google.protobuf.text_format.Merge(
            str(f.read()), self.model_conf)
G
guru4elephant 已提交
175
        # check config here
176
        # print config here
G
guru4elephant 已提交
177

M
MRXLT 已提交
178 179 180
    def use_mkl(self):
        self.mkl_flag = True

M
MRXLT 已提交
181 182
    def get_device_version(self):
        avx_flag = False
M
MRXLT 已提交
183
        mkl_flag = self.mkl_flag
M
MRXLT 已提交
184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210
        openblas_flag = False
        r = os.system("cat /proc/cpuinfo | grep avx > /dev/null 2>&1")
        if r == 0:
            avx_flag = True
        if avx_flag:
            if mkl_flag:
                device_version = "serving-cpu-avx-mkl-"
            else:
                device_version = "serving-cpu-avx-openblas-"
        else:
            device_version = "serving-cpu-noavx-openblas-"
        return device_version

    def download_bin(self):
        os.chdir(self.module_path)
        need_download = False
        device_version = self.get_device_version()
        floder_name = device_version + serving_server_version
        tar_name = floder_name + ".tar.gz"
        bin_url = "https://paddle-serving.bj.bcebos.com/bin/" + tar_name
        self.server_path = os.path.join(self.module_path, floder_name)
        if not os.path.exists(self.server_path):
            print('Frist time run, downloading PaddleServing components ...')
            r = os.system('wget ' + bin_url + ' --no-check-certificate')
            if r != 0:
                if os.path.exists(tar_name):
                    os.remove(tar_name)
M
MRXLT 已提交
211 212 213
                raise SystemExit(
                    'Download failed, please check your network or permission of {}.'.
                    format(self.module_path))
M
MRXLT 已提交
214 215 216 217 218 219 220 221 222
            else:
                try:
                    print('Decompressing files ..')
                    tar = tarfile.open(tar_name)
                    tar.extractall()
                    tar.close()
                except:
                    if os.path.exists(exe_path):
                        os.remove(exe_path)
M
MRXLT 已提交
223 224 225
                    raise SystemExit(
                        'Decompressing failed, please check your permission of {} or disk space left.'.
                        foemat(self.module_path))
M
MRXLT 已提交
226 227 228 229 230
                finally:
                    os.remove(tar_name)
        os.chdir(self.cur_path)
        self.bin_path = self.server_path + "/serving"

G
guru4elephant 已提交
231
    def prepare_server(self, workdir=None, port=9292, device="cpu"):
G
guru4elephant 已提交
232 233 234
        if workdir == None:
            workdir = "./tmp"
            os.system("mkdir {}".format(workdir))
G
guru4elephant 已提交
235 236
        else:
            os.system("mkdir {}".format(workdir))
G
guru4elephant 已提交
237
        os.system("touch {}/fluid_time_file".format(workdir))
G
guru4elephant 已提交
238

M
MRXLT 已提交
239
        if not self.port_is_available(port):
M
MRXLT 已提交
240
            raise SystemExit("Prot {} is already used".format(port))
G
guru4elephant 已提交
241
        self._prepare_resource(workdir)
G
guru4elephant 已提交
242
        self._prepare_engine(self.model_config_path, device)
G
guru4elephant 已提交
243 244 245
        self._prepare_infer_service(port)
        self.workdir = workdir

G
guru4elephant 已提交
246 247 248 249
        infer_service_fn = "{}/{}".format(workdir, self.infer_service_fn)
        workflow_fn = "{}/{}".format(workdir, self.workflow_fn)
        resource_fn = "{}/{}".format(workdir, self.resource_fn)
        model_toolkit_fn = "{}/{}".format(workdir, self.model_toolkit_fn)
G
guru4elephant 已提交
250 251 252 253 254 255

        self._write_pb_str(infer_service_fn, self.infer_service_conf)
        self._write_pb_str(workflow_fn, self.workflow_conf)
        self._write_pb_str(resource_fn, self.resource_conf)
        self._write_pb_str(model_toolkit_fn, self.model_toolkit_conf)

M
MRXLT 已提交
256
    def port_is_available(self, port):
M
MRXLT 已提交
257 258
        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
            sock.settimeout(2)
259
            result = sock.connect_ex(('0.0.0.0', port))
M
MRXLT 已提交
260 261 262 263 264
        if result != 0:
            return True
        else:
            return False

G
guru4elephant 已提交
265 266
    def run_server(self):
        # just run server with system command
G
guru4elephant 已提交
267
        # currently we do not load cube
M
MRXLT 已提交
268
        self.check_local_bin()
M
MRXLT 已提交
269 270
        if not self.use_local_bin:
            self.download_bin()
G
guru4elephant 已提交
271
        else:
M
MRXLT 已提交
272
            print("Use local bin : {}".format(self.bin_path))
M
MRXLT 已提交
273 274
        command = "{} " \
                  "-enable_model_toolkit " \
G
guru4elephant 已提交
275 276 277 278 279 280 281 282 283
                  "-inferservice_path {} " \
                  "-inferservice_file {} " \
                  "-max_concurrency {} " \
                  "-num_threads {} " \
                  "-port {} " \
                  "-reload_interval_s {} " \
                  "-resource_path {} " \
                  "-resource_file {} " \
                  "-workflow_path {} " \
M
MRXLT 已提交
284
                  "-workflow_file {} " \
M
MRXLT 已提交
285
                  "-bthread_concurrency {} ".format(
M
MRXLT 已提交
286
                      self.bin_path,
G
guru4elephant 已提交
287 288 289 290 291 292 293 294
                      self.workdir,
                      self.infer_service_fn,
                      self.max_concurrency,
                      self.num_threads,
                      self.port,
                      self.reload_interval_s,
                      self.workdir,
                      self.resource_fn,
M
MRXLT 已提交
295 296
                      self.workdir,
                      self.workflow_fn,
G
guru4elephant 已提交
297 298 299
                      self.num_threads)
        print("Going to Run Comand")
        print(command)
G
guru4elephant 已提交
300
        os.system(command)