local_service_handler.py 12.9 KB
Newer Older
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.

B
barriery 已提交
15
import os
16
import logging
B
barriery 已提交
17
import multiprocessing
Z
zhangjun 已提交
18 19
#from paddle_serving_server import OpMaker, OpSeqMaker
#from paddle_serving_server import Server as GpuServer
20
#from paddle_serving_server import Server as CpuServer
B
barriery 已提交
21
from . import util
22
#from paddle_serving_app.local_predict import LocalPredictor
23 24

_LOGGER = logging.getLogger(__name__)
B
barriery 已提交
25
_workdir_name_gen = util.NameGenerator("workdir_")
26 27


W
fix bug  
wangjiawei04 已提交
28
class LocalServiceHandler(object):
29 30 31 32 33 34 35
    """
    LocalServiceHandler is the processor of the local service, contains
    three client types, brpc, grpc and local_predictor.If you use the 
    brpc or grpc, serveing startup ability is provided.If you use
    local_predictor, local predict ability is provided by paddle_serving_app.
    """

36
    def __init__(self,
B
barriery 已提交
37
                 model_config,
W
wangjiawei04 已提交
38
                 client_type='local_predictor',
B
barriery 已提交
39
                 workdir="",
40
                 thread_num=2,
41
                 device_type=-1,
42
                 devices="",
43
                 fetch_names=None,
44 45
                 mem_optim=True,
                 ir_optim=False,
46
                 available_port_generator=None,
47
                 use_profile=False,
T
TeslaZhao 已提交
48 49 50 51 52
                 precision="fp32",
                 use_mkldnn=False,
                 mkldnn_cache_capacity=0,
                 mkldnn_op_list=None,
                 mkldnn_bf16_op_list=None):
53 54 55 56 57 58 59 60
        """
        Initialization of localservicehandler

        Args:
           model_config: model config path
           client_type: brpc, grpc and local_predictor[default]
           workdir: work directory
           thread_num: number of threads, concurrent quantity.
61 62
           device_type: support multiple devices. -1=Not set, determined by
               `devices`. 0=cpu, 1=gpu, 2=tensorRT, 3=arm cpu, 4=kunlun xpu
63 64 65 66 67 68 69
           devices: gpu id list[gpu], "" default[cpu]
           fetch_names: get fetch names out of LocalServiceHandler in 
               local_predictor mode. fetch_names_ is compatible for Client().
           mem_optim: use memory/graphics memory optimization, True default.
           ir_optim: use calculation chart optimization, False default.
           available_port_generator: generate available ports
           use_profile: use profiling, False default.
70
           precision: inference precesion, e.g. "fp32", "fp16", "int8"
T
TeslaZhao 已提交
71 72 73 74
           use_mkldnn: use mkldnn, default False.
           mkldnn_cache_capacity: cache capacity of mkldnn, 0 means no limit.
           mkldnn_op_list: OP list optimized by mkldnn, None default.
           mkldnn_bf16_op_list: OP list optimized by mkldnn bf16, None default.
75 76 77 78

        Returns:
           None
        """
79
        if available_port_generator is None:
B
barriery 已提交
80
            available_port_generator = util.GetAvailablePortGenerator()
81

B
barriery 已提交
82
        self._model_config = model_config
83
        self._port_list = []
84 85 86 87 88
        self._device_name = "cpu"
        self._use_gpu = False
        self._use_trt = False
        self._use_lite = False
        self._use_xpu = False
T
TeslaZhao 已提交
89 90 91 92
        self._use_mkldnn = False
        self._mkldnn_cache_capacity = 0
        self._mkldnn_op_list = None
        self._mkldnn_bf16_op_list = None
93 94 95 96 97 98 99

        if device_type == -1:
            # device_type is not set, determined by `devices`, 
            if devices == "":
                # CPU
                self._device_name = "cpu"
                devices = [-1]
Z
zhangjun 已提交
100
            else:
101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118
                # GPU
                self._device_name = "gpu"
                self._use_gpu = True
                devices = [int(x) for x in devices.split(",")]

        elif device_type == 0:
            # CPU
            self._device_name = "cpu"
            devices = [-1]
        elif device_type == 1:
            # GPU
            self._device_name = "gpu"
            self._use_gpu = True
            devices = [int(x) for x in devices.split(",")]
        elif device_type == 2:
            # Nvidia Tensor RT
            self._device_name = "gpu"
            self._use_gpu = True
119
            devices = [int(x) for x in devices.split(",")]
