Repository: databiosphere/toil
Issue: PromisedRequirement doesn't work with static DAGs defined in job functions

```
#!/usr/bin/env python2.7
'''
Author : Arjun Arkal Rao
Affiliation : UCSC BME, UCSC Genomics Institute
File : toy_code_for_unfulfilled_promise_error.py
'''
from __future__ import print_function
from toil.job import Job, PromisedRequirement

import argparse


def test_test():
    '''
    '''
    parser = argparse.ArgumentParser()
    parser.add_argument('--dummy', dest='dummy', default=None, required=False)
    Job.Runner.addToilOptions(parser)
    params = parser.parse_args()

    F1 = Job.wrapJobFn(returner, disk='100M')
    F2 = Job.wrapJobFn(caller, F1.rv(), disk='200M')
    F1.addChild(F2)
    Job.Runner.startToil(F1, params)


def returner(job):
    job.fileStore.logToMaster(str(job.effectiveRequirements(job.fileStore.jobStore.config).disk))
    return 10


def caller(job, x):
    G = job.wrapJobFn(returner, disk=PromisedRequirement(oop, x))
    H = job.wrapJobFn(returner, disk=PromisedRequirement(lambda x: x, G.rv()))
    job.addChild(G)
    G.addChild(H)


def oop(x):
    return x + 1000


if __name__ == '__main__':
    test_test()
```

Gives an unpickling error

