Repository: databiosphere/toil
Issue: Harbingerfile deletion race condition

```
l/k/jobIi7VIc:  Traceback (most recent call last):
l/k/jobIi7VIc:    File "/home/alden/progressiveCactus3/python/local/lib/python2.7/site-packages/toil/worker.py", line 332, in main
l/k/jobIi7VIc:      job._runner(jobWrapper=jobWrapper, jobStore=jobStore, fileStore=fileStore)
l/k/jobIi7VIc:    File "/home/alden/progressiveCactus3/python/local/lib/python2.7/site-packages/toil/job.py", line 2323, in _runner
l/k/jobIi7VIc:      returnValues = self._run(jobWrapper, fileStore)
l/k/jobIi7VIc:    File "/home/alden/progressiveCactus3/python/local/lib/python2.7/site-packages/toil/job.py", line 2271, in _run
l/k/jobIi7VIc:      return self.run(fileStore)
l/k/jobIi7VIc:    File "/home/alden/progressiveCactus3/submodules/cactus/pipeline/cactus_workflow.py", line 900, in run
l/k/jobIi7VIc:      self.cactusSequencesPath = fileStore.readGlobalFile(self.cactusSequencesID)
l/k/jobIi7VIc:    File "/home/alden/progressiveCactus3/python/local/lib/python2.7/site-packages/toil/job.py", line 816, in readGlobalFile
l/k/jobIi7VIc:      harbingerFile.waitOnDownload(lockFileHandle)
l/k/jobIi7VIc:    File "/home/alden/progressiveCactus3/python/local/lib/python2.7/site-packages/toil/job.py", line 1603, in waitOnDownload
l/k/jobIi7VIc:      pid = self.read()
l/k/jobIi7VIc:    File "/home/alden/progressiveCactus3/python/local/lib/python2.7/site-packages/toil/job.py", line 1616, in read
l/k/jobIi7VIc:      return int(open(self.harbingerFileName).read())
l/k/jobIi7VIc:  IOError: [Errno 2] No such file or directory: '/tmp/toil-269fb642-c572-42a1-8c9f-aa751fa3e5e9/cache-269fb642-c572-42a1-8c9f-aa751fa3e5e9/.di9QL3RtcHo0d3RtVC50bXA=.harbinger'
```

I haven't fully investigated this yet, but it looks like the harbinger file can be deleted by the asyncWrite thread after self.exists() is checked but before self.read() is called in waitOnDownload(). Maybe HarbingerFile.delete() should acquire the cache lock? Also, should line https://github.com/BD2KGenomics/toil/blob/edf86b5e253e9003a441a68773e79ecaded1a599/src/toil/job.py#L1613 be self.delete() instead of harbingerFile.delete()?


Changed files:
src/toil/job.py

diff --git a/src/toil/job.py b/src/toil/job.py
index 59b9c164..ea634647 100644
--- a/src/toil/job.py
+++ b/src/toil/job.py
@@ -852,8 +852,7 @@ class Job(object):
                                 # We don't need to return the file size here because addToCache
                                 # already does it for us
                         finally:
-                            # Reacquire the file lock and delete the partial file.
-                            flock(lockFileHandle, LOCK_EX)
+                            # In any case, delete the harbinger file.
                             harbingerFile.delete()
                     else:
                         # Release the cache lock since the remaining stuff is not cache related.
@@ -1569,8 +1568,6 @@ class Job(object):
                 :param class fileStore: The 'self' object of the fileStore class
                 :param str fileStoreID: The file store ID for an input file
                 :param str cachedFileName: The cache file name corresponding to a given file
-                :return: Harbinger file name
-                :rtype: str
                 """
                 # We need either a file store ID, or a cached file name, but not both (XOR).
                 assert (fileStoreID is None) != (cachedFileName is None)
@@ -1579,9 +1576,12 @@ class Job(object):
                     cachedFileName = fileStore.encodedFileID(fileStoreID)
                 else:
                     self.fileStoreID = fileStore.decodedFileID(cachedFileName)
+                self.fileStore = fileStore
                 self.harbingerFileName = '/.'.join(os.path.split(cachedFileName)) + '.harbinger'
 
             def write(self):
+                self.fileStore.logToMaster('CACHE: Creating a harbinger file for (%s). '
+                                           % self.fileStoreID, logging.DEBUG)
                 with open(self.harbingerFileName + '.tmp', 'w') as harbingerFile:
                     harbingerFile.write(str(os.getpid()))
                 # Make this File read only to prevent overwrites
@@ -1611,7 +1611,7 @@ class Job(object):
                     else:
                         # The process that was supposed to download the file has died so we need
                         # to remove the harbinger.
-                        harbingerFile.delete()
+                        self._delete()
 
             def read(self):
                 return int(open(self.harbingerFileName).read())
@@ -1620,6 +1620,19 @@ class Job(object):
                 return os.path.exists(self.harbingerFileName)
 
             def delete(self):
+                """
+                Acquires the cache lock then attempts to delete the harbinger file.
+                """
+                with self.fileStore.cacheLock():
+                    self._delete()
+
+            def _delete(self):
+                """
+                This function assumes you already have the cache lock!
+                """
+                assert self.exists()
+                self.fileStore.logToMaster('CACHE: Deleting the harbinger file for (%s)' %
+                                           self.fileStoreID, logging.DEBUG)
                 os.remove(self.harbingerFileName)
 
             @staticmethod