{AH} fixes for gevent 1.5

Author: AndreasHeger

.gitignore

@@ -12,3 +12,4 @@ ruffus_development.*
 *~
 *.swp
 
+*.time

ruffus/file_name_parameters.py

@@ -1113,7 +1113,17 @@ def yield_io_params_per_job(input_params,
             else:
                 input_param = orig_input_param
 
-            # extras
+            # extras - this statement applies transformations on the
+            # extra parameters. It traverses nested data
+            # structures. The transformation forces a copy of each
+            # parameter which causes memory duplication. The memory use
+            # can be significant if extras is large.
+            #
+            # For example, if extras contains a list of 10000 files,
+            # there will be one copy for each input parameter. If
+            # there are 1000 input files, this means there will be
+            # 1000 * 100000 filenames (instead of 1000 references to
+            # the same list containing 10000 strings).
             extra_params = tuple(file_names_transform.substitute(
                 filenames, p) for p in extra_specs)
 

ruffus/ruffus_utility.py

@@ -878,20 +878,29 @@ def __call__(self, p):
 #
 
 
-def regex_replace(filename, regex_str, compiled_regex, substitution_patterns, regex_or_suffix=REGEX_SUBSTITUTE):
-    return apply_func_to_sequence(substitution_patterns, t_regex_replace(filename, regex_str, compiled_regex, regex_or_suffix))
+def regex_replace(filename, regex_str, compiled_regex, substitution_patterns,
+                  regex_or_suffix=REGEX_SUBSTITUTE):
+    return apply_func_to_sequence(
+        substitution_patterns,
+        t_regex_replace(filename, regex_str, compiled_regex, regex_or_suffix))
 
 
 def formatter_replace(filenames, regex_str, compiled_regex, substitution_patterns):
-    return apply_func_to_sequence(substitution_patterns, t_formatter_replace(filenames, regex_str, compiled_regex))
+    return apply_func_to_sequence(
+        substitution_patterns,
+        t_formatter_replace(filenames, regex_str, compiled_regex))
 
 
 def nested_formatter_replace(filenames, regex_strings, compiled_regexes, substitution_patterns):
-    return apply_func_to_sequence(substitution_patterns, t_nested_formatter_replace(filenames, regex_strings, compiled_regexes))
+    return apply_func_to_sequence(
+        substitution_patterns,
+        t_nested_formatter_replace(filenames, regex_strings, compiled_regexes))
 
 
 def nested_string_replace(prev_str, new_str, substitution_patterns):
-    return apply_func_to_sequence(substitution_patterns, t_nested_string_replace(prev_str, new_str))
+    return apply_func_to_sequence(
+        substitution_patterns,
+        t_nested_string_replace(prev_str, new_str))
 
 
 # _________________________________________________________________________________________

ruffus/task.py

@@ -456,7 +456,8 @@ def generic_job_descriptor(unglobbed_params, verbose_abbreviated_path, runtime_d
 
 
 def io_files_job_descriptor(unglobbed_params, verbose_abbreviated_path, runtime_data):
-    extra_param = ", " + shorten_filenames_encoder(unglobbed_params[2:], verbose_abbreviated_path)[1:-1] \
+    extra_param = ", " + shorten_filenames_encoder(map(str, unglobbed_params[2:]),
+                                                   verbose_abbreviated_path)[1:-1] \
         if len(unglobbed_params) > 2 else ""
     out_param = shorten_filenames_encoder(unglobbed_params[1], verbose_abbreviated_path) \
         if len(unglobbed_params) > 1 else "??"
@@ -873,10 +874,11 @@ def _create_task(self, task_func, task_name=None):
                     if task_name not in self.task_names:
                         break
                 else:
-                    raise ruffus_exceptions.error_duplicate_task_name("The task string '%s' is ambiguous for "
-                                                    "Pipeline '%s'. You must disambiguate "
-                                                    "explicitly with different task names "
-                                                    % (task_str, self.name))
+                    raise ruffus_exceptions.error_duplicate_task_name(
+                        "The task string '{}' is ambiguous for "
+                        "Pipeline '{}'. You must disambiguate "
+                        "explicitly with different task names ".format(
+                            task_str, self.name))
             return Task(task_func, task_name, self)
 
         #
@@ -2342,7 +2344,7 @@ def _is_up_to_date(self, verbose_logger_job_history):
             #
             # rethrow exception after adding task name
             #
-            if exceptionType == error_task:
+            if exceptionType == ruffus_exceptions.error_task:
                 exceptionValue.specify
                 inst.specify_task(self, "Exceptions in dependency checking")
                 raise
@@ -5028,7 +5030,10 @@ def pipeline_run(target_tasks=[],
             import gevent.event
             import gevent.queue
             import gevent.pool
-            import gevent.signal
+            try:
+                from gevent.signal import signal as gevent_signal
+            except:
+                import gevent.signal as gevent_signal
             try:
                 import gevent.lock as gevent_lock
             except:
@@ -5038,9 +5043,9 @@ def pipeline_run(target_tasks=[],
             pool_t = gevent.pool.Pool
             pool = pool_t(parallelism)
             queue_t = gevent.queue.Queue
-            gevent.signal(signal.SIGINT, functools.partial(handle_sigint, pool=pool, pipeline=pipeline))
-            gevent.signal(signal.SIGUSR1, functools.partial(handle_sigusr1, pool=pool, pipeline=pipeline))
-            gevent.signal(signal.SIGUSR2, functools.partial(handle_sigusr2, pool=pool, pipeline=pipeline))
+            gevent_signal(signal.SIGINT, functools.partial(handle_sigint, pool=pool, pipeline=pipeline))
+            gevent_signal(signal.SIGUSR1, functools.partial(handle_sigusr1, pool=pool, pipeline=pipeline))
+            gevent_signal(signal.SIGUSR2, functools.partial(handle_sigusr2, pool=pool, pipeline=pipeline))
         else:
             raise ValueError("unknown pool manager '{}'".format(pool_manager))
 

ruffus/test/Makefile

@@ -17,7 +17,7 @@ NOTEST_RUN=$(NOTEST:%.py=%.run)
 all: $(PYTEST_RUN) $(NOTEST_RUN)
 
 %.pytest: %.py
-	pytest $(PYTEST_OPTIONS) $<
+	/usr/bin/time --output=$*.time -v pytest $(PYTEST_OPTIONS) $<
 
 %.run: %.py
-	python $<
+	/usr/bin/time --output=$*.time -v python $<