Repository: databiosphere/toil
Issue: Auto-scaling creates twice as many nodes as needed

… because non-preemptable nodes are scaled up even though all jobs are preemptable.


Changed files:
run_tests.py, src/toil/test/__init__.py, src/toil/batchSystems/mesos/batchSystem.py, src/toil/test/provisioners/cgcloud/provisionerTest.py, src/toil/job.py, src/toil/provisioners/clusterScaler.py, src/toil/common.py, src/toil/leader.py

diff --git a/run_tests.py b/run_tests.py
index 75d01ab7..89450633 100644
--- a/run_tests.py
+++ b/run_tests.py
@@ -29,7 +29,8 @@ test_suites = {
         None],
     'integration-test': [
         'CGCloudRNASeqTest',
-        'CGCloudRestartTest'
+        'CGCloudRestartTest',
+        'PremptableDeficitCompensationTest'
     ]}
 
 
diff --git a/src/toil/batchSystems/mesos/batchSystem.py b/src/toil/batchSystems/mesos/batchSystem.py
index 6dddf66e..0a02e799 100644
--- a/src/toil/batchSystems/mesos/batchSystem.py
+++ b/src/toil/batchSystems/mesos/batchSystem.py
@@ -23,6 +23,7 @@ import socket
 import time
 from Queue import Queue, Empty
 from collections import defaultdict
+from operator import attrgetter
 from struct import unpack
 
 import itertools
