_test_multiprocessing_.py

_test_multiprocessing_.py

#!/usr/local/bin/python
# -*- coding: utf-8 -*-
'''_test_multiprocessing_
test with APScheduler https://www.youtube.com/watch?v=NffYUw4M2ls
logging with multiprocessing
 (not use FileHandler) should use SocketHandler or QueueHandler
 https://docs.python.org/ja/3/library/logging.handlers.html
QueueListener and QueueHandler
 https://ikatakos.com/pot/programming/python/packages/multiprocessing/queuehandler
 https://docs.python.org/ja/3/howto/logging-cookbook.html#logging-to-a-single-file-from-multiple-processes
 https://stackoverflow.com/questions/76172847/python-logging-using-queuelisteners-and-queuehandlers-along-with-file-config
 https://qiita.com/simonritchie/items/1ce3914eb5444d2157ac#multiprocessing%E3%81%AEprocess%E3%81%A8pool%E3%81%AE%E3%81%A9%E3%81%A1%E3%82%89%E3%82%92%E4%BD%BF%E3%81%86%E3%81%B9%E3%81%8D%E3%81%8B
'''

import sys, os
import time
import logging
import logging.handlers
import multiprocessing

LOGF = 'e:/virtual2/_e_r_r_/_test_multiprocessing_log_.txt'

class ListenerProccess(object):
  def __init__(self):
    self.running = False

  def start(self, que):
    self.running = True
    self.ch, ch = multiprocessing.Pipe()
    self.proc = multiprocessing.Process(target=self.child, args=(que, ch),
      daemon=True)
    self.proc.start()

  def child(self, que, ch):
    '''parent:self.running != child:self.running'''
    handler = [] # console, file
    handler.append(logging.StreamHandler(sys.stderr))
    console_formatter = logging.Formatter('%(processName)s %(message)s')
    handler[-1].setFormatter(console_formatter)
    handler.append(logging.handlers.RotatingFileHandler(LOGF, 'a', 3000, 5))
    file_formatter = logging.Formatter(
      '%(asctime)s %(processName)-10s %(levelname)-8s %(message)s')
    handler[-1].setFormatter(file_formatter)
    listener = logging.handlers.QueueListener(que, *handler)
    listener.start()
    while self.running: time.sleep(1); self.running = ch.recv()[0]
    listener.stop()

  def stop(self):
    self.running = False
    self.ch.send([False])
    # self.ch.close()
    self.proc.join()

def worker(num, que):
  logger = logging.getLogger()
  handler = logging.handlers.QueueHandler(que)
  logger.addHandler(handler)
  logger.setLevel(logging.DEBUG)
  name = multiprocessing.current_process().name
  logger.info(f'worker started: {name}')
  print(f'{num}')
  logger.info(f'{name}: {num}')
  time.sleep(5)
  logger.info(f'worker finished: {name}')

if __name__ == '__main__':
  que = multiprocessing.Queue()
  lp = ListenerProccess()
  lp.start(que)

  w = [multiprocessing.Process(target=worker, args=(i, que)) for i in range(5)]
  for p in w: p.start()
  if False:
    for p in w: p.join()
  print('done.')
  if True:
    time.sleep(6)
    lp.stop()