From fe4828d4f0da37d0ea95cc961694be8178198b37 Mon Sep 17 00:00:00 2001
From: barrierye <barriery@qq.com>
Date: Mon, 10 Aug 2020 17:30:30 +0800
Subject: [PATCH] bug fix

---
 python/pipeline/operator.py        | 4 ++--
 python/pipeline/pipeline_server.py | 8 ++++++--
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/python/pipeline/operator.py b/python/pipeline/operator.py
index ddac492e..9d4398fb 100644
--- a/python/pipeline/operator.py
+++ b/python/pipeline/operator.py
@@ -575,7 +575,7 @@ class Op(object):
                 _LOGGER.debug("{} Stop.".format(op_info_prefix))
                 self._finalize(is_thread_op)
                 break
-            if len(parsed_data_dict) == 0:
+            if len(preped_data_dict) == 0:
                 continue
 
             # process
@@ -610,7 +610,7 @@ class Op(object):
             try:
                 for data_id, err_channeldata in err_channeldata_dict.items():
                     self._push_to_output_channels(
-                        data=error_channeldata,
+                        data=err_channeldata,
                         channels=output_channels,
                         client_need_profile=need_profile_dict[data_id],
                         profile_set=profile_dict[data_id])
diff --git a/python/pipeline/pipeline_server.py b/python/pipeline/pipeline_server.py
index 6ad34e34..04d2c4fc 100644
--- a/python/pipeline/pipeline_server.py
+++ b/python/pipeline/pipeline_server.py
@@ -114,7 +114,9 @@ class PipelineServer(object):
                     worker.join()
         else:
             server = grpc.server(
-                futures.ThreadPoolExecutor(max_workers=self._worker_num))
+                futures.ThreadPoolExecutor(max_workers=self._worker_num),
+                options=[('grpc.max_send_message_length', 256 * 1024 * 1024),
+                    ('grpc.max_receive_message_length', 256 * 1024 * 1024)])
             pipeline_service_pb2_grpc.add_PipelineServiceServicer_to_server(
                 PipelineServicer(self._response_op, self._conf), server)
             server.add_insecure_port('[::]:{}'.format(self._port))
@@ -122,7 +124,9 @@ class PipelineServer(object):
             server.wait_for_termination()
 
     def _run_server_func(self, bind_address, response_op, dag_conf, worker_idx):
-        options = (('grpc.so_reuseport', 1), )
+        options = [('grpc.so_reuseport', 1),
+                ('grpc.max_send_message_length', 256 * 1024 * 1024),
+                ('grpc.max_send_message_length', 256 * 1024 * 1024)]
         server = grpc.server(
             futures.ThreadPoolExecutor(
                 max_workers=1, ), options=options)
-- 
GitLab