Merge pull request #839 from jmchilton/text_file_busy

Attempt to fix "Text file busy" errors (testing and production)

lib/galaxy/jobs/command_factory.py

@@ -73,9 +73,9 @@ def build_command(
 
 def __externalize_commands(job_wrapper, commands_builder, remote_command_params, script_name="tool_script.sh"):
     local_container_script = join( job_wrapper.working_directory, script_name )
-    fh = file( local_container_script, "w" )
-    fh.write( "#!%s\n%s" % (DEFAULT_SHELL, commands_builder.build()))
-    fh.close()
+    with open( local_container_script, "w" ) as f:
+        script_contents = "#!%s\n%s" % (DEFAULT_SHELL, commands_builder.build())
+        f.write( script_contents )
     chmod( local_container_script, 0755 )
 
     commands = local_container_script

lib/galaxy/jobs/runners/__init__.py

@@ -311,6 +311,18 @@ class BaseJobRunner( object ):
         options.update(**kwds)
         return job_script(**options)
 
+    def write_executable_script( self, path, contents, mode=0o755 ):
+        with open( path, 'w' ) as f:
+            f.write( contents )
+        os.chmod( path, mode )
+        try:
+            # sync file system to avoid "Text file busy" problems.
+            # These have occurred both in Docker containers and on EC2 clusters
+            # under high load.
+            subprocess.check_call(["sync"])
+        except Exception:
+            pass
+
     def _complete_terminal_job( self, ajs, **kwargs ):
         if ajs.job_wrapper.get_state() != model.Job.states.DELETED:
             self.work_queue.put( ( self.finish_job, ajs ) )

lib/galaxy/jobs/runners/cli.py

@@ -72,9 +72,7 @@ class ShellJobRunner( AsynchronousJobRunner ):
         )
 
         try:
-            fh = file(ajs.job_file, "w")
-            fh.write(script)
-            fh.close()
+            self.write_executable_script( ajs.job_file, script )
         except:
             log.exception("(%s) failure writing job script" % galaxy_id_tag )
             job_wrapper.fail("failure preparing job script", exception=True)

lib/galaxy/jobs/runners/condor.py

@@ -90,10 +90,7 @@ class CondorJobRunner( AsynchronousJobRunner ):
             slots_statement=galaxy_slots_statement,
         )
         try:
-            fh = file( executable, "w" )
-            fh.write( script )
-            fh.close()
-            os.chmod( executable, 0o750 )
+            self.write_executable_script( executable.job_file, script )
         except:
             job_wrapper.fail( "failure preparing job script", exception=True )
             log.exception( "(%s) failure preparing job script" % galaxy_id_tag )

lib/galaxy/jobs/runners/drmaa.py

@@ -147,10 +147,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ):
         # fill in the DRM's job run template
         script = self.get_job_file(job_wrapper, exit_code_path=ajs.exit_code_file)
         try:
-            fh = file( ajs.job_file, "w" )
-            fh.write( script )
-            fh.close()
-            os.chmod( ajs.job_file, 0o755 )
+            self.write_executable_script( ajs.job_file, script )
         except:
             job_wrapper.fail( "failure preparing job script", exception=True )
             log.exception( "(%s) failure writing job script" % galaxy_id_tag )

lib/galaxy/jobs/runners/local.py

@@ -68,9 +68,7 @@ class LocalJobRunner( BaseJobRunner ):
             'working_directory': job_wrapper.working_directory,
         }
         job_file_contents = self.get_job_file( job_wrapper, **job_script_props )
-        with open( job_file, 'w' ) as f:
-            f.write( job_file_contents )
-        os.chmod( job_file, 0o755 )
+        self.write_executable_script( job_file, job_file_contents )
         return job_file, exit_code_path
 
     def queue_job( self, job_wrapper ):

lib/galaxy/jobs/runners/pbs.py

@@ -296,10 +296,7 @@ class PBSJobRunner( AsynchronousJobRunner ):
         env_setup_commands = [ stage_commands ]
         script = self.get_job_file(job_wrapper, exit_code_path=ecfile, env_setup_commands=env_setup_commands)
         job_file = "%s/%s.sh" % (self.app.config.cluster_files_directory, job_wrapper.job_id)
-        fh = file(job_file, "w")
-        fh.write(script)
-        fh.close()
-
+        self.write_executable_script( job_file, script )
         # job was deleted while we were preparing it
         if job_wrapper.get_state() == model.Job.states.DELETED:
             log.debug( "Job %s deleted by user before it entered the PBS queue" % job_wrapper.job_id )