120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137
            self._use_trt = True
        elif device_type == 3:
            # ARM CPU
            self._device_name = "arm"
            devices = [-1]
            self._use_lite = True
        elif device_type == 4:
            # Kunlun XPU
            self._device_name = "arm"
            devices = [int(x) for x in devices.split(",")]
            self._use_lite = True
            self._use_xpu = True
        else:
            _LOGGER.error(
                "LocalServiceHandler initialization fail. device_type={}"
                .format(device_type))

        if client_type == "brpc" or client_type == "grpc":
138 139
            for _ in devices:
                self._port_list.append(available_port_generator.next())
140 141 142
            _LOGGER.info("Create ports for devices:{}. Port:{}"
                         .format(devices, self._port_list))

143
        self._client_type = client_type
144 145 146 147 148
        self._workdir = workdir
        self._devices = devices
        self._thread_num = thread_num
        self._mem_optim = mem_optim
        self._ir_optim = ir_optim
149
        self._local_predictor_client = None
150 151
        self._rpc_service_list = []
        self._server_pros = []
152
        self._use_profile = use_profile
153
        self._fetch_names = fetch_names
154
        self._precision = precision
T
TeslaZhao 已提交
155 156 157 158
        self._use_mkldnn = use_mkldnn
        self._mkldnn_cache_capacity = mkldnn_cache_capacity
        self._mkldnn_op_list = mkldnn_op_list
        self._mkldnn_bf16_op_list = mkldnn_bf16_op_list
159 160 161 162 163

        _LOGGER.info(
            "Models({}) will be launched by device {}. use_gpu:{}, "
            "use_trt:{}, use_lite:{}, use_xpu:{}, device_type:{}, devices:{}, "
            "mem_optim:{}, ir_optim:{}, use_profile:{}, thread_num:{}, "
T
TeslaZhao 已提交
164 165 166
            "client_type:{}, fetch_names:{}, precision:{}, use_mkldnn:{}, "
            "mkldnn_cache_capacity:{}, mkldnn_op_list:{}, "
            "mkldnn_bf16_op_list:{}".format(
167
                model_config, self._device_name, self._use_gpu, self._use_trt,
T
TeslaZhao 已提交
168 169 170 171 172
                self._use_lite, self._use_xpu, device_type, self._devices,
                self._mem_optim, self._ir_optim, self._use_profile,
                self._thread_num, self._client_type, self._fetch_names,
                self._precision, self._use_mkldnn, self._mkldnn_cache_capacity,
                self._mkldnn_op_list, self._mkldnn_bf16_op_list))
173 174

    def get_fetch_list(self):
175
        return self._fetch_names
176 177 178 179

    def get_port_list(self):
        return self._port_list

180
    def get_client(self, concurrency_idx):
181 182 183
        """
        Function get_client is only used for local predictor case, creates one
        LocalPredictor object, and initializes the paddle predictor by function
184
        load_model_config.The concurrency_idx is used to select running devices.  
185 186

        Args:
187
            concurrency_idx: process/thread index
188 189 190 191

        Returns:
            _local_predictor_client
        """
192 193 194 195 196 197 198 199

        #checking the legality of concurrency_idx.
        device_num = len(self._devices)
        if device_num <= 0:
            _LOGGER.error("device_num must be not greater than 0. devices({})".
                          format(self._devices))
            raise ValueError("The number of self._devices error")

T
TeslaZhao 已提交
200
        if concurrency_idx < 0:
201 202 203 204 205 206
            _LOGGER.error("concurrency_idx({}) must be one positive number".
                          format(concurrency_idx))
            concurrency_idx = 0
        elif concurrency_idx >= device_num:
            concurrency_idx = concurrency_idx % device_num

T
TeslaZhao 已提交
207 208
        _LOGGER.info("GET_CLIENT : concurrency_idx={}, device_num={}".format(
            concurrency_idx, device_num))
209 210 211
        from paddle_serving_app.local_predict import LocalPredictor
        if self._local_predictor_client is None:
            self._local_predictor_client = LocalPredictor()
T
TeslaZhao 已提交
212
            # load model config and init predictor
213 214
            self._local_predictor_client.load_model_config(
                model_path=self._model_config,
215
                use_gpu=self._use_gpu,
216
                gpu_id=self._devices[concurrency_idx],
217 218 219 220
                use_profile=self._use_profile,
                thread_num=self._thread_num,
                mem_optim=self._mem_optim,
                ir_optim=self._ir_optim,
Z
zhangjun 已提交
221
                use_trt=self._use_trt,
222
                use_lite=self._use_lite,
223
                use_xpu=self._use_xpu,
T
TeslaZhao 已提交
224 225 226 227 228
                precision=self._precision,
                use_mkldnn=self._use_mkldnn,
                mkldnn_cache_capacity=self._mkldnn_cache_capacity,
                mkldnn_op_list=self._mkldnn_op_list,
                mkldnn_bf16_op_list=self._mkldnn_bf16_op_list)