@@ -286,7 +287,13 @@ class MesosBatchSystem(BatchSystemSupport,
 
     def _sortJobsByResourceReq(self):
         jobTypes = self.jobQueues.keys()
-        jobTypes.sort(key=ResourceRequirement.size)
+        # The dominant criteria is preemptability of jobs. Non-preemptable (NP) jobs should be
+        # considered first because they can only be run on on NP nodes while P jobs can run on
+        # both. Without this prioritization of NP jobs, P jobs could steal NP cores from NP jobs,
+        # leaving subsequently offered P cores unused. Despite the prioritization of NP jobs,
+        # NP jobs can not steal P cores from P jobs, simply because the offer-acceptance logic
+        # would not accept a P offer with a NP job.
+        jobTypes.sort(key=attrgetter('preemptable', 'size'))
         jobTypes.reverse()
         return jobTypes
 
diff --git a/src/toil/common.py b/src/toil/common.py
index 3c4b3003..7981a590 100644
--- a/src/toil/common.py
+++ b/src/toil/common.py
@@ -23,6 +23,7 @@ import tempfile
 import time
 from argparse import ArgumentParser
 
+from bd2k.util.exceptions import require
 from bd2k.util.humanize import bytes2human
 
 from toil.lib.bioio import addLoggingOptions, getLogLevelString, setLoggingFromOptions
@@ -78,6 +79,7 @@ class Config(object):
         self.alphaPacking = 0.8
         self.betaInertia = 1.2
         self.scaleInterval = 10
+        self.preemptableCompensation = 0.0
 
         #Resource requirements
         self.defaultMemory = 2147483648
@@ -197,7 +199,12 @@ class Config(object):
         setOption("betaInertia", float)
         setOption("scaleInterval", float)
 
-        #Resource requirements
+        setOption("preemptableCompensation", float)
+        require(0.0 <= self.preemptableCompensation <= 1.0,
+                '--preemptableCompensation (%f) must be >= 0.0 and <= 1.0',
+                self.preemptableCompensation)
+
+        # Resource requirements
         setOption("defaultMemory", h2b, iC(1))
         setOption("defaultCores", float, fC(1.0))
         setOption("defaultDisk", h2b, iC(1))
@@ -264,13 +271,14 @@ def _addOptions(addGroupFn, config):
                      "variables (TMPDIR, TEMP, TMP) via mkdtemp. This directory needs to exist on "
                      "all machines running jobs.")
     addOptionFn("--stats", dest="stats", action="store_true", default=None,
-                      help="Records statistics about the toil workflow to be used by 'toil stats'.")
-    addOptionFn("--clean", dest="clean", choices=['always', 'onError','never', 'onSuccess'], default=None,
-                      help=("Determines the deletion of the jobStore upon completion of the program. "
-                            "Choices: 'always', 'onError','never', 'onSuccess'. The --stats option requires "
-                            "information from the jobStore upon completion so the jobStore will never be deleted with"
-                            "that flag. If you wish to be able to restart the run, choose \\'never\\' or \\'onSuccess\\'. "
-                            "Default is \\'never\\' if stats is enabled, and \\'onSuccess\\' otherwise"))
+                help="Records statistics about the toil workflow to be used by 'toil stats'.")
+    addOptionFn("--clean", dest="clean", choices=['always', 'onError', 'never', 'onSuccess'],
+                default=None,
+                help=("Determines the deletion of the jobStore upon completion of the program. "
+                      "Choices: 'always', 'onError','never', 'onSuccess'. The --stats option requires "
+                      "information from the jobStore upon completion so the jobStore will never be deleted with"
+                      "that flag. If you wish to be able to restart the run, choose \\'never\\' or \\'onSuccess\\'. "
+                      "Default is \\'never\\' if stats is enabled, and \\'onSuccess\\' otherwise"))
     addOptionFn("--cleanWorkDir", dest="cleanWorkDir",
                 choices=['always', 'never', 'onSuccess', 'onError'], default='always',
                 help=("Determines deletion of temporary worker directory upon completion of a job. Choices: 'always', "
@@ -354,6 +362,15 @@ def _addOptions(addGroupFn, config):
     addOptionFn("--scaleInterval", dest="scaleInterval", default=None,
                 help=("The interval (seconds) between assessing if the scale of"
                       " the cluster needs to change. default=%s" % config.scaleInterval))
+    addOptionFn("--preemptableCompensation", dest="preemptableCompensation",
+                default=None,
+                help=("The preference of the autoscaler to replace preemptable nodes with "
+                      "non-preemptable nodes, when preemptable nodes cannot be started for some "
+                      "reason. Defaults to %s. This value must be between 0.0 and 1.0, inclusive. "
+                      "A value of 0.0 disables such compensation, a value of 0.5 compensates two "
+                      "missing preemptable nodes with a non-preemptable one. A value of 1.0 "
+                      "replaces every missing pre-emptable node with a non-preemptable one." %
+                      config.preemptableCompensation))
 
     #
     #Resource requirements
diff --git a/src/toil/job.py b/src/toil/job.py
index a729435d..0d80d616 100644
--- a/src/toil/job.py
+++ b/src/toil/job.py
@@ -1199,41 +1199,59 @@ class JobGraphDeadlockException( JobException ):
 
 class FunctionWrappingJob(Job):
     """
-    Job used to wrap a function. In its run method the wrapped function is called.
+    Job used to wrap a function. In its `run` method the wrapped function is called.
     """
     def __init__(self, userFunction, *args, **kwargs):
         """
-        :param userFunction: The function to wrap. The userFunction will be called \\
-        with the ``*args`` and ``**kwargs`` as arguments.
+        :param callable userFunction: The function to wrap. It will be called with ``*args`` and
+               ``**kwargs`` as arguments.
 
-        The keywords "memory", "cores", "disk", "cache" are reserved keyword arguments \\
-        that if specified will be used to determine the resources for the job, \\
-        as :func:`toil.job.Job.__init__`. If they are keyword arguments to the function
-        they will be extracted from the function definition, but may be overridden by
-        the user (as you would expect).
+        The keywords ``memory``, ``cores``, ``disk``, ``preemptable`` and ``checkpoint`` are
+        reserved keyword arguments that if specified will be used to determine the resources
+        required for the job, as :func:`toil.job.Job.__init__`. If they are keyword arguments to
+        the function they will be extracted from the function definition, but may be overridden
+        by the user (as you would expect).
         """
-        # Use the user specified resource argument, if specified, else
-        # grab the default argument from the function, if specified, else default to None
+        # Use the user-specified requirements, if specified, else grab the default argument
+        # from the function, if specified, else default to None
         argSpec = inspect.getargspec(userFunction)
-        argDict = dict(zip(argSpec.args[-len(argSpec.defaults):],argSpec.defaults)) \\
-                        if argSpec.defaults != None else {}
-        argFn = lambda x : kwargs.pop(x) if x in kwargs else \\
-                            (human2bytes(str(argDict[x])) if x in argDict.keys() else None)
-        Job.__init__(self, memory=argFn("memory"), cores=argFn("cores"),
-                     disk=argFn("disk"), preemptable=argFn("preemptable"),
-                     checkpoint=kwargs.pop("checkpoint") if "checkpoint" in kwargs else False)
-        #If dill is installed pickle the user function directly
-        #TODO: Add dill support
-        #else use indirect method
+        if argSpec.defaults is None:
+            argDict = {}
+        else:
+            argDict = dict(zip(argSpec.args[-len(argSpec.defaults):], argSpec.defaults))
+
+        def resolve(key, default=None, dehumanize=False):
+            try:
+                # First, try constructor arguments, ...
+                value = kwargs.pop(key)
+            except KeyError:
+                try:
+                    # ..., then try default value for function keyword arguments, ...
+                    value = argDict[key]
+                except KeyError:
+                    # ... and finally fall back to a default value.
+                    value = default
+            # Optionally, convert strings with metric or binary prefixes.
+            if dehumanize and isinstance(value, basestring):
+                value = human2bytes(value)
+            return value
+
+        Job.__init__(self,
+                     memory=resolve('memory', dehumanize=True),
+                     cores=resolve('cores', dehumanize=True),
+                     disk=resolve('disk', dehumanize=True),
+                     preemptable=resolve('preemptable'),
+                     checkpoint=resolve('checkpoint', default=False))
+
         self.userFunctionModule = ModuleDescriptor.forModule(userFunction.__module__).globalize()
         self.userFunctionName = str(userFunction.__name__)
-        self._args=args
-        self._kwargs=kwargs
+        self._args = args
+        self._kwargs = kwargs
 
     def _getUserFunction(self):
         logger.debug('Loading user function %s from module %s.',
                      self.userFunctionName,
-                     self.userFunctionModule) 
+                     self.userFunctionModule)
         userFunctionModule = self._loadUserModule(self.userFunctionModule)
         return getattr(userFunctionModule, self.userFunctionName)
 
diff --git a/src/toil/leader.py b/src/toil/leader.py
index 7d870a52..29e7a911 100644
--- a/src/toil/leader.py
+++ b/src/toil/leader.py
@@ -154,6 +154,7 @@ class JobBatcher:
         # Optional parameter which may be set if doing autoscaling
         self.clusterScaler = None
         self.jobsIssued = 0
+        self._preemptableJobsIssued = 0
         self.reissueMissingJobs_missingHash = {} #Hash to store number of observed misses
         self.serviceManager = serviceManager
 
@@ -162,6 +163,8 @@ class JobBatcher:
         Add a job to the queue of jobs
         """
         self.jobsIssued += 1
+        if preemptable:
+            self._preemptableJobsIssued += 1
         jobCommand = ' '.join((resolveEntryPoint('_toil_worker'), self.jobStoreLocator, jobStoreID))
         jobBatchSystemID = self.batchSystem.issueBatchJob(jobCommand, memory, cores, disk, preemptable)
         self.jobBatchSystemIDToIssuedJob[jobBatchSystemID] = IssuedJob(jobStoreID, memory, cores, disk, preemptable)
@@ -176,13 +179,22 @@ class JobBatcher:
         for jobStoreID, memory, cores, disk, preemptable in jobs:
             self.issueJob(jobStoreID, memory, cores, disk, preemptable)
 
-    def getNumberOfJobsIssued(self):
+    def getNumberOfJobsIssued(self, preemptable=None):
         """
         Gets number of jobs that have been added by issueJob(s) and not
         removed by removeJobID
+
+        :param None or boolean preemptable: If none, return all types of jobs.
+          If true, return just the number of preemptable jobs. If false, return
+          just the number of non-preemptable jobs.
         """
-        assert self.jobsIssued >= 0
-        return self.jobsIssued
+        assert self.jobsIssued >= 0 and self._preemptableJobsIssued >= 0
+        if preemptable is None:
+            return self.jobsIssued
+        elif preemptable:
+            return self._preemptableJobsIssued
+        else:
+            return (self.jobsIssued - self._preemptableJobsIssued)
 
     def getJob(self, jobBatchSystemID):
         """
@@ -208,6 +220,9 @@ class JobBatcher:
         """
         assert jobBatchSystemID in self.jobBatchSystemIDToIssuedJob
         self.jobsIssued -= 1
+        if self.jobBatchSystemIDToIssuedJob[jobBatchSystemID].preemptable:
+            assert self._preemptableJobsIssued > 0
+            self._preemptableJobsIssued -= 1
         jobStoreID = self.jobBatchSystemIDToIssuedJob.pop(jobBatchSystemID).jobStoreID
         return jobStoreID
 
diff --git a/src/toil/provisioners/clusterScaler.py b/src/toil/provisioners/clusterScaler.py
index a99953ad..4f790c99 100644
--- a/src/toil/provisioners/clusterScaler.py
+++ b/src/toil/provisioners/clusterScaler.py
@@ -28,6 +28,18 @@ from toil.provisioners.abstractProvisioner import AbstractProvisioner, Shape
 
 logger = logging.getLogger(__name__)
 
+# A *deficit* exists when we have more jobs that can run on preemptable nodes than we have
+# preemptable nodes. In order to not block these jobs, we want to increase the number of non-
+# preemptable nodes that we have and need for just non-preemptable jobs. However, we may still
+# prefer waiting for preemptable instances to come available.
+#
+# To accomodate this, we set the delta to the difference between the number of provisioned
+# preemptable nodes and the number of nodes that were requested. when the non-preemptable thread
+# wants to provision nodes, it will multiply this delta times a preference for preemptable vs.
+# non-preemptable nodes.
+
+_preemptableNodeDeficit = 0
+
 
 class RecentJobShapes(object):
     """
@@ -282,6 +294,8 @@ class ScalerThread(ExceptionalThread):
         self.maxNodes = scaler.config.maxPreemptableNodes if preemptable else scaler.config.maxNodes
 
     def tryRun(self):
+        global _preemptableNodeDeficit
+
         if isinstance(self.scaler.jobBatcher.batchSystem, AbstractScalableBatchSystem):
             totalNodes = len(self.scaler.jobBatcher.batchSystem.getNodes(self.preemptable))
         else:
@@ -291,7 +305,8 @@ class ScalerThread(ExceptionalThread):
             with throttle(self.scaler.config.scaleInterval):
                 # Calculate the approx. number nodes needed
                 # TODO: Correct for jobs already running which can be considered fractions of a job
-                queueSize = self.scaler.jobBatcher.getNumberOfJobsIssued()
+                queueSize = self.scaler.jobBatcher.getNumberOfJobsIssued(preemptable=self.preemptable)
+
                 recentJobShapes = self.jobShapes.get()
                 assert len(recentJobShapes) > 0
                 nodesToRunRecentJobs = binPacking(recentJobShapes, self.nodeShape)
@@ -301,6 +316,19 @@ class ScalerThread(ExceptionalThread):
                     * float(queueSize)
                     / len(recentJobShapes))))
 
+                # If we're the non-preemptable scaler, we need to see if we have a deficit of
+                # preemptable nodes that we should compensate for.
+                if not self.preemptable:
+                    compensation = self.scaler.config.preemptableCompensation
+                    assert 0.0 <= compensation <= 1.0
+                    # The number of nodes we provision as compensation for missing preemptable
+                    # nodes is the product of the deficit (the number of preemptable nodes we did
+                    # _not_ allocate) and configuration preference.
+                    compensationNodes = int(round(_preemptableNodeDeficit * compensation))
+                    logger.info('Adding %d preemptable nodes to compensate for a deficit of %d '
+                                'non-preemptable ones.', compensationNodes, _preemptableNodeDeficit)
+                    estimatedNodes += compensationNodes
+
                 fix_my_name = (0 if nodesToRunRecentJobs <= 0
                                else len(recentJobShapes) / float(nodesToRunRecentJobs))
                 logger.debug('Estimating that cluster needs %s nodes of shape %s, from current '
@@ -331,6 +359,19 @@ class ScalerThread(ExceptionalThread):
                                 estimatedNodes)
                     totalNodes = self.scaler.provisioner.setNodeCount(numNodes=estimatedNodes,
                                                                       preemptable=self.preemptable)
+                    
+                    # If we were scaling up the number of preemptable nodes and failed to meet
+                    # our target, we need to update the slack so that non-preemptable nodes will
+                    # be allocated instead and we won't block. If we _did_ meet our target,
+                    # we need to reset the slack to 0.
+                    if self.preemptable:
+                        if totalNodes < estimatedNodes:
+                            deficit = estimatedNodes - totalNodes
+                            logger.info('Preemptable scaler detected deficit of %d nodes.', deficit)
+                            _preemptableNodeDeficit = deficit
+                        else:
+                            _preemptableNodeDeficit = 0
+                    
         logger.info('Forcing provisioner to reduce cluster size to zero.')
         totalNodes = self.scaler.provisioner.setNodeCount(numNodes=0,
                                                           preemptable=self.preemptable,
diff --git a/src/toil/test/__init__.py b/src/toil/test/__init__.py
index 6252c8ed..e8b17770 100644
--- a/src/toil/test/__init__.py
+++ b/src/toil/test/__init__.py
@@ -186,7 +186,8 @@ class ToilTest(unittest.TestCase):
         assert all(path.startswith('src') for path in dirty)
         dirty = set(dirty)
         dirty.difference_update(excluded)
-        assert not dirty, "Run 'make sdist'. Files newer than %s: %r" % (sdistPath, list(dirty))
+        assert not dirty, \\
+            "Run 'make clean_sdist sdist'. Files newer than %s: %r" % (sdistPath, list(dirty))
         return sdistPath
 
     @classmethod
@@ -198,8 +199,9 @@ class ToilTest(unittest.TestCase):
 
         :param str args: Any arguments to be passed to the command.
 
-        :param Any kwargs: keyword arguments for subprocess.Popen constructor. Pass capture=True to
-                           have the process' stdout returned.
+        :param Any kwargs: keyword arguments for subprocess.Popen constructor. Pass capture=True
+               to have the process' stdout returned. Pass input='some string' to feed input to the
+               process' stdin.
 
         :rtype: None|str
 
diff --git a/src/toil/test/provisioners/cgcloud/provisionerTest.py b/src/toil/test/provisioners/cgcloud/provisionerTest.py
index a69095b2..a912495b 100644
--- a/src/toil/test/provisioners/cgcloud/provisionerTest.py
+++ b/src/toil/test/provisioners/cgcloud/provisionerTest.py
@@ -86,15 +86,6 @@ class AbstractCGCloudProvisionerTest(ToilTest, CgcloudTestCase):
     else:
         debugEggPath = '/Applications/PyCharm 2016.1.app/Contents/debug-eggs/pycharm-debug.egg'
 
-    # A pip-installable release of toil-scripts or a URL pointing to a source distribution.
-    # Typically you would specify a GitHub archive URL of a specific branch, tag or commit. The
-    # first path component of each tarball entry will be stripped.
-    #
-    if True:
-        toilScripts = '2.1.0a1.dev455'
-    else:
-        toilScripts = 'https://api.github.com/repos/BD2KGenomics/toil-scripts/tarball/master'
-
     # The instance types to use for leader and workers
     #
     instanceType = 'm3.large'
@@ -102,7 +93,11 @@ class AbstractCGCloudProvisionerTest(ToilTest, CgcloudTestCase):
 
     # The spot bid to use for preemptable instances. A safe bet is the on-demand price for the
     # selected instance type.
-    spotBid = '0.133'
+    safeSpotBid = '0.133'
+
+    # The spot market should never fulfill a spot bid of one hundredth of one cent for m3.large
+    # (the average bid price is ~$0.02).
+    unfullfillableSpotBid = "0.001"
 
     @classmethod
     def setUpClass(cls):
@@ -116,7 +111,9 @@ class AbstractCGCloudProvisionerTest(ToilTest, CgcloudTestCase):
         binPath = os.path.join(path, 'bin')
         cls.oldPath = os.environ['PATH']
         os.environ['PATH'] = os.pathsep.join(concat(binPath, cls.oldPath.split(os.pathsep)))
-        cls._run('pip', 'install', 'cgcloud-toil==' + cgcloudVersion)
+        # Suppress pip's cache to avoid concurrency issues. Keep in mind that the concrete
+        # subclasses may be run concurrently. See run_tests.sh.
+        cls._run('pip', 'install', '--no-cache', 'cgcloud-toil==' + cgcloudVersion)
         if cls.createImage:
             cls._cgcloud('create', '-IT',
                          '--option', 'toil_sdists=%s[aws,mesos]' % cls.sdistPath,
@@ -138,8 +135,8 @@ class AbstractCGCloudProvisionerTest(ToilTest, CgcloudTestCase):
             os.environ['CGCLOUD_PLUGINS'] = self.saved_cgcloud_plugins
         super(AbstractCGCloudProvisionerTest, self).tearDown()
 
-    def _test(self, autoScaled=False, spotInstances=False):
-        self.assertTrue(not spotInstances or autoScaled,
+    def _test(self, autoScaled=False, spotBid=None, preemptableCompensation=None):
+        self.assertTrue(spotBid is None or autoScaled,
                         'This test does not support a static cluster of spot instances.')
         if self.createCluster:
             self._cgcloud('create-cluster',
@@ -170,14 +167,17 @@ class AbstractCGCloudProvisionerTest(ToilTest, CgcloudTestCase):
                                     '--nodeType=' + self.instanceType,
                                     '--maxNodes=%s' % self.numWorkers,
                                     '--logDebug'])
-            if spotInstances:
+            if spotBid is not None:
                 toilOptions.extend([
-                    '--preemptableNodeType=%s:%s' % (self.instanceType, self.spotBid),
+                    '--preemptableNodeType=%s:%s' % (self.instanceType, spotBid),
                     # The RNASeq pipeline does not specify a preemptability requirement so we
                     # need to specify a default, otherwise jobs would never get scheduled.
                     '--defaultPreemptable',
                     '--maxPreemptableNodes=%s' % self.numWorkers])
 
+            if preemptableCompensation:
+                toilOptions.extend(['--preemptableCompensation', preemptableCompensation])
+
             self._runScript(toilOptions)
 
         finally:
@@ -194,10 +194,11 @@ class AbstractCGCloudProvisionerTest(ToilTest, CgcloudTestCase):
     @abstractmethod
     def _runScript(self, toilOptions):
         """
-        Modify the provided Toil options to suit the test Toil script, then run the script with those arguments.
+        Modify the provided Toil options to suit the test Toil script, then run the script with
+        those arguments.
 
-        :param toilOptions: List of Toil command line arguments. This list may need to be modified to suit
-            the test script's requirements.
+        :param toilOptions: List of Toil command line arguments. This list may need to be
+               modified to suit the test script's requirements.
         """
         raise NotImplementedError()
 
@@ -216,9 +217,11 @@ class AbstractCGCloudProvisionerTest(ToilTest, CgcloudTestCase):
             admin = False
         else:
             del kwargs['admin']
-
-        cls._cgcloud(
-            *filter(None, concat('ssh', '-a' if admin else None, role, cls.sshOptions, args)), **kwargs)
+        cls._cgcloud(*filter(None, concat('ssh',
+                                          '-a' if admin else None,
+                                          role,
+                                          cls.sshOptions,
+                                          args)), **kwargs)
 
     @classmethod
     def _rsync(cls, role, *args):
@@ -230,6 +233,18 @@ class AbstractCGCloudProvisionerTest(ToilTest, CgcloudTestCase):
 
 
 class CGCloudRNASeqTest(AbstractCGCloudProvisionerTest):
+    """
+    Test the CGCloud autoscaler.
+    """
+    # A pip-installable release of toil-scripts or a URL pointing to a source distribution.
+    # Typically you would specify a GitHub archive URL of a specific branch, tag or commit. The
+    # first path component of each tarball entry will be stripped.
+    #
+    if True:
+        toilScripts = '2.1.0a1.dev455'
+    else:
+        toilScripts = 'https://api.github.com/repos/BD2KGenomics/toil-scripts/tarball/master'
+
     def __init__(self, name):
         super(CGCloudRNASeqTest, self).__init__(name)
         # The number of samples to run the test workflow on
@@ -267,41 +282,41 @@ class CGCloudRNASeqTest(AbstractCGCloudProvisionerTest):
 
     @integrative
     def testAutoScaledSpotCluster(self):
-        self._test(autoScaled=True, spotInstances=True)
-
-
-def restartScript():
-    from toil.job import Job
-    import argparse
-    import os
-
-    def f0(job):
-        if 'FAIL' in os.environ:
-            raise RuntimeError('failed on purpose')
-
-    if __name__ == '__main__':
-        parser = argparse.ArgumentParser()
-        Job.Runner.addToilOptions(parser)
-        options = parser.parse_args()
-        i = Job.Runner.startToil(Job.wrapJobFn(f0, cores=0.5, memory='50 M', disk='50 M'),
-                                 options)
+        self._test(autoScaled=True, spotBid=self.safeSpotBid)
 
 
 class CGCloudRestartTest(AbstractCGCloudProvisionerTest):
     """
     This test insures autoscaling works on a restarted Toil run
     """
+
     def setUp(self):
         super(CGCloudRestartTest, self).setUp()
         # CGCloud provisioner requires that the node has at least 1 ephemeral drive so this is the
         # smallest instance we can use
         self.instanceType = 'm3.medium'
         self.leaderInstanceType = 'm3.medium'
+        self.scriptName = "restartScript.py"
 
     def _getScript(self):
-        self.scriptName= "restartScript.py"
-        self._leader('tee', self.scriptName, input=dedent('\\n'.join(getsource(restartScript).split('\\n')[1:])),
-)
+        def restartScript():
+            from toil.job import Job
+            import argparse
+            import os
+
+            def f0(job):
+                if 'FAIL' in os.environ:
+                    raise RuntimeError('failed on purpose')
+
+            if __name__ == '__main__':
+                parser = argparse.ArgumentParser()
+                Job.Runner.addToilOptions(parser)
+                options = parser.parse_args()
+                rootJob = Job.wrapJobFn(f0, cores=0.5, memory='50 M', disk='50 M')
+                Job.Runner.startToil(rootJob, options)
+
+        script = dedent('\\n'.join(getsource(restartScript).split('\\n')[1:]))
+        self._leader('tee', self.scriptName, input=script)
 
     def _runScript(self, toilOptions):
         # clean = onSuccess
@@ -319,3 +334,39 @@ class CGCloudRestartTest(AbstractCGCloudProvisionerTest):
     @integrative
     def testAutoScaledCluster(self):
         self._test(autoScaled=True)
+
+
+class PremptableDeficitCompensationTest(AbstractCGCloudProvisionerTest):
+    def test(self):
+        self._test(autoScaled=True, spotBid=self.unfullfillableSpotBid)
+
+    def _getScript(self):
+        def userScript():
+            from toil.job import Job
+            from toil.common import Toil
+
+            # Because this is the only job in the pipeline and because it is preemptable,
+            # there will be no non-preemptable jobs. The non-preemptable scaler will therefore
+            # not request any nodes initially. And since we made it impossible for the
+            # preemptable scaler to allocate any nodes (using an abnormally low spot bid),
+            # we will observe a deficit of preemptable nodes that the non-preemptable scaler will
+            # compensate for by spinning up non-preemptable nodes instead.
+            #
+            def job(job, disk='10M', cores=1, memory='10M', preemptable=True):
+                pass
+
+            if __name__ == '__main__':
+                options = Job.Runner.getDefaultArgumentParser().parse_args()
+                with Toil(options) as toil:
+                    if toil.config.restart:
+                        toil.restart()
+                    else:
+                        toil.start(Job.wrapJobFn(job))
+
+        script = dedent('\\n'.join(getsource(userScript).split('\\n')[1:]))
+        self._leader('tee userScript.py', input=script)
+
+    def _runScript(self, toilOptions):
+        toilOptions.extend([
+            '--preemptableCompensation=1.0', self.jobStore])
+        self._leader('python', 'userScript.py', *toilOptions)