diff --git a/python/paddle_serving_server/pyserver.py b/python/paddle_serving_server/pyserver.py index 8dda9a68e560df67d717024ef5cb905f3893afa8..409debc4e8550407191a747bf28d1c7cab8763f7 100644 --- a/python/paddle_serving_server/pyserver.py +++ b/python/paddle_serving_server/pyserver.py @@ -14,11 +14,13 @@ # pylint: disable=doc-string-missing import threading import multiprocessing +import multiprocessing.queues import Queue import os import sys import paddle_serving_server from paddle_serving_client import MultiLangClient as Client +from paddle_serving_client import MultiLangPredictFuture from concurrent import futures import numpy as np import grpc @@ -87,6 +89,7 @@ class ChannelDataType(enum.Enum): CHANNEL_PBDATA = 0 CHANNEL_FUTURE = 1 CHANNEL_NPDATA = 2 + ERROR = 3 class ChannelData(object): @@ -108,35 +111,29 @@ class ChannelData(object): 4. ChannelData(ChannelDataType.CHANNEL_PBDATA.value, npdata, data_id) 5. ChannelData(ChannelDataType.CHANNEL_NPDATA.value, npdata, data_id) 6. ChannelData(ecode, error_info, data_id) + + Protobufs are not pickle-able: + https://stackoverflow.com/questions/55344376/how-to-import-protobuf-module ''' if ecode is not None: if data_id is None or error_info is None: raise ValueError("data_id and error_info cannot be None") - pbdata = channel_pb2.ChannelData() - pbdata.ecode = ecode - pbdata.id = data_id - pbdata.error_info = error_info + datatype = ChannelDataType.ERROR.value else: if datatype == ChannelDataType.CHANNEL_FUTURE.value: - if pbdata is None: - if data_id is None: - raise ValueError("data_id cannot be None") - pbdata = channel_pb2.ChannelData() - pbdata.ecode = ChannelDataEcode.OK.value - pbdata.id = data_id + if data_id is None: + raise ValueError("data_id cannot be None") + ecode = ChannelDataEcode.OK.value elif datatype == ChannelDataType.CHANNEL_PBDATA.value: if pbdata is None: if data_id is None: raise ValueError("data_id cannot be None") pbdata = channel_pb2.ChannelData() - pbdata.id = data_id ecode, error_info = self._check_npdata(npdata) - pbdata.ecode = ecode - if pbdata.ecode != ChannelDataEcode.OK.value: - pbdata.error_info = error_info - logging.error(pbdata.error_info) + if ecode != ChannelDataEcode.OK.value: + logging.error(error_info) else: - for name, value in postped_data.items(): + for name, value in npdata.items(): inst = channel_pb2.Inst() inst.data = value.tobytes() inst.name = name @@ -146,23 +143,18 @@ class ChannelData(object): pbdata.insts.append(inst) elif datatype == ChannelDataType.CHANNEL_NPDATA.value: ecode, error_info = self._check_npdata(npdata) - pbdata = channel_pb2.ChannelData() - pbdata.id = data_id - pbdata.ecode = ecode - if pbdata.ecode != ChannelDataEcode.OK.value: - pbdata.error_info = error_info - logging.error(pbdata.error_info) + if ecode != ChannelDataEcode.OK.value: + logging.error(error_info) else: raise ValueError("datatype not match") - if not isinstance(pbdata, channel_pb2.ChannelData): - raise TypeError( - "pbdata must be pyserving_channel_pb2.ChannelData type({})". - format(type(pbdata))) self.future = future self.pbdata = pbdata self.npdata = npdata self.datatype = datatype self.callback_func = callback_func + self.id = data_id + self.ecode = ecode + self.error_info = error_info def _check_npdata(self, npdata): ecode = ChannelDataEcode.OK.value @@ -174,8 +166,8 @@ class ChannelData(object): "be str, but get {}".format(type(name))) break if not isinstance(value, np.ndarray): - pbdata.ecode = ChannelDataEcode.TYPE_ERROR.value - pbdata.error_info = log("the value of postped_data must " \ + ecode = ChannelDataEcode.TYPE_ERROR.value + error_info = log("the value of postped_data must " \ "be np.ndarray, but get {}".format(type(value))) break return ecode, error_info @@ -195,16 +187,15 @@ class ChannelData(object): elif self.datatype == ChannelDataType.CHANNEL_NPDATA.value: feed = self.npdata else: - raise TypeError("Error type({}) in pbdata.type.".format( - self.pbdata.type)) + raise TypeError("Error type({}) in datatype.".format(self.datatype)) return feed def __str__(self): - return "type[{}], ecode[{}]".format( - ChannelDataType(self.datatype).name, self.pbdata.ecode) + return "type[{}], ecode[{}], id[{}]".format( + ChannelDataType(self.datatype).name, self.ecode, self.id) -class Channel(Queue.Queue): +class Channel(multiprocessing.queues.Queue): """ The channel used for communication between Ops. @@ -222,23 +213,30 @@ class Channel(Queue.Queue): and can only be called during initialization. """ - def __init__(self, name=None, maxsize=-1, timeout=None): - Queue.Queue.__init__(self, maxsize=maxsize) + def __init__(self, manager, name=None, maxsize=0, timeout=None): + # https://stackoverflow.com/questions/39496554/cannot-subclass-multiprocessing-queue-in-python-3-5/ + multiprocessing.queues.Queue.__init__(self, maxsize=maxsize) self._maxsize = maxsize self._timeout = timeout self.name = name self._stop = False - self._cv = threading.Condition() + self._cv = multiprocessing.Condition() self._producers = [] - self._producer_res_count = {} # {data_id: count} - self._push_res = {} # {data_id: {op_name: data}} - - self._consumers = {} # {op_name: idx} - self._idx_consumer_num = {} # {idx: num} - self._consumer_base_idx = 0 - self._front_res = [] + self._producer_res_count = manager.dict() # {data_id: count} + # self._producer_res_count = {} # {data_id: count} + self._push_res = manager.dict() # {data_id: {op_name: data}} + # self._push_res = {} # {data_id: {op_name: data}} + + self._consumers = manager.dict() # {op_name: idx} + # self._consumers = {} # {op_name: idx} + self._idx_consumer_num = manager.dict() # {idx: num} + # self._idx_consumer_num = {} # {idx: num} + self._consumer_base_idx = manager.Value('i', 0) + # self._consumer_base_idx = 0 + self._front_res = manager.list() + # self._front_res = [] def get_producers(self): return self._producers @@ -286,9 +284,13 @@ class Channel(Queue.Queue): try: self.put(channeldata, timeout=0) break - except Queue.Empty: + except Queue.Full: self._cv.wait() + logging.debug( + self._log("{} channel size: {}".format(op_name, + self.qsize()))) self._cv.notify_all() + logging.debug(self._log("{} notify all".format(op_name))) logging.debug(self._log("{} push data succ!".format(op_name))) return True elif op_name is None: @@ -297,7 +299,7 @@ class Channel(Queue.Queue): "There are multiple producers, so op_name cannot be None.")) producer_num = len(self._producers) - data_id = channeldata.pbdata.id + data_id = channeldata.id put_data = None with self._cv: logging.debug(self._log("{} get lock".format(op_name))) @@ -307,7 +309,12 @@ class Channel(Queue.Queue): for name in self._producers } self._producer_res_count[data_id] = 0 - self._push_res[data_id][op_name] = channeldata + # see: https://docs.python.org/3.6/library/multiprocessing.html?highlight=multiprocess#proxy-objects + # self._push_res[data_id][op_name] = channeldata + tmp_push_res = self._push_res[data_id] + tmp_push_res[op_name] = channeldata + self._push_res[data_id] = tmp_push_res + if self._producer_res_count[data_id] + 1 == producer_num: put_data = self._push_res[data_id] self._push_res.pop(data_id) @@ -322,6 +329,9 @@ class Channel(Queue.Queue): else: while self._stop is False: try: + logging.debug( + self._log("{} push data succ: {}".format( + op_name, put_data.__str__()))) self.put(put_data, timeout=0) break except Queue.Empty: @@ -333,7 +343,7 @@ class Channel(Queue.Queue): return True def front(self, op_name=None): - logging.debug(self._log("{} try to get data".format(op_name))) + logging.debug(self._log("{} try to get data...".format(op_name))) if len(self._consumers) == 0: raise Exception( self._log( @@ -344,9 +354,18 @@ class Channel(Queue.Queue): with self._cv: while self._stop is False and resp is None: try: - resp = self.get(timeout=0) + logging.debug( + self._log("{} try to get(with channel size: {})". + format(op_name, self.qsize()))) + #TODO: bug to fix + # (multiple processes) the queue is not empty, but it raise Queue.Empty + resp = self.get(timeout=1e-3) break except Queue.Empty: + logging.debug( + self._log( + "{} wait for empty queue(with channel size: {})". + format(op_name, self.qsize()))) self._cv.wait() logging.debug( self._log("{} get data succ: {}".format(op_name, resp.__str__( @@ -360,16 +379,31 @@ class Channel(Queue.Queue): with self._cv: # data_idx = consumer_idx - base_idx while self._stop is False and self._consumers[ - op_name] - self._consumer_base_idx >= len(self._front_res): + op_name] - self._consumer_base_idx.value >= len( + self._front_res): + logging.debug( + self._log( + "({}) B self._consumers: {}, self._consumer_base_idx: {}, len(self._front_res): {}". + format(op_name, self._consumers, self. + _consumer_base_idx.value, len(self._front_res)))) try: - channeldata = self.get(timeout=0) + logging.debug( + self._log("{} try to get(with channel size: {})".format( + op_name, self.qsize()))) + #TODO: bug to fix + # (multiple processes) the queue is not empty, but it raise Queue.Empty + channeldata = self.get(timeout=1e-3) self._front_res.append(channeldata) break except Queue.Empty: + logging.debug( + self._log( + "{} wait for empty queue(with channel size: {})". + format(op_name, self.qsize()))) self._cv.wait() consumer_idx = self._consumers[op_name] - base_idx = self._consumer_base_idx + base_idx = self._consumer_base_idx.value data_idx = consumer_idx - base_idx resp = self._front_res[data_idx] logging.debug(self._log("{} get data: {}".format(op_name, resp))) @@ -379,14 +413,19 @@ class Channel(Queue.Queue): consumer_idx] == 0: self._idx_consumer_num.pop(consumer_idx) self._front_res.pop(0) - self._consumer_base_idx += 1 + self._consumer_base_idx.value += 1 self._consumers[op_name] += 1 new_consumer_idx = self._consumers[op_name] if self._idx_consumer_num.get(new_consumer_idx) is None: self._idx_consumer_num[new_consumer_idx] = 0 self._idx_consumer_num[new_consumer_idx] += 1 - + logging.debug( + self._log( + "({}) A self._consumers: {}, self._consumer_base_idx: {}, len(self._front_res): {}". + format(op_name, self._consumers, self._consumer_base_idx. + value, len(self._front_res)))) + logging.debug(self._log("{} notify all".format(op_name))) self._cv.notify_all() logging.debug(self._log("multi | {} get data succ!".format(op_name))) @@ -412,33 +451,42 @@ class Op(object): concurrency=1, timeout=-1, retry=2): - self._run = False + self._is_run = False self.name = name # to identify the type of OP, it must be globally unique self._concurrency = concurrency # amount of concurrency self.set_input_ops(inputs) - self.set_client(client_config, server_name, fetch_names) - self._server_model = server_model - self._server_port = server_port - self._device = device self._timeout = timeout self._retry = max(1, retry) self._input = None self._outputs = [] - def set_client(self, client_config, server_name, fetch_names): - self._client = None - if client_config is None or \ - server_name is None or \ - fetch_names is None: + self.with_serving = False + self._client_config = client_config + self._server_name = server_name + self._fetch_names = fetch_names + self._server_model = server_model + self._server_port = server_port + self._device = device + if self._client_config is not None and \ + self._server_name is not None and \ + self._fetch_names is not None and \ + self._server_model is not None and \ + self._server_port is not None and \ + self._device is not None: + self.with_serving = True + + def init_client(self, client_config, server_name, fetch_names): + if self.with_serving == False: + logging.debug("{} no client".format(self.name)) return + logging.debug("{} client_config: {}".format(self.name, client_config)) + logging.debug("{} server_name: {}".format(self.name, server_name)) + logging.debug("{} fetch_names: {}".format(self.name, fetch_names)) self._client = Client() self._client.load_client_config(client_config) self._client.connect([server_name]) self._fetch_names = fetch_names - def with_serving(self): - return self._client is not None - def get_input_channel(self): return self._input @@ -503,48 +551,59 @@ class Op(object): self._input.stop() for channel in self._outputs: channel.stop() - self._run = False + self._is_run = False def _parse_channeldata(self, channeldata): - data_id, error_pbdata = None, None + data_id, error_channeldata = None, None if isinstance(channeldata, dict): parsed_data = {} key = channeldata.keys()[0] - data_id = channeldata[key].pbdata.id + data_id = channeldata[key].id for _, data in channeldata.items(): - if data.pbdata.ecode != ChannelDataEcode.OK.value: - error_pbdata = data.pbdata + if data.ecode != ChannelDataEcode.OK.value: + error_channeldata = data break else: - data_id = channeldata.pbdata.id - if channeldata.pbdata.ecode != ChannelDataEcode.OK.value: - error_pbdata = channeldata.pbdata - return data_id, error_pbdata + data_id = channeldata.id + if channeldata.ecode != ChannelDataEcode.OK.value: + error_channeldata = channeldata + return data_id, error_channeldata - def _push_to_output_channels(self, data, name=None): + def _push_to_output_channels(self, data, channels, name=None): if name is None: name = self.name - for channel in self._outputs: + for channel in channels: channel.push(data, name) - def start(self, concurrency_idx): + def start(self): + proces = [] + for concurrency_idx in range(self._concurrency): + p = multiprocessing.Process( + target=self._run, + args=(concurrency_idx, self.get_input_channel(), + self.get_output_channels())) + p.start() + proces.append(p) + return proces + + def _run(self, concurrency_idx, input_channel, output_channels): + self.init_client(self._client_config, self._server_name, + self._fetch_names) op_info_prefix = "[{}|{}]".format(self.name, concurrency_idx) log = self._get_log_func(op_info_prefix) - self._run = True - while self._run: + self._is_run = True + while self._is_run: _profiler.record("{}-get_0".format(op_info_prefix)) - channeldata = self._input.front(self.name) + channeldata = input_channel.front(self.name) _profiler.record("{}-get_1".format(op_info_prefix)) logging.debug(log("input_data: {}".format(channeldata))) - data_id, error_pbdata = self._parse_channeldata(channeldata) + data_id, error_channeldata = self._parse_channeldata(channeldata) # error data in predecessor Op - if error_pbdata is not None: - self._push_to_output_channels( - ChannelData( - datatype=ChannelDataType.CHANNEL_PBDATA.value, - pbdata=error_pbdata)) + if error_channeldata is not None: + self._push_to_output_channels(error_channeldata, + output_channels) continue # preprecess @@ -560,31 +619,34 @@ class Op(object): ChannelData( ecode=ChannelDataEcode.NOT_IMPLEMENTED.value, error_info=error_info, - data_id=data_id)) + data_id=data_id), + output_channels) continue except TypeError as e: - # Error type in channeldata.pbdata.type + # Error type in channeldata.datatype error_info = log(e) logging.error(error_info) self._push_to_output_channels( ChannelData( ecode=ChannelDataEcode.TYPE_ERROR.value, error_info=error_info, - data_id=data_id)) + data_id=data_id), + output_channels) continue except Exception as e: error_info = log(e) logging.error(error_info) self._push_to_output_channels( ChannelData( - ecode=ChannelDataEcode.TYPE_ERROR.value, + ecode=ChannelDataEcode.UNKNOW.value, error_info=error_info, - data_id=data_id)) + data_id=data_id), + output_channels) continue # midprocess call_future = None - if self.with_serving(): + if self.with_serving: ecode = ChannelDataEcode.OK.value _profiler.record("{}-midp_0".format(op_info_prefix)) if self._timeout <= 0: @@ -620,20 +682,30 @@ class Op(object): self._push_to_output_channels( ChannelData( ecode=ecode, error_info=error_info, - data_id=data_id)) + data_id=data_id), + output_channels) continue _profiler.record("{}-midp_1".format(op_info_prefix)) # postprocess output_data = None _profiler.record("{}-postp_0".format(op_info_prefix)) - if self.with_serving(): + if self.with_serving: # use call_future output_data = ChannelData( datatype=ChannelDataType.CHANNEL_FUTURE.value, future=call_future, data_id=data_id, callback_func=self.postprocess) + #TODO: for future are not picklable + npdata = self.postprocess(call_future.result()) + self._push_to_output_channels( + ChannelData( + ChannelDataType.CHANNEL_NPDATA.value, + npdata=npdata, + data_id=data_id), + output_channels) + continue else: try: postped_data = self.postprocess(preped_data) @@ -644,7 +716,8 @@ class Op(object): self._push_to_output_channels( ChannelData( ecode=ecode, error_info=error_info, - data_id=data_id)) + data_id=data_id), + output_channels) continue if not isinstance(postped_data, dict): ecode = ChannelDataEcode.TYPE_ERROR.value @@ -654,7 +727,8 @@ class Op(object): self._push_to_output_channels( ChannelData( ecode=ecode, error_info=error_info, - data_id=data_id)) + data_id=data_id), + output_channels) continue output_data = ChannelData( @@ -665,7 +739,7 @@ class Op(object): # push data to channel (if run succ) _profiler.record("{}-push_0".format(op_info_prefix)) - self._push_to_output_channels(output_data) + self._push_to_output_channels(output_data, output_channels) _profiler.record("{}-push_1".format(op_info_prefix)) def _log(self, info): @@ -701,22 +775,25 @@ class VirtualOp(Op): channel.add_producer(op.name) self._outputs.append(channel) - def start(self, concurrency_idx): + def _run(self, input_channel, output_channels): op_info_prefix = "[{}|{}]".format(self.name, concurrency_idx) log = self._get_log_func(op_info_prefix) - self._run = True - while self._run: + self._is_run = True + while self._is_run: _profiler.record("{}-get_0".format(op_info_prefix)) - channeldata = self._input.front(self.name) + channeldata = input_channel.front(self.name) _profiler.record("{}-get_1".format(op_info_prefix)) _profiler.record("{}-push_0".format(op_info_prefix)) if isinstance(channeldata, dict): for name, data in channeldata.items(): - self._push_to_output_channels(data, name=name) + self._push_to_output_channels( + data, channels=output_channels, name=name) else: - self._push_to_output_channels(channeldata, - self._virtual_pred_ops[0].name) + self._push_to_output_channels( + channeldata, + channels=output_channels, + name=self._virtual_pred_ops[0].name) _profiler.record("{}-push_1".format(op_info_prefix)) @@ -768,7 +845,7 @@ class GeneralPythonService( self._log('data must be ChannelData type, but get {}'. format(type(channeldata)))) with self._cv: - data_id = channeldata.pbdata.id + data_id = channeldata.id self._globel_resp_dict[data_id] = channeldata self._cv.notify_all() @@ -788,33 +865,33 @@ class GeneralPythonService( def _pack_data_for_infer(self, request): logging.debug(self._log('start inferce')) - pbdata = channel_pb2.ChannelData() data_id = self._get_next_id() - pbdata.id = data_id - pbdata.ecode = ChannelDataEcode.OK.value + npdata = {} try: for idx, name in enumerate(request.feed_var_names): logging.debug( self._log('name: {}'.format(request.feed_var_names[idx]))) logging.debug( self._log('data: {}'.format(request.feed_insts[idx]))) - inst = channel_pb2.Inst() - inst.data = request.feed_insts[idx] - inst.shape = request.shape[idx] - inst.name = name - inst.type = request.type[idx] - pbdata.insts.append(inst) + npdata[name] = np.frombuffer( + request.feed_insts[idx], dtype=request.type[idx]) + npdata[name].shape = np.frombuffer( + request.shape[idx], dtype="int32") except Exception as e: - pbdata.ecode = ChannelDataEcode.RPC_PACKAGE_ERROR.value - pbdata.error_info = "rpc package error" - return ChannelData( - datatype=ChannelDataType.CHANNEL_PBDATA.value, - pbdata=pbdata), data_id + return ChannelData( + ecode=ChannelDataEcode.RPC_PACKAGE_ERROR.value, + error_info="rpc package error", + data_id=data_id), data_id + else: + return ChannelData( + datatype=ChannelDataType.CHANNEL_NPDATA.value, + npdata=npdata, + data_id=data_id), data_id def _pack_data_for_resp(self, channeldata): logging.debug(self._log('get channeldata')) resp = pyservice_pb2.Response() - resp.ecode = channeldata.pbdata.ecode + resp.ecode = channeldata.ecode if resp.ecode == ChannelDataEcode.OK.value: if channeldata.datatype == ChannelDataType.CHANNEL_PBDATA.value: for inst in channeldata.pbdata.insts: @@ -834,10 +911,10 @@ class GeneralPythonService( resp.type.append(str(var.dtype)) else: raise TypeError( - self._log("Error type({}) in pbdata.type.".format( + self._log("Error type({}) in datatype.".format( channeldata.datatype))) else: - resp.error_info = channeldata.pbdata.error_info + resp.error_info = channeldata.error_info return resp def inference(self, request, context): @@ -857,11 +934,11 @@ class GeneralPythonService( resp_channeldata = self._get_data_in_globel_resp_dict(data_id) _profiler.record("{}-fetch_1".format(self.name)) - if resp_channeldata.pbdata.ecode == ChannelDataEcode.OK.value: + if resp_channeldata.ecode == ChannelDataEcode.OK.value: break if i + 1 < self._retry: logging.warn("retry({}): {}".format( - i + 1, resp_channeldata.pbdata.error_info)) + i + 1, resp_channeldata.error_info)) _profiler.record("{}-postpack_0".format(self.name)) resp = self._pack_data_for_resp(resp_channeldata) @@ -875,12 +952,12 @@ class PyServer(object): self._channels = [] self._user_ops = [] self._actual_ops = [] - self._op_threads = [] self._port = None self._worker_num = None self._in_channel = None self._out_channel = None self._retry = retry + self._manager = multiprocessing.Manager() _profiler.enable(profile) def add_channel(self, channel): @@ -989,7 +1066,7 @@ class PyServer(object): for o_idx, op in enumerate(actual_next_view): if op.name in processed_op: continue - channel = Channel(name=channel_name_gen.next()) + channel = Channel(self._manager, name=channel_name_gen.next()) channels.append(channel) logging.debug("{} => {}".format(channel.name, op.name)) op.add_input_channel(channel) @@ -1020,7 +1097,7 @@ class PyServer(object): other_op.name)) other_op.add_input_channel(channel) processed_op.add(other_op.name) - output_channel = Channel(name=channel_name_gen.next()) + output_channel = Channel(self._manager, name=channel_name_gen.next()) channels.append(output_channel) last_op = dag_views[-1][0] last_op.add_output_channel(output_channel) @@ -1044,30 +1121,22 @@ class PyServer(object): self._in_channel = input_channel self._out_channel = output_channel for op in self._actual_ops: - if op.with_serving(): + if op.with_serving: self.prepare_serving(op) self.gen_desc() - def _op_start_wrapper(self, op, concurrency_idx): - return op.start(concurrency_idx) - def _run_ops(self): + proces = [] for op in self._actual_ops: - op_concurrency = op.get_concurrency() - logging.debug("run op: {}, op_concurrency: {}".format( - op.name, op_concurrency)) - for c in range(op_concurrency): - th = threading.Thread( - target=self._op_start_wrapper, args=(op, c)) - th.start() - self._op_threads.append(th) + proces.extend(op.start()) + return proces def _stop_ops(self): for op in self._actual_ops: op.stop() def run_server(self): - self._run_ops() + op_proces = self._run_ops() server = grpc.server( futures.ThreadPoolExecutor(max_workers=self._worker_num)) general_python_service_pb2_grpc.add_GeneralPythonServiceServicer_to_server( @@ -1077,8 +1146,8 @@ class PyServer(object): server.start() server.wait_for_termination() self._stop_ops() # TODO - for th in self._op_threads: - th.join() + for p in op_proces: + p.join() def prepare_serving(self, op): model_path = op._server_model