Merge pull request #36 from HBS-HBX/#35_open_multiprocessing_log_in_context_handler

closes #35 open multiprocessing log in context handler

Author: codekiln

CHANGELOG.rst

@@ -1,9 +1,15 @@
 Changelog
 ---------
 
+0.7.5 (2018-08-20)
+~~~~~~~~~~~~~~~~~~
+* fix `#35 open multiprocessing log in context handler
+<https://github.com/HBS-HBX/django-elastic-migrations/issues/35>`_
+
 0.7.4 (2018-08-15)
 ~~~~~~~~~~~~~~~~~~
-* fix #33 error when nothing to resume using --resume
+* fix `#33 error when nothing to resume using --resume
+<https://github.com/HBS-HBX/django-elastic-migrations/issues/33>`_
 
 0.7.3 (2018-08-14)
 ~~~~~~~~~~~~~~~~~~

django_elastic_migrations/__init__.py

@@ -10,7 +10,7 @@
 from django_elastic_migrations.utils import loading
 from django_elastic_migrations.utils.django_elastic_migrations_log import get_logger
 
-__version__ = '0.7.4'
+__version__ = '0.7.5'
 
 default_app_config = 'django_elastic_migrations.apps.DjangoElasticMigrationsConfig'  # pylint: disable=invalid-name
 

django_elastic_migrations/utils/django_elastic_migrations_log.py

@@ -1,13 +1,31 @@
 from __future__ import (absolute_import, division, print_function, unicode_literals)
+
 import logging
+
 from multiprocessing_logging import install_mp_handler
 
 
-install_mp_handler()
+mp_logging_enabled = False
 
 
 def get_logger(name="django_elastic_migrations"):
     real_logger = logging.getLogger(name)
     for level in ['CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG']:
         setattr(real_logger, level, getattr(logging, level))
     return real_logger
+
+
+def start_multiprocessing_logging():
+    """
+    start multiprocessing-logging to make each handler a synchronized queue.
+    Called by django_elastic_migrations.utils.multiprocessing_utils.DjangoMultiProcess.__enter__
+    See https://github.com/jruere/multiprocessing-logging/blob/master/multiprocessing_logging.py
+    TBD: uninstall mp_logging; currently we don't stop and reset mp logging; it's a one-shot for the process.
+    :return:
+    :rtype: None
+    """
+    global mp_logging_enabled
+
+    if not mp_logging_enabled:
+        mp_logging_enabled = True
+        install_mp_handler()

django_elastic_migrations/utils/multiprocessing_utils.py

@@ -5,6 +5,8 @@
 
 from django.db import reset_queries
 
+from django_elastic_migrations.utils.django_elastic_migrations_log import start_multiprocessing_logging
+
 try:
     # python 2
     # noinspection PyCompatibility
@@ -32,7 +34,6 @@
 https://engineering.talentpair.com/django-multiprocessing-153dbcf51dab
 """
 
-
 USE_ALL_WORKERS = 999
 
 
@@ -185,6 +186,7 @@ def __init__(self, num_workers=None, max_workers=None, log_debug_info=False, sta
         self.queue = None
 
     def __enter__(self):
+        start_multiprocessing_logging()
         close_service_connections()
         return self
 

tests/test_django_elastic_migrations_log.py

@@ -0,0 +1,11 @@
+from django.test import TestCase
+
+from django_elastic_migrations.utils import django_elastic_migrations_log
+
+
+class TestDEMLog(TestCase):
+
+    def test_mp_logging_setup(self):
+        django_elastic_migrations_log.start_multiprocessing_logging()
+
+        self.assertTrue(django_elastic_migrations_log.mp_logging_enabled)

tests/test_multiprocessing_utils.py

@@ -0,0 +1,34 @@
+from multiprocessing import cpu_count
+
+from django.test import TestCase
+
+from django_elastic_migrations.utils.multiprocessing_utils import DjangoMultiProcess
+
+
+def add_1(num):
+    return { 'job_id': num, 'result': num + 1 }
+
+
+class TestMultiprocessingUtils(TestCase):
+
+    def test_basic_multiprocessing(self):
+        """
+        Do a basic test of DjangoMultiProcess that doesn't touch the database
+        :return:
+        :rtype:
+        """
+
+        one_to_ten = range(1, 10)
+
+        workers = cpu_count()
+        django_multiprocess = DjangoMultiProcess(workers, log_debug_info=3)
+
+        with django_multiprocess:
+            django_multiprocess.map(add_1, one_to_ten)
+
+        results = django_multiprocess.results()
+
+        for result_obj in results:
+            job_id = result_obj.get('job_id')
+            result = result_obj.get('result')
+            self.assertEqual(job_id + 1, result)