Repository: databiosphere/toil
Issue: Caching logic will fault if file_size > cacheSize

Lines 680-686 in _cleanLocalTempDir 

```
            while totalCachedFileSizes > cacheSize:
                fileSize, fileStoreID =  cachedFileSizes.pop()
                filePath = self.jobStoreFileIDToCacheLocation[fileStoreID]
                self.jobStoreFileIDToCacheLocation.pop(fileStoreID)
                os.remove(filePath)
                totalCachedFileSizes -= fileSize
                assert totalCachedFileSizes >= 0
```

show that if totalCachedFileSizes > cacheSize the filePath is removed.

If a file larger than cacheSize is created, it is added to jobStoreFileIDToCacheLocation and it starts getting `os.link`ed however the _cleanLocalTempDir will delete the file before the link is complete. If the first thing in the next job is to retrieve this file from filestore ( which it probably will be), it will attempt to get it from filestore since it is NOT in cache.

However, depending on the file size, I/O, etc, if the file is still being written asynchronously to filestore, readGlobalFile starts processing an incomplete file. 

Either _cleanLocalTempDir needs to wat for the write to complete before deleting the file (since it is the rate limiting step), of readGlobalFile should have a way of asserting the integrity of the input file before reading it (Do not read a file till it is fully in filestore)


Changed files:
src/toil/worker.py, src/toil/test/src/jobFileStoreTest.py, src/toil/job.py

diff --git a/src/toil/job.py b/src/toil/job.py
index 3b7edbf1..80177523 100644
--- a/src/toil/job.py
+++ b/src/toil/job.py
@@ -28,7 +28,7 @@ import cPickle
 import logging
 import shutil
 import stat
-from threading import Thread, Semaphore
+from threading import Thread, Semaphore, Event
 from Queue import Queue, Empty
 from bd2k.util.humanize import human2bytes
 from io import BytesIO
@@ -366,8 +366,15 @@ class Job(object):
         Class used to manage temporary files and log messages, 
         passed as argument to the Job.run method.
         """
-        def __init__(self, jobStore, jobWrapper, localTempDir, 
-                     inputBlockFn, jobStoreFileIDToCacheLocation, terminateEvent):
+        #Variables used for synching reads/writes
+        lockFilesLock = Semaphore()
+        lockFiles = set()
+        #For files in jobStore that are on the local disk, 
+        #map of jobStoreFileIDs to locations in localTempDir.
+        jobStoreFileIDToCacheLocation = {}
+        terminateEvent = Event() #Used to signify crashes in threads
+        
+        def __init__(self, jobStore, jobWrapper, localTempDir, inputBlockFn):
             """
             This constructor should not be called by the user, 
             FileStore instances are only provided as arguments 
@@ -383,7 +390,6 @@ class Job(object):
             self.workerNumber = 2
             self.queue = Queue()
             self.updateSemaphore = Semaphore() 
-            self.terminateEvent = terminateEvent
             #Function to write files to job store
             def asyncWrite():
                 try:
@@ -394,7 +400,7 @@ class Job(object):
                         except Empty:
                             #Check if termination event is signaled 
                             #(set in the event of an exception in the worker)
-                            if terminateEvent.isSet():
+                            if self.terminateEvent.isSet():
                                 raise RuntimeError("The termination flag is set, exiting")
                             continue
                         #Normal termination condition is getting None from queue
@@ -412,8 +418,11 @@ class Job(object):
                                     break
                                 outputFileHandle.write(copyBuffer)
                         inputFileHandle.close()
+                        #Remove the file from the lock files
+                        with self.lockFilesLock:
+                            self.lockFiles.remove(jobStoreFileID)
                 except:
