Add time prefix for PoolManager.

fluid/DeepASR/data_utils/async_data_reader.py

@@ -218,8 +218,6 @@ class AsyncDataReader(object):
         self._sample_proc_num = self._proc_num - 2
         self._verbose = verbose
         self._force_exit = ForceExitWrapper(self._manager.Value('b', False))
-        self._pool_manager = SharedMemoryPoolManager(self._batch_buffer_size *
-                                                     3, self._manager)
 
     def generate_bucket_list(self, is_shuffle):
         if self._block_info_list is None:
@@ -424,6 +422,9 @@ class AsyncDataReader(object):
         sample_queue = self._start_async_processing()
         batch_queue = self._manager.Queue(self._batch_buffer_size)
 
+        self._pool_manager = SharedMemoryPoolManager(self._batch_buffer_size *
+                                                     3, self._manager)
+
         assembling_proc = DaemonProcessGroup(
             proc_num=1,
             target=batch_assembling_task,
@@ -439,3 +440,6 @@ class AsyncDataReader(object):
                 if isinstance(batch_data, EpochEndSignal):
                     break
                 yield batch_data
+
+        # clean the shared memory
+        del self._pool_manager

fluid/DeepASR/data_utils/util.py

 from __future__ import absolute_import
 from __future__ import division
 from __future__ import print_function
-import sys
+import sys, time
 from six import reraise
 from tblib import Traceback
 from multiprocessing import Manager, Process
@@ -161,9 +161,10 @@ class SharedMemoryPoolManager(object):
     def __init__(self, pool_size, manager, name_prefix='/deep_asr'):
         self._names = []
         self._dict = manager.dict()
+        self._time_prefix = time.strftime('%Y%m%d%H%M%S')
 
         for i in xrange(pool_size):
-            name = name_prefix + '_' + str(i)
+            name = name_prefix + '_' + self._time_prefix + '_' + str(i)
             self._dict[name] = SharedNDArray(name)
             self._names.append(name)