```
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,817 INFO: toil.lib.bioio: Logging set at level: INFO
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,817 INFO: toil.lib.bioio: Logging set at level: INFO
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,822 INFO: toil.jobStores.fileJobStore: Jobstore directory is: /tmp/oppo
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,823 INFO: toil.jobStores.abstractJobStore: The workflow ID is: '6bd05e7c-68b1-4480-9d6e-63a3d8620fb1'
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,834 INFO: toil.common: Using the single machine batch system
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,834 WARNING: toil.batchSystems.singleMachine: Limiting maxCores to CPU count of system (4).
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,834 WARNING: toil.batchSystems.singleMachine: Limiting maxMemory to physically available memory (8589934592).
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,834 INFO: toil.batchSystems.singleMachine: Setting up the thread pool with 40 workers, given a minimum CPU fraction of 0.100000 and a maximum CPU value of 4.
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,841 INFO: toil.common: Written the environment for the jobs to the environment file
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,841 INFO: toil.common: Caching all jobs in job store
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,841 INFO: toil.common: 0 jobs downloaded.
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,846 INFO: toil.realtimeLogger: Real-time logging disabled
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,846 INFO: toil.leader: (Re)building internal scheduler state
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,847 INFO: toil.leader: Checked batch system has no running jobs and no updated jobs
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,847 INFO: toil.leader: Found 1 jobs to start and 0 jobs with successors to run
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,850 INFO: toil.leader: Starting the main loop
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:54,852 INFO: toil.batchSystems.singleMachine: Executing command: '/Users/arjun/cached_toil/toil/venv/bin/_toil_worker /tmp/oppo I/G/job79W4uy'.
INFO:toil.common:Created the workflow directory at /tmp/toil-6bd05e7c-68b1-4480-9d6e-63a3d8620fb1
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,189 INFO: toil.batchSystems.singleMachine: Executing command: '/Users/arjun/cached_toil/toil/venv/bin/_toil_worker /tmp/oppo s/p/jobYpr2z2'.
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,355 INFO: toil.leader: Got message from job at time 07-03-2016 00:49:55: 104857600
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,415 INFO: toil.batchSystems.singleMachine: Executing command: '/Users/arjun/cached_toil/toil/venv/bin/_toil_worker /tmp/oppo 3/L/jobp7fI6l'.
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,417 INFO: toil.batchSystems.singleMachine: Executing command: '/Users/arjun/cached_toil/toil/venv/bin/_toil_worker /tmp/oppo o/Y/jobtB78cQ'.
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,649 WARNING: toil.leader: The jobWrapper seems to have left a log file, indicating failure: o/Y/jobtB78cQ
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,649 WARNING: toil.leader: Reporting file: o/Y/jobtB78cQ
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,649 WARNING: toil.leader: o/Y/jobtB78cQ:  ---TOIL WORKER OUTPUT LOG---
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,649 WARNING: toil.leader: o/Y/jobtB78cQ:  WARNING:toil.resource:The localize() method should only be invoked on a worker.
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,649 WARNING: toil.leader: o/Y/jobtB78cQ:  WARNING:toil.resource:Can't find resource for leader path '/Users/arjun/LUSC_scaling'
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,649 WARNING: toil.leader: o/Y/jobtB78cQ:  WARNING:toil.resource:Can't localize module ModuleDescriptor(dirPath='/Users/arjun/LUSC_scaling', name='pootest')
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:  INFO:toil.jobStores.fileJobStore:Jobstore directory is: /tmp/oppo
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:  Traceback (most recent call last):
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:    File "/Users/arjun/cached_toil/toil/src/toil/worker.py", line 318, in main
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:      job = Job._loadJob(jobWrapper.command, jobStore)
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:    File "/Users/arjun/cached_toil/toil/src/toil/job.py", line 1911, in _loadJob
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:      return cls._unpickle(userModule, fileHandle)
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:    File "/Users/arjun/cached_toil/toil/src/toil/job.py", line 1934, in _unpickle
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:      return unpickler.load()
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:  EOFError
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:  Exiting the worker because of a failed jobWrapper on host Arjuns-MacBook-Pro.local
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,650 WARNING: toil.leader: o/Y/jobtB78cQ:  ERROR:toil.worker:Exiting the worker because of a failed jobWrapper on host Arjuns-MacBook-Pro.local
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,651 WARNING: toil.leader: o/Y/jobtB78cQ:  WARNING:toil.jobWrapper:Due to failure we are reducing the remaining retry count of job o/Y/jobtB78cQ to 0
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,651 WARNING: toil.leader: Job: o/Y/jobtB78cQ is completely failed
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,858 INFO: toil.leader: Got message from job at time 07-03-2016 00:49:55: 1010
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:55,859 INFO: toil.leader: Got message from job at time 07-03-2016 00:49:55: Added file with ID 's/N/tmpVn48W4.tmp' to the list of files to be globally deleted.
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:57,654 INFO: toil.leader: No jobs left to run so exiting.
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:57,655 INFO: toil.leader: Finished the main loop
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:57,655 INFO: toil.leader: Waiting for stats and logging collator process to finish ...
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:57,870 INFO: toil.leader: ... finished collating stats and logs. Took 0.214813947678 seconds
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:57,870 INFO: toil.leader: Waiting for service manager thread to finish ...
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:58,857 INFO: toil.leader: ... finished shutting down the service manager. Took 0.986219882965 seconds
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:58,857 INFO: toil.leader: Finished toil run with 3 failed jobs
Arjuns-MacBook-Pro.local: 2016-07-03 00:49:58,857 INFO: toil.leader: Failed jobs at end of the run: set(['s/p/jobYpr2z2', 'o/Y/jobtB78cQ', 'I/G/job79W4uy'])
Traceback (most recent call last):
  File "/Users/arjun/LUSC_scaling/pootest.py", line 44, in <module>
    test_test()
  File "/Users/arjun/LUSC_scaling/pootest.py", line 24, in test_test
    Job.Runner.startToil(F1, params)
  File "/Users/arjun/cached_toil/toil/src/toil/job.py", line 515, in startToil
    return toil.start(job)
  File "/Users/arjun/cached_toil/toil/src/toil/common.py", line 546, in start
    return self._runMainLoop(job)
  File "/Users/arjun/cached_toil/toil/src/toil/common.py", line 756, in _runMainLoop
    jobCache=self._jobCache)
  File "/Users/arjun/cached_toil/toil/src/toil/leader.py", line 694, in mainLoop
    raise FailedJobsException( config.jobStore, len(toilState.totalFailedJobs) )
toil.leader.FailedJobsException: The job store '/tmp/oppo' contains 3 failed jobs
```