-                    terminateEvent.set()
+                    self.terminateEvent.set()
                     raise
                     
             self.workers = map(lambda i : Thread(target=asyncWrite), 
@@ -421,11 +430,6 @@ class Job(object):
             for worker in self.workers:
                 worker.start()
             self.inputBlockFn = inputBlockFn
-            #Caching 
-            #
-            #For files in jobStore that are on the local disk, 
-            #map of jobStoreFileIDs to locations in localTempDir.
-            self.jobStoreFileIDToCacheLocation = jobStoreFileIDToCacheLocation
 
         def getLocalTempDir(self):
             """
@@ -471,6 +475,8 @@ class Job(object):
                 self.queue.put((open(absLocalFileName, 'r'), jobStoreFileID))
                 #Chmod to make file read only to try to prevent accidental user modification
                 os.chmod(absLocalFileName, stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH)
+                with self.lockFilesLock:
+                    self.lockFiles.add(jobStoreFileID)
                 self.jobStoreFileIDToCacheLocation[jobStoreFileID] = absLocalFileName
             else:
                 #Write the file directly to the file store
@@ -668,8 +674,10 @@ class Job(object):
             #Remove files so that the total cached files are smaller than a cacheSize
             
             #List of pairs of (fileCreateTime, fileStoreID) for cached files
+            with self.lockFilesLock:
+                deletableCacheFiles = set(self.jobStoreFileIDToCacheLocation.keys()) - self.lockFiles
             cachedFileCreateTimes = map(lambda x : (os.stat(self.jobStoreFileIDToCacheLocation[x]).st_ctime, x),
-                                        self.jobStoreFileIDToCacheLocation.keys())
+                                        deletableCacheFiles)
             #Total number of bytes stored in cached files
             totalCachedFileSizes = sum([os.stat(self.jobStoreFileIDToCacheLocation[x]).st_size for x in
                                         self.jobStoreFileIDToCacheLocation.keys()])
@@ -678,8 +686,8 @@ class Job(object):
             cachedFileCreateTimes.sort()
             cachedFileCreateTimes.reverse()
             #Now do the actual file removal
-            while totalCachedFileSizes > cacheSize:
-                fileCreateTime, fileStoreID =  cachedFileCreateTimes.pop()
+            while totalCachedFileSizes > cacheSize and len(cachedFileCreateTimes) > 0:
+                fileCreateTime, fileStoreID = cachedFileCreateTimes.pop()
                 fileSize = os.stat(self.jobStoreFileIDToCacheLocation[fileStoreID]).st_size
                 filePath = self.jobStoreFileIDToCacheLocation[fileStoreID]
                 self.jobStoreFileIDToCacheLocation.pop(fileStoreID)
diff --git a/src/toil/test/src/jobFileStoreTest.py b/src/toil/test/src/jobFileStoreTest.py
index 43e638d7..2c6abb98 100644
--- a/src/toil/test/src/jobFileStoreTest.py
+++ b/src/toil/test/src/jobFileStoreTest.py
@@ -13,7 +13,7 @@
 # limitations under the License.
 from __future__ import absolute_import
 import random
-
+import os
 from toil.job import Job
 from toil.test import ToilTest
 
@@ -23,81 +23,110 @@ class JobFileStoreTest(ToilTest):
     """
     Tests testing the Job.FileStore class
     """
-    def testJobFileStore(self):
+    def testJobFileStoreWithSmallCache(self, retryCount=0, badWorker=0.0, 
+                         stringNo=1, stringLength=1000000, cacheSize=10000, testNo=2):
         """
         Creates a chain of jobs, each reading and writing files using the 
         Job.FileStore interface. Verifies the files written are always what we expect.
         The chain tests the caching behavior. 
         """
-        for test in xrange(10):
+        for test in xrange(testNo):
             #Make a list of random strings, each of 100k chars and hash the first 200 
             #base prefix to the string
             def randomString():
                 chars = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
-                s = "".join(map(lambda i : random.choice(chars), xrange(100000)))
+                s = "".join(map(lambda i : random.choice(chars), xrange(stringLength)))
                 return s[:PREFIX_LENGTH], s
             #Total length is 2 million characters (20 strings of length 100K each) 
-            testStrings = dict(map(lambda i : randomString(), xrange(20)))
+            testStrings = dict(map(lambda i : randomString(), xrange(stringNo)))
             options = Job.Runner.getDefaultOptions(self._getTestJobStorePath())
             options.logLevel = "INFO"
-            options.cacheSize = 1000000
-            options.retryCount=100
-            options.badWorker=0.5
+            options.cacheSize = cacheSize
+            options.retryCount=retryCount
+            options.badWorker=badWorker
             options.badWorkerFailInterval = 1.0
             chainLength = 10
             # Run the workflow, the return value being the number of failed jobs
             Job.Runner.startToil(Job.wrapJobFn(fileTestJob, [], 
                                                testStrings, chainLength), 
                                  options)
+            
+    def testJobFileStoreWithOverLargeCache(self):
+        """
+        Tests case that all files are cached.
+        """
+        self.testJobFileStoreWithSmallCache(retryCount=0, badWorker=0.0, 
+                         stringNo=5, stringLength=1000000, 
+                         cacheSize=10000000)
+        
+    def testJobFileStoreWithMediumCache(self):
+        """
+        Tests case that about half the files are cached
+        """
+        self.testJobFileStoreWithSmallCache(retryCount=0, badWorker=0.0, 
+                         stringNo=5, stringLength=1000000, 
+                         cacheSize=3000000)
+    
+    def testJobFileStoreWithMediumCacheAndBadWorker(self):
+        """
+        Tests case that about half the files are cached and the worker is randomly
+        failing.
+        """
+        self.testJobFileStoreWithSmallCache(retryCount=100, badWorker=0.5, 
+                         stringNo=5, stringLength=1000000, 
+                         cacheSize=3000000)
         
 def fileTestJob(job, inputFileStoreIDs, testStrings, chainLength):
     """
     Test job exercises Job.FileStore functions
     """
+    outputFileStoreIds = [] #Strings passed to the next job in the chain
+    
+    #Load the input jobStoreFileIDs and check that they map to the 
+    #same set of random input strings, exercising the different functions in the fileStore interface
     for fileStoreID in inputFileStoreIDs:
-        #Load the input jobStoreFileIDs and check that they map to the 
-        #same set of random input strings
-        if random.random() > 0.666:
+        if random.random() > 0.5:
             #Read the file for the fileStoreID, randomly picking a way to invoke readGlobalFile
             if random.random() > 0.5:
                 tempFile = job.fileStore.readGlobalFile(fileStoreID, 
+                                                        job.fileStore.getLocalTempFile() if 
+                                                        random.random() > 0.5 else None,
                                                         cache=random.random() > 0.5)
+                with open(tempFile, 'r') as fH:
+                    string = fH.readline()
             else:
-                userPath = job.fileStore.getLocalTempFile()
-                tempFile = job.fileStore.readGlobalFile(fileStoreID, userPath, 
-                                                        cache=random.random() > 0.5)
-                assert userPath == tempFile
-            #Check the local file is as we expect
-            with open(tempFile, 'r') as fH:
-                string = fH.readline()
-                assert testStrings[string[:PREFIX_LENGTH]] == string
-        elif random.random() > 0.666:
-            #Try reading the file using the stream method.
-            with job.fileStore.readGlobalFileStream(fileStoreID) as fH:
-                string = fH.readline()
-                assert testStrings[string[:PREFIX_LENGTH]] == string
+                #Check the local file is as we expect
+                with job.fileStore.readGlobalFileStream(fileStoreID) as fH:
+                    string = fH.readline()
+                    
+            #Check the string we get back is what we expect
+            assert testStrings[string[:PREFIX_LENGTH]] == string
+            
+            #This allows the file to be passed to the next job
+            outputFileStoreIds.append(fileStoreID)
         else:
             #This tests deletion
             job.fileStore.deleteGlobalFile(fileStoreID)
+    
+    #Fill out the output strings until we have the same number as the input strings
+    #exercising different ways of writing files to the file store
+    while len(outputFileStoreIds) < len(testStrings):
+        #Pick a string and write it into a file
+        testString = random.choice(testStrings.values())
+        if random.random() > 0.5:
+            #Make a local copy of the file
+            tempFile = job.fileStore.getLocalTempFile() if random.random() > 0.5 \\
+            else os.path.join(job.fileStore.getLocalTempDir(), "temp.txt")
+            with open(tempFile, 'w') as fH:
+                fH.write(testString)
+            #Write a local copy of the file using the local file
+            outputFileStoreIds.append(job.fileStore.writeGlobalFile(tempFile))
+        else:
+            #Use the writeGlobalFileStream method to write the file
+            with job.fileStore.writeGlobalFileStream() as (fH, fileStoreID):
+                fH.write(testString)
+                outputFileStoreIds.append(fileStoreID)
 
     if chainLength > 0:
-        #For each job write some these strings into the file store, collecting 
-        #the fileIDs
-        outputFileStoreIds = []
-        for testPrefix in testStrings:
-            if random.random() > 0.5:
-                #Make a local copy of the file
-                tempFile = job.fileStore.getLocalTempFile()
-                with open(tempFile, 'w') as fH:
-                    fH.write(testStrings[testPrefix])
-                #Write a local copy of the file using the local file
-                outputFileStoreIds.append(job.fileStore.writeGlobalFile(tempFile))
-            else:
-                with job.fileStore.writeGlobalFileStream() as (fH, fileStoreID):
-                    fH.write(testStrings[testPrefix])
-                    outputFileStoreIds.append(fileStoreID)
-    
         #Make a child that will read these files and check it gets the same results
         job.addChildJobFn(fileTestJob, outputFileStoreIds, testStrings, chainLength-1)
-        
-        
\\ No newline at end of file
diff --git a/src/toil/worker.py b/src/toil/worker.py
index 20dc3fc1..e0a69def 100644
--- a/src/toil/worker.py
+++ b/src/toil/worker.py
@@ -33,7 +33,7 @@ import logging
 import xml.etree.cElementTree as ET
 import cPickle
 import shutil
-from threading import Thread, Event
+from threading import Thread
 import signal
 
 logger = logging.getLogger( __name__ )
@@ -206,8 +206,6 @@ def main():
     messageNode = ET.SubElement(elementNode, "messages")
     messages = []
     blockFn = lambda : True
-    jobStoreFileIDToCacheLocation = {}
-    terminateEvent = Event() #This is used to signify crashes in threads
     try:
 
         #Put a message at the top of the log, just to make sure it's working.
@@ -270,8 +268,7 @@ def main():
                     
                     #Create a fileStore object for the job
                     fileStore = Job.FileStore(jobStore, jobWrapper, localTempDir, 
-                                              blockFn, jobStoreFileIDToCacheLocation, 
-                                              terminateEvent)
+                                              blockFn)
                     #Get the next block function and list that will contain any messages
                     blockFn = fileStore._blockFn
                     messages = fileStore.loggingMessages
@@ -293,7 +290,7 @@ def main():
                 #been scheduled after a failure to cleanup
                 break
             
-            if terminateEvent.isSet():
+            if Job.FileStore.terminateEvent.isSet():
                 raise RuntimeError("The termination flag is set")
             
             ##########################################
@@ -362,8 +359,7 @@ def main():
             assert jobWrapper.cores >= successorJob.cores
             
             #Build a fileStore to update the job
-            fileStore = Job.FileStore(jobStore, jobWrapper, localTempDir, blockFn, 
-                                      jobStoreFileIDToCacheLocation, terminateEvent)
+            fileStore = Job.FileStore(jobStore, jobWrapper, localTempDir, blockFn)
             
             #Update blockFn
             blockFn = fileStore._blockFn
@@ -399,7 +395,7 @@ def main():
     except: #Case that something goes wrong in worker
         traceback.print_exc()
         logger.error("Exiting the worker because of a failed jobWrapper on host %s", socket.gethostname())
-        terminateEvent.set()
+        Job.FileStore.terminateEvent.set()
     
     ##########################################
     #Wait for the asynchronous chain of writes/updates to finish
@@ -412,7 +408,7 @@ def main():
     #so safe to test if they completed okay
     ########################################## 
     
-    if terminateEvent.isSet():
+    if Job.FileStore.terminateEvent.isSet():
         jobWrapper = jobStore.load(jobStoreID)
         jobWrapper.setupJobAfterFailure(config)
         workerFailed = True