From d4e5b946eb4bf1ad9a8e74112c0411939b3a9c28 Mon Sep 17 00:00:00 2001
From: TeslaZhao <zhaolisoftware@163.com>
Date: Mon, 8 Feb 2021 18:33:41 +0800
Subject: [PATCH] fix pipeline bugs of py3.6

---
 doc/{ => deprecated}/IMDB_GO_CLIENT.md        |  0
 doc/{ => deprecated}/IMDB_GO_CLIENT_CN.md     |  0
 .../MODEL_ENSEMBLE_IN_PADDLE_SERVING.md       |  0
 .../MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md    |  0
 doc/{ => deprecated}/NEW_OPERATOR.md          |  0
 doc/{ => deprecated}/NEW_OPERATOR_CN.md       |  0
 python/pipeline/channel.py                    | 20 ++++++++++++-------
 python/pipeline/profiler.py                   |  2 +-
 8 files changed, 14 insertions(+), 8 deletions(-)
 rename doc/{ => deprecated}/IMDB_GO_CLIENT.md (100%)
 rename doc/{ => deprecated}/IMDB_GO_CLIENT_CN.md (100%)
 rename doc/{ => deprecated}/MODEL_ENSEMBLE_IN_PADDLE_SERVING.md (100%)
 rename doc/{ => deprecated}/MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md (100%)
 rename doc/{ => deprecated}/NEW_OPERATOR.md (100%)
 rename doc/{ => deprecated}/NEW_OPERATOR_CN.md (100%)

diff --git a/doc/IMDB_GO_CLIENT.md b/doc/deprecated/IMDB_GO_CLIENT.md
similarity index 100%
rename from doc/IMDB_GO_CLIENT.md
rename to doc/deprecated/IMDB_GO_CLIENT.md
diff --git a/doc/IMDB_GO_CLIENT_CN.md b/doc/deprecated/IMDB_GO_CLIENT_CN.md
similarity index 100%
rename from doc/IMDB_GO_CLIENT_CN.md
rename to doc/deprecated/IMDB_GO_CLIENT_CN.md
diff --git a/doc/MODEL_ENSEMBLE_IN_PADDLE_SERVING.md b/doc/deprecated/MODEL_ENSEMBLE_IN_PADDLE_SERVING.md
similarity index 100%
rename from doc/MODEL_ENSEMBLE_IN_PADDLE_SERVING.md
rename to doc/deprecated/MODEL_ENSEMBLE_IN_PADDLE_SERVING.md
diff --git a/doc/MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md b/doc/deprecated/MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md
similarity index 100%
rename from doc/MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md
rename to doc/deprecated/MODEL_ENSEMBLE_IN_PADDLE_SERVING_CN.md
diff --git a/doc/NEW_OPERATOR.md b/doc/deprecated/NEW_OPERATOR.md
similarity index 100%
rename from doc/NEW_OPERATOR.md
rename to doc/deprecated/NEW_OPERATOR.md
diff --git a/doc/NEW_OPERATOR_CN.md b/doc/deprecated/NEW_OPERATOR_CN.md
similarity index 100%
rename from doc/NEW_OPERATOR_CN.md
rename to doc/deprecated/NEW_OPERATOR_CN.md
diff --git a/python/pipeline/channel.py b/python/pipeline/channel.py
index 9f06c445..e4c38b9e 100644
--- a/python/pipeline/channel.py
+++ b/python/pipeline/channel.py
@@ -326,7 +326,10 @@ class ProcessChannel(object):
             with self._cv:
                 while self._stop.value == 0:
                     try:
-                        self._que.put({op_name: channeldata}, timeout=0)
+                        self._que.put((channeldata.id, {
+                            op_name: channeldata
+                        }),
+                                      timeout=0)
                         break
                     except Queue.Full:
                         self._cv.wait()
@@ -378,7 +381,7 @@ class ProcessChannel(object):
             else:
                 while self._stop.value == 0:
                     try:
-                        self._que.put(put_data, timeout=0)
+                        self._que.put((data_id, put_data), timeout=0)
                         break
                     except Queue.Empty:
                         self._cv.wait()
@@ -414,7 +417,7 @@ class ProcessChannel(object):
             with self._cv:
                 while self._stop.value == 0 and resp is None:
                     try:
-                        resp = self._que.get(timeout=0)
+                        resp = self._que.get(timeout=0)[1]
                         break
                     except Queue.Empty:
                         if timeout is not None:
@@ -459,7 +462,7 @@ class ProcessChannel(object):
             while self._stop.value == 0 and self._consumer_cursors[
                     op_name] - self._base_cursor.value >= len(self._output_buf):
                 try:
-                    channeldata = self._que.get(timeout=0)
+                    channeldata = self._que.get(timeout=0)[1]
                     self._output_buf.append(channeldata)
                     list_values = list(channeldata.values())
                     _LOGGER.debug(
@@ -633,7 +636,10 @@ class ThreadChannel(Queue.PriorityQueue):
             with self._cv:
                 while self._stop is False:
                     try:
-                        self.put({op_name: channeldata}, timeout=0)
+                        self.put((channeldata.id, {
+                            op_name: channeldata
+                        }),
+                                 timeout=0)
                         break
                     except Queue.Full:
                         self._cv.wait()
@@ -680,7 +686,7 @@ class ThreadChannel(Queue.PriorityQueue):
             else:
                 while self._stop is False:
                     try:
-                        self.put(put_data, timeout=0)
+                        self.put((data_id, put_data), timeout=0)
                         break
                     except Queue.Empty:
                         self._cv.wait()
@@ -716,7 +722,7 @@ class ThreadChannel(Queue.PriorityQueue):
             with self._cv:
                 while self._stop is False and resp is None:
                     try:
-                        resp = self.get(timeout=0)
+                        resp = self.get(timeout=0)[1]
                         break
                     except Queue.Empty:
                         if timeout is not None:
diff --git a/python/pipeline/profiler.py b/python/pipeline/profiler.py
index b83bdd1d..18845a75 100644
--- a/python/pipeline/profiler.py
+++ b/python/pipeline/profiler.py
@@ -120,7 +120,7 @@ class PerformanceTracer(object):
                                                          tot_cost))
 
             if "DAG" in op_cost:
-                calls = op_cost["DAG"].values()
+                calls = list(op_cost["DAG"].values())
                 calls.sort()
                 tot = len(calls)
                 qps = 1.0 * tot / self._interval_s
-- 
GitLab