229
        return self._local_predictor_client
W
wangjiawei04 已提交
230

B
barriery 已提交
231 232
    def get_client_config(self):
        return os.path.join(self._model_config, "serving_server_conf.prototxt")
233 234

    def _prepare_one_server(self, workdir, port, gpuid, thread_num, mem_optim,
235
                            ir_optim, precision):
236
        """
237
        According to self._device_name, generating one Cpu/Gpu/Arm Server, and
238 239 240 241 242 243 244 245 246
        setting the model config amd startup params.

        Args:
            workdir: work directory
            port: network port
            gpuid: gpu id
            thread_num: thread num
            mem_optim: use memory/graphics memory optimization
            ir_optim: use calculation chart optimization
247
            precision: inference precison, e.g."fp32", "fp16", "int8"
248 249 250 251

        Returns:
            server: CpuServer/GpuServer
        """
252
        if self._device_name == "cpu":
253 254 255 256 257 258 259 260 261 262 263 264 265
            from paddle_serving_server import OpMaker, OpSeqMaker, Server
            op_maker = OpMaker()
            read_op = op_maker.create('general_reader')
            general_infer_op = op_maker.create('general_infer')
            general_response_op = op_maker.create('general_response')

            op_seq_maker = OpSeqMaker()
            op_seq_maker.add_op(read_op)
            op_seq_maker.add_op(general_infer_op)
            op_seq_maker.add_op(general_response_op)

            server = Server()
        else:
Z
zhangjun 已提交
266
            #gpu or arm
Z
zhangjun 已提交
267
            from paddle_serving_server import OpMaker, OpSeqMaker, Server
268 269 270 271 272 273 274 275 276 277 278 279 280
            op_maker = OpMaker()
            read_op = op_maker.create('general_reader')
            general_infer_op = op_maker.create('general_infer')
            general_response_op = op_maker.create('general_response')

            op_seq_maker = OpSeqMaker()
            op_seq_maker.add_op(read_op)
            op_seq_maker.add_op(general_infer_op)
            op_seq_maker.add_op(general_response_op)

            server = Server()
            if gpuid >= 0:
                server.set_gpuid(gpuid)
Z
zhangjun 已提交
281
            # TODO: support arm or arm + xpu later
Z
fix bug  
zhangjun 已提交
282
            server.set_device(self._device_name)
283

284 285 286 287
        server.set_op_sequence(op_seq_maker.get_op_sequence())
        server.set_num_threads(thread_num)
        server.set_memory_optimize(mem_optim)
        server.set_ir_optimize(ir_optim)
288
        server.set_precision(precision)
289 290

        server.load_model_config(self._model_config)
291
        server.prepare_server(
292 293 294
            workdir=workdir, port=port, device=self._device_name)
        if self._fetch_names is None:
            self._fetch_names = server.get_fetch_list()
295 296 297
        return server

    def _start_one_server(self, service_idx):
298 299 300 301 302 303 304 305 306
        """
        Start one server
     
        Args:
            service_idx: server index
 
        Returns:
            None
        """
307 308 309
        self._rpc_service_list[service_idx].run_server()

    def prepare_server(self):
310 311 312
        """
        Prepare all servers to be started, and append them into list. 
        """
313
        for i, device_id in enumerate(self._devices):
B
barriery 已提交
314
            if self._workdir != "":
315 316 317 318 319 320 321 322 323 324
                workdir = "{}_{}".format(self._workdir, i)
            else:
                workdir = _workdir_name_gen.next()
            self._rpc_service_list.append(
                self._prepare_one_server(
                    workdir,
                    self._port_list[i],
                    device_id,
                    thread_num=self._thread_num,
                    mem_optim=self._mem_optim,
325 326
                    ir_optim=self._ir_optim,
                    precision=self._precision))
327 328

    def start_server(self):
329 330 331
        """
        Start multiple processes and start one server in each process
        """
332
        for i, _ in enumerate(self._rpc_service_list):
B
barriery 已提交
333 334 335
            p = multiprocessing.Process(
                target=self._start_one_server, args=(i, ))
            p.daemon = True
336 337 338
            self._server_pros.append(p)
        for p in self._server_pros:
            p.start()