On the plus side, functions can be passed to PromisedRequirement, the second call (function H) fails whether its a lambda or a function.

@jpfeil @hannes-ucsc can you look at this?

*As always pay no attention to the names of my files lol.


Changed files:
src/toil/test/src/promisedRequirementTest.py, src/toil/job.py

diff --git a/src/toil/job.py b/src/toil/job.py
index c615ce70..7e714a05 100644
--- a/src/toil/job.py
+++ b/src/toil/job.py
@@ -54,6 +54,7 @@ from toil.resource import ModuleDescriptor
 logger = logging.getLogger( __name__ )
 
 
+
 class Job(object):
     """
     Class represents a unit of work in toil.
@@ -208,7 +209,7 @@ class Job(object):
         :rtype: toil.job.FunctionWrappingJob
         """
         if PromisedRequirement.convertPromises(kwargs):
-            return self.addChild(PromisedRequirementFunctionWrappingJob(fn, *args, **kwargs))
+            return self.addChild(PromisedRequirementFunctionWrappingJob.create(fn, *args, **kwargs))
         else:
             return self.addChild(FunctionWrappingJob(fn, *args, **kwargs))
 
@@ -223,7 +224,7 @@ class Job(object):
         :rtype: toil.job.FunctionWrappingJob
         """
         if PromisedRequirement.convertPromises(kwargs):
-            return self.addFollowOn(PromisedRequirementFunctionWrappingJob(fn, *args, **kwargs))
+            return self.addFollowOn(PromisedRequirementFunctionWrappingJob.create(fn, *args, **kwargs))
         else:
             return self.addFollowOn(FunctionWrappingJob(fn, *args, **kwargs))
 
@@ -239,7 +240,7 @@ class Job(object):
         :rtype: toil.job.JobFunctionWrappingJob
         """
         if PromisedRequirement.convertPromises(kwargs):
-            return self.addChild(PromisedRequirementJobFunctionWrappingJob(fn, *args, **kwargs))
+            return self.addChild(PromisedRequirementJobFunctionWrappingJob.create(fn, *args, **kwargs))
         else:
             return self.addChild(JobFunctionWrappingJob(fn, *args, **kwargs))
 
@@ -255,7 +256,7 @@ class Job(object):
         :rtype: toil.job.JobFunctionWrappingJob
         """
         if PromisedRequirement.convertPromises(kwargs):
-            return self.addFollowOn(PromisedRequirementJobFunctionWrappingJob(fn, *args, **kwargs))
+            return self.addFollowOn(PromisedRequirementJobFunctionWrappingJob.create(fn, *args, **kwargs))
         else:
             return self.addFollowOn(JobFunctionWrappingJob(fn, *args, **kwargs))
 
@@ -272,7 +273,7 @@ class Job(object):
         :rtype: toil.job.FunctionWrappingJob
         """
         if PromisedRequirement.convertPromises(kwargs):
-            return PromisedRequirementFunctionWrappingJob(fn, *args, **kwargs)
+            return PromisedRequirementFunctionWrappingJob.create(fn, *args, **kwargs)
         else:
             return FunctionWrappingJob(fn, *args, **kwargs)
 
@@ -289,7 +290,7 @@ class Job(object):
         :rtype: toil.job.JobFunctionWrappingJob
         """
         if PromisedRequirement.convertPromises(kwargs):
-            return PromisedRequirementJobFunctionWrappingJob(fn, *args, **kwargs)
+            return PromisedRequirementJobFunctionWrappingJob.create(fn, *args, **kwargs)
         else:
             return JobFunctionWrappingJob(fn, *args, **kwargs)
 
@@ -2450,35 +2451,40 @@ class PromisedRequirementFunctionWrappingJob(FunctionWrappingJob):
     """
     Handles dynamic resource allocation using :class:`toil.job.Promise` instances.
     Spawns child function using parent function parameters and fulfilled promised
-    resource requirements. See :class:`toil.job.FunctionWrappingJob` class
+    resource requirements.
     """
     def __init__(self, userFunction, *args, **kwargs):
-        self.promisedRequirements = {}
-        requirements = {'disk': '1M', 'memory': '32M', 'cores': 0.1}
-        # Replace PromisedRequirements in intermediate job with small
-        # resource requirements.
-        for name, value in requirements.items():
-            try:
-                if isinstance(kwargs[name], PromisedRequirement):
-                    self.promisedRequirements[name] = kwargs[name]
-                    kwargs[name] = value
-            except KeyError:
-                pass
+        self._promisedKwargs = kwargs.copy()
+        # Replace resource requirements in intermediate job with small values.
+        kwargs.update(dict(disk='1M', memory='32M', cores=0.1))
         super(PromisedRequirementFunctionWrappingJob, self).__init__(userFunction, *args, **kwargs)
 
+    @classmethod
+    def create(cls, userFunction, *args, **kwargs):
+        """
+        Creates an encapsulated Toil job function with unfulfilled promised resource
+        requirements. After the promises are fulfilled, a child job function is created
+        using updated resource values. The subgraph is encapsulated to ensure that this
+        child job function is run before other children in the workflow. Otherwise, a
+        different child may try to use an unresolved promise return value from the parent.
+        """
+        return EncapsulatedJob(cls(userFunction, *args, **kwargs))
+
     def run(self, fileStore):
         # Assumes promises are fulfilled when parent job is run
         self.evaluatePromisedRequirements()
         userFunction = self._getUserFunction()
-        return self.addChildFn(userFunction, *self._args, **self._kwargs).rv()
+        return self.addChildFn(userFunction, *self._args, **self._promisedKwargs).rv()
 
     def evaluatePromisedRequirements(self):
         requirements = ["disk", "memory", "cores"]
+        # Fulfill resource requirement promises
         for requirement in requirements:
             try:
-                self._kwargs[requirement] = self.promisedRequirements[requirement].getValue()
+                if isinstance(self._promisedKwargs[requirement], PromisedRequirement):
+                    self._promisedKwargs[requirement] = self._promisedKwargs[requirement].getValue()
             except KeyError:
-                self._kwargs[requirement] = getattr(self, requirement)
+                pass
 
 
 class PromisedRequirementJobFunctionWrappingJob(PromisedRequirementFunctionWrappingJob):
@@ -2490,7 +2496,7 @@ class PromisedRequirementJobFunctionWrappingJob(PromisedRequirementFunctionWrapp
     def run(self, fileStore):
         self.evaluatePromisedRequirements()
         userFunction = self._getUserFunction()
-        return self.addChildJobFn(userFunction, *self._args, **self._kwargs).rv()
+        return self.addChildJobFn(userFunction, *self._args, **self._promisedKwargs).rv()
 
 
 class EncapsulatedJob(Job):
@@ -2514,10 +2520,12 @@ class EncapsulatedJob(Job):
         """
         :param toil.job.Job job: the job to encapsulate.
         """
-        Job.__init__(self)
+        # Giving the root of the subgraph the same resources as the first job in the subgraph.
+        Job.__init__(self, disk=job.disk, memory=job.memory, cores=job.cores)
         self.encapsulatedJob = job
         Job.addChild(self, job)
-        self.encapsulatedFollowOn = Job()
+        # Use small resource requirements for dummy Job instance.
+        self.encapsulatedFollowOn = Job(disk='1M', memory='32M', cores=0.1)
         Job.addFollowOn(self, self.encapsulatedFollowOn)
 
     def addChild(self, childJob):
diff --git a/src/toil/test/src/promisedRequirementTest.py b/src/toil/test/src/promisedRequirementTest.py
index 530bea08..56ffab6c 100644
--- a/src/toil/test/src/promisedRequirementTest.py
+++ b/src/toil/test/src/promisedRequirementTest.py
@@ -15,6 +15,7 @@
 from __future__ import absolute_import, print_function
 import os
 import logging
+import time
 import toil.test.batchSystems.batchSystemTest as batchSystemTest
 
 from toil.job import Job
@@ -37,7 +38,7 @@ class hidden:
         An abstract base class for testing Toil workflows with promised requirements.
         """
 
-        def testPromisedRequirementDynamic(self):
+        def testConcurrencyDynamic(self):
             """
             Asserts that promised core resources are allocated properly using a dynamic Toil workflow
             """
@@ -51,7 +52,7 @@ class hidden:
                 maxValue = max(values)
                 self.assertEqual(maxValue, self.cpuCount / coresPerJob)
 
-        def testPromisedRequirementStatic(self):
+        def testConcurrencyStatic(self):
             """
             Asserts that promised core resources are allocated properly using a static DAG
             """
@@ -98,7 +99,6 @@ class hidden:
             assert (minValue, maxValue) == (0, 0)
             return counterPath
 
-        # Disable the concurrency test in this case
 
         def testJobConcurrency(self):
             pass
@@ -115,14 +115,25 @@ class hidden:
             file2 = 512
             F1 = Job.wrapJobFn(_writer, file1)
             F2 = Job.wrapJobFn(_writer, file2)
-            G = Job.wrapJobFn(_follower, file1+file2,
+            G = Job.wrapJobFn(_follower, file1 + file2,
                               disk=PromisedRequirement(lambda x, y: x.size + y.size,
                                                        F1.rv(), F2.rv()))
             F1.addChild(F2)
             F2.addChild(G)
+
             Job.Runner.startToil(F1, self.getOptions(self._createTempDir('testFiles')))
 
 
+        def testPromiseRequirementRaceStatic(self):
+            """
+            Checks for a race condition when using promised requirements and child job functions.
+            """
+            A = Job.wrapJobFn(logDiskUsage, 'A', sleep=5, disk=PromisedRequirement(1024))
+            B = Job.wrapJobFn(logDiskUsage, 'B', disk=PromisedRequirement(lambda x: x + 1024, A.rv()))
+            A.addChild(B)
+            Job.Runner.startToil(A, self.getOptions(self._createTempDir('testFiles')))
+
+
 def _writer(job, fileSize):
     '''
     Write a local file and return the FileID obtained from running writeGlobalFile on
@@ -149,9 +160,6 @@ def _follower(job, expectedDisk):
     assert job.effectiveRequirements(job.fileStore.jobStore.config).disk == expectedDisk
 
 
-
-
-
 def maxConcurrency(job, cpuCount, filename, coresPerJob):
     """
     Returns the max number of concurrent tasks when using a PromisedRequirement instance
@@ -183,6 +191,18 @@ def getThirtyTwoMb():
     return '32M'
 
 
+def logDiskUsage(job, funcName, sleep=0):
+    """
+    Logs the job's disk usage to master and sleeps for specified amount of time.
+
+    :return: job function's disk usage
+    """
+    diskUsage = job.effectiveRequirements(job.fileStore.jobStore.config).disk
+    job.fileStore.logToMaster('{}: {}'.format(funcName, diskUsage))
+    time.sleep(sleep)
+    return diskUsage
+
+
 class SingleMachinePromisedRequirementsTest(hidden.AbstractPromisedRequirementsTest):
     """
     Tests against the SingleMachine batch system
@@ -207,3 +227,4 @@ class MesosPromisedRequirementsTest(hidden.AbstractPromisedRequirementsTest, Mes
 
     def tearDown(self):
         self._stopMesos()
+