bug fix

python/pipeline/dag.py

@@ -559,7 +559,8 @@ class DAG(object):
         self._pack_func = pack_func
         self._unpack_func = unpack_func
 
-        self._tracer.set_channels(self._channels)
+        if self._tracer is not None:
+            self._tracer.set_channels(self._channels)
 
         return self._input_channel, self._output_channel, self._pack_func, self._unpack_func
 

python/pipeline/operator.py

@@ -364,8 +364,9 @@ class Op(object):
                     input_offset.append(offset)
             else:
                 _LOGGER.critical(
-                        "{} Failed to process: expect input type is dict or listi"
-                        ", but get {}".format(op_info_prefix, type(one_input)))
+                        "{} Failed to process: expect input type is dict(sample"
+                        " input) or list(batch input), but get {}".format(
+                            op_info_prefix, type(one_input)))
                 os._exit(-1)
 
             midped_batch = None

python/pipeline/pipeline_server.py

@@ -92,8 +92,9 @@ class PipelineServer(object):
             json.dumps(
                 conf, indent=4, separators=(',', ':'))))
         if self._build_dag_each_worker is True:
-            _LOGGER.info(
-                "(Make sure that install grpcio whl with --no-binary flag)")
+            _LOGGER.warning(
+                "(Make sure that install grpcio whl with --no-binary flag: "
+                "pip install grpcio --no-binary grpcio)")
         _LOGGER.info("-------------------------------------------")
 
         self._conf = conf
@@ -181,7 +182,7 @@ class ServerYamlConfChecker(object):
 
     @staticmethod
     def check_tracer_conf(conf):
-        default_conf = {"interval_s": 600, }
+        default_conf = {"interval_s": -1, }
 
         conf_type = {"interval_s": int, }
 

python/pipeline/profiler.py

@@ -130,7 +130,7 @@ class PerformanceTracer(object):
                 _LOGGER.info("\tQuery count[{}]".format(tot))
                 _LOGGER.info("\tQPS[{} q/s]".format(qps))
                 _LOGGER.info("\tSucc[{}]".format(1 - 1.0 * err_count / tot))
-                _LOGGER.info("\tError req[{}]".format([", ".join([str(x) for x in err_request])]))
+                _LOGGER.info("\tError req[{}]".format(", ".join([str(x) for x in err_request)]))
                 _LOGGER.info("\tLatency:")
                 _LOGGER.info("\t\tave[{} ms]".format(ave_cost))
                 for latency in latencys:

python/pipeline/util.py

@@ -16,6 +16,7 @@ import sys
 import logging
 import threading
 import multiprocessing
+import multiprocessing.managers
 if sys.version_info.major == 2:
     import Queue
     from Queue import PriorityQueue