fix profile in client

python/pipeline/channel.py

@@ -90,12 +90,12 @@ class ChannelData(object):
         self.ecode = ecode
         self.error_info = error_info
         self.client_need_profile = client_need_profile
-        self.profile_data_list = []
+        self.profile_data_set = set()
 
-    def add_profile(self, profile_list):
+    def add_profile(self, profile_set):
         if self.client_need_profile is False:
             self.client_need_profile = True
-        self.profile_data_list.extend(profile_list)
+        self.profile_data_set |= profile_set
 
     @staticmethod
     def check_dictdata(dictdata):

python/pipeline/dag.py

@@ -241,9 +241,9 @@ class DAGExecutor(object):
         # add profile info into rpc_resp
         profile_value = ""
         if resp_channeldata.client_need_profile:
-            profile_list = resp_channeldata.profile_data_list
-            profile_list.append(profile_str)
-            profile_value = "".join(profile_list)
+            profile_set = resp_channeldata.profile_data_set
+            profile_set.add(profile_str)
+            profile_value = "".join(list(profile_set))
         rpc_resp.key.append(self._client_profile_key)
         rpc_resp.value.append(profile_value)
 

python/pipeline/operator.py

@@ -162,7 +162,7 @@ class Op(object):
 
     def _parse_channeldata(self, channeldata_dict):
         data_id, error_channeldata = None, None
-        client_need_profile, profile_list = False, []
+        client_need_profile, profile_set = False, set()
         parsed_data = {}
 
         key = list(channeldata_dict.keys())[0]
@@ -175,32 +175,32 @@ class Op(object):
                 break
             parsed_data[name] = data.parse()
             if client_need_profile:
-                profile_list.extend(data.profile_data_list)
+                profile_set |= data.profile_data_set
         return (data_id, error_channeldata, parsed_data, client_need_profile,
-                profile_list)
+                profile_set)
 
     def _push_to_output_channels(self,
                                  data,
                                  channels,
                                  name=None,
                                  client_need_profile=False,
-                                 profile_list=None):
+                                 profile_set=None):
         if name is None:
             name = self.name
         self._add_profile_into_channeldata(data, client_need_profile,
-                                           profile_list)
+                                           profile_set)
         for channel in channels:
             channel.push(data, name)
 
     def _add_profile_into_channeldata(self, data, client_need_profile,
-                                      profile_list):
+                                      profile_set):
         profile_str = self._profiler.gen_profile_str()
         if self._server_use_profile:
             sys.stderr.write(profile_str)
 
-        if client_need_profile and profile_list is not None:
-            profile_list.append(profile_str)
-            data.add_profile(profile_list)
+        if client_need_profile and profile_set is not None:
+            profile_set.add(profile_str)
+            data.add_profile(profile_set)
 
     def start_with_process(self, client_type):
         proces = []
@@ -398,7 +398,7 @@ class Op(object):
             _LOGGER.debug(log("input_data: {}".format(channeldata_dict)))
 
             (data_id, error_channeldata, parsed_data, client_need_profile,
-             profile_list) = self._parse_channeldata(channeldata_dict)
+             profile_set) = self._parse_channeldata(channeldata_dict)
             # error data in predecessor Op
             if error_channeldata is not None:
                 try:
@@ -421,7 +421,7 @@ class Op(object):
                         error_channeldata,
                         output_channels,
                         client_need_profile=client_need_profile,
-                        profile_list=profile_list)
+                        profile_set=profile_set)
                 except ChannelStopError:
                     _LOGGER.debug(log("stop."))
                     break
@@ -438,7 +438,7 @@ class Op(object):
                         error_channeldata,
                         output_channels,
                         client_need_profile=client_need_profile,
-                        profile_list=profile_list)
+                        profile_set=profile_set)
                 except ChannelStopError:
                     _LOGGER.debug(log("stop."))
                     break
@@ -455,7 +455,7 @@ class Op(object):
                         error_channeldata,
                         output_channels,
                         client_need_profile=client_need_profile,
-                        profile_list=profile_list)
+                        profile_set=profile_set)
                 except ChannelStopError:
                     _LOGGER.debug(log("stop."))
                     break
@@ -468,7 +468,7 @@ class Op(object):
                     output_data,
                     output_channels,
                     client_need_profile=client_need_profile,
-                    profile_list=profile_list)
+                    profile_set=profile_set)
             except ChannelStopError:
                 _LOGGER.debug(log("stop."))
                 break