Блокировка для очереди, общей для процессов - PullRequest
Новая
       15

Блокировка для очереди, общей для процессов

0 голосов
/ 09 июля 2019

В основном я переписываю свой код с Thread на Process и у меня есть некоторые "сюрпризы".

Как заблокировать очередь, совместно используемую несколькими процессами?(сейчас я использую threading.Lock) 

class DataProviderProcess:
    def __init__(self, dataset_dir, n_images, batch_size, use_queue_lock,
                 input_img_w=299, input_img_h=299):
        img_filepaths = sorted(get_image_filepaths(dataset_dir))
        img_filepaths = img_filepaths[:n_images]
        self.img_filepath_queue = multiprocessing.Manager().Queue()
        for img_filepath in img_filepaths:
            self.img_filepath_queue.put_nowait(img_filepath)
        self.input_img_w = input_img_w
        self.input_img_h = input_img_h
        self.batch_size = batch_size
        self.use_queue_lock = use_queue_lock
        self.queue_lock = Lock()

    def get_batch(self, thread_id):    
        img_batch = []
        try:
            if self.use_queue_lock:
                self.queue_lock.acquire()

            for _ in range(self.batch_size):
                img_filepath = self.img_filepath_queue.get(block=False)
                print('DEBUG: self.img_filepath_queue.qsize()', self.img_filepath_queue.qsize(),
                      'thread_id:', thread_id)
                img = cv2.imread(img_filepath)
                img = cv2.resize(img, (self.input_img_w, self.input_img_h), interpolation=cv2.INTER_LINEAR)
                img_batch.append(img)
            img_batch = np.array(img_batch)

            if self.use_queue_lock:
                self.queue_lock.release()

            return img_batch
        except queue.Empty:
            if len(img_batch) > 0:
                img_batch = np.array(img_batch)

                if self.use_queue_lock:
                    self.queue_lock.release()

                return img_batch
            else:

                if self.use_queue_lock:
                    self.queue_lock.release()

                return None

Вот почему я не могу использовать pass self._process to Process?

Почему multiprocessing.Manager().Queue() следует использовать сProcess а не queue.Queue()? 

class BatchLoaderProcess:
    def __init__(self, data_provider: DataProviderProcess, n_threads=8):
        #self.batch_queue = Queue(data_provider.batch_size * 4) # v1
        self.batch_queue = multiprocessing.Manager().Queue(data_provider.batch_size * 4)
        self.data_provider = data_provider

        self.thread_list = []
        for thread_id in range(n_threads):
            #self.thread_list.append(Process(target=self._process)) # v1
            #self.thread_list.append(Process(target=self._process, args=(self,))) # v2
            self.thread_list.append(Process(target=BatchLoaderProcess._process, args=(self, thread_id))) # v3

        for t in self.thread_list:
            t.start()

    def _process(self, thread_id):
        while True:
            img_batch = self.data_provider.get_batch(thread_id)
            if img_batch is None:
                break
            self.batch_queue.put(img_batch)

    def get_batch(self):
        try:
            img_batch = self.batch_queue.get(block=True, timeout=1)
            return img_batch
        except queue.Empty:
            return None

1 Ответ

0 голосов
/ 09 июля 2019

При использовании multiprocessing должны использоваться механизмы очереди и блокировки, см. https://docs.python.org/3.5/library/multiprocessing.html. Я подозреваю, что это имеет много общего с GIL, см. Многопроцессорная обработка по отношению к Python для многопоточности

Многопроцессорные очереди имеют встроенный механизм блокировки. Следовательно, вам не нужно снова блокировать очередь.

Вот почему я не могу использовать pass self._process для Process?

Потому что self является аргументом «статического» метода.

...