#   Copyright 2015 Ufora Inc.
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

import logging
import random
import time
import unittest
import ufora.cumulus.test.InMemoryCumulusSimulation as InMemoryCumulusSimulation
import ufora.distributed.S3.InMemoryS3Interface as InMemoryS3Interface
import ufora.native.CallbackScheduler as CallbackScheduler
import ufora.native.Cumulus as CumulusNative
import ufora.native.Hash as HashNative

#import ufora.distributed.Storage.HdfsObjectStore as HdfsObjectStore

callbackScheduler = CallbackScheduler.singletonForTesting()

#note that we are being careful not to use anything from the builtins in these examples
#so that checkpointing is fast. Currently, "FORA.eval" doesn't use the module subsampler,
#so we end up holding the entire builtins if we use it.

def expensiveChildCachecalls(ix):
    return  """
        let sumf = fun(a,b) { 
            if (a+1 >= b)
                return [sum(a, 10**12) + %s]

            let mid = (a+b)/2

            return sumf(a,mid) + sumf(mid,b)
            }

        sumf(0,10)
        """ % ix

vecOfVecCalcText = """
    let sum = fun(a,b,f) {
        if (a >= b) return 0
        if (a+1 >= b) return f(a)
        let mid = (a+b)/2;
        return sum(a,mid,f) + sum(mid,b,f)
        }
    let res = [];
    let ix = 0;
    while (ix < 100) {
        res = res + sum(0,100, fun(x) { [[sum(x, 10**6, fun(x){x+1})].paged].paged })
        ix = ix + 1
        }
    res.sum()
    """

vecLoopSumText = """
    let res = 0;
    let ix = 0;
    while (ix < 1000000) {
        res = res + Vector.range(100000).sum()
        ix = ix + 1
        }
    res
    """

repeatedVecInLoop = """
    let v = Vector.range(1000000)
    let ix = 0

    while (ix < 1000000) {
        v = v.apply({_+1}).paged
        ix = ix + 1
        }
    v.sum()
    """

simpleSumInLoopText = """
    let v = Vector.range(1000000).paged;
    let sum = fun(a,b,f) {
        if (a >= b) return 0
        if (a+1 >= b) return f(a)
        let mid = (a+b)/2;
        return sum(a,mid,f) + sum(mid,b,f)
        }

    let res = 0;

    let ix = 0
    while (ix < 100) {
        ix = ix + 1
        res = res + sum(ix, Int64(10**9), {if (_ % 16 == 0) v[(_ * 1024) % size(v)] else _ });
        }

    res
    """

sumInLoopText = """
    let sum = fun(a,b,f) {
        if (a >= b) return 0
        if (a+1 >= b) return f(a)
        let mid = (a+b)/2;
        return sum(a,mid,f) + sum(mid,b,f)
        }

    let res = 0;
    let ix = 0;
    while (ix < 10000) {
        let bound = match(ix%3) with (0) { 1 } (1) { 10 } (2) { 100 };

        res = res + sum(0,bound * 1000000, fun(x) { x + 1 })

        ix = ix + 1
        }
    res
    """

bigSumText = """
    let sum = fun(a,b,f) {
        if (a >= b) return 0
        if (a+1 >= b) return f(a)
        let mid = (a+b)/2;
        return sum(a,mid,f) + sum(mid,b,f)
        }

    sum(0,10**13,{_})
    """

cachedBigSumText = """
    let sum = fun(a,b,f) {
        if (a >= b) return 0
        if (a+1 >= b) return f(a)
        let mid = (a+b)/2;
        return sum(a,mid,f) + sum(mid,b,f)
        }

    cached(sum(0,10**13,{_}))[0]
    """

TIMEOUT = 30

class CheckpointingTest(unittest.TestCase):
    def waitForAllCheckpointsToClear(self, simulation, timeout = TIMEOUT):
        t0 = time.time()
        while time.time() < t0 + timeout:
            if not simulation.getGlobalScheduler().anyOutstandingTriggeredCheckpoints():
                return
            time.sleep(0.01)

        assert False, "timed out"

    def timestampOfMostRecentFullCheckpoint(self, simulation, onlyUnfinished = True, onlySuccessful = True):
        if not simulation.getGlobalScheduler():
            return None

        statuses = simulation.getGlobalScheduler().currentOutstandingCheckpointStatuses(onlyUnfinished, True)
        if len(statuses):
            (computation, (checkpointStatus, checkpointRequest)) = statuses[0]

            timestamp = checkpointRequest.timestamp
            isFull = checkpointRequest.writeToStorage and (not onlySuccessful or checkpointStatus.checkpointSuccessful)

            if isFull:
                return timestamp

    def timestampOfMostRecentCheckpoint(self, simulation):
        if not simulation.getGlobalScheduler():
            return None

        statuses = simulation.getGlobalScheduler().currentOutstandingCheckpointStatuses(True, False)
        if len(statuses):
            (computation, (checkpointStatus, checkpointRequest)) = statuses[0]

            return checkpointRequest.timestamp

    def timeElapsedOfMostRecentCheckpoints(self, simulation, onlyUnfinished = True, onlyCommitted = False):
        if not simulation.getGlobalScheduler():
            return {}

        statuses = simulation.getGlobalScheduler().currentOutstandingCheckpointStatuses(onlyUnfinished, onlyCommitted)
        return {status[0]: status[1][0].statistics.timeElapsed.timeSpentInCompiledCode for status in statuses}

    def totalTimeElapsedOfMostRecentCheckpoints(self, simulation, onlyUnfinished = True, onlyCommitted = False):
        return sum(self.timeElapsedOfMostRecentCheckpoints(simulation, onlyUnfinished, onlyCommitted).values(), 0)

    def waitForCheckpoint(self, simulation, priorCheckpoint = None, checkInterval = 0.1, onlyUnfinished = True):
        t1 = time.time()
        foundFullCheckpoint = False
        while time.time() - t1 < TIMEOUT and not foundFullCheckpoint:
            scheduler = simulation.getGlobalScheduler()
            if scheduler:
                statuses = simulation.getGlobalScheduler().currentOutstandingCheckpointStatuses(onlyUnfinished, False)
                if statuses:
                    checkpointSecondsElapsed = statuses[0][1][0].statistics.timeElapsed.timeSpentInCompiledCode

                    if priorCheckpoint is None or priorCheckpoint < checkpointSecondsElapsed:
                        foundFullCheckpoint = True

            if not foundFullCheckpoint:
                time.sleep(checkInterval)

        if foundFullCheckpoint:
            return checkpointSecondsElapsed

    def waitForNFullCheckpoints(self, simulation, count, checkInterval = 0.1):
        t1 = time.time()
        while time.time() - t1 < TIMEOUT:
            scheduler = simulation.getGlobalScheduler()
            found = []
            if scheduler:
                statuses = simulation.getGlobalScheduler().currentOutstandingCheckpointStatuses(True, True)
                for (computation, (stats, checkpoint)) in statuses:
                    if checkpoint.writeToStorage and stats.checkpointSuccessful:
                        checkpointSecondsElapsed = stats.statistics.timeElapsed.timeSpentInCompiledCode
                        found.append(checkpointSecondsElapsed)

            if len(found) < count:
                time.sleep(checkInterval)
            else:
                return found


    def waitForFullCheckpoint(self, simulation, priorCheckpoint = None, checkInterval = 0.1, onlyUnfinished = True, onlySuccessful = False):
        t1 = time.time()
        foundFullCheckpoint = False
        while time.time() - t1 < TIMEOUT and not foundFullCheckpoint:
            scheduler = simulation.getGlobalScheduler()
            if scheduler:
                statuses = simulation.getGlobalScheduler().currentOutstandingCheckpointStatuses(onlyUnfinished, True)
                if statuses:
                    (computation, (stats, checkpoint)) = statuses[0]
                    if checkpoint.writeToStorage and (not onlySuccessful or stats.checkpointSuccessful):
                        checkpointSecondsElapsed = stats.statistics.timeElapsed.timeSpentInCompiledCode

                        if priorCheckpoint is None or priorCheckpoint < checkpointSecondsElapsed:
                            foundFullCheckpoint = True

            if not foundFullCheckpoint:
                time.sleep(checkInterval)

        if foundFullCheckpoint:
            return checkpointSecondsElapsed

    def createSimulation(self,
            useHdfsObjectStore=False,
            objectStore=None,
            sharedStateViewFactory=None,
            workerCount=4,
            machineIdHashSeed=None,
            s3Service=None
            ):
        s3 = s3Service or InMemoryS3Interface.InMemoryS3InterfaceFactory()
        return InMemoryCumulusSimulation.InMemoryCumulusSimulation(
            workerCount,
            1,
            memoryPerWorkerMB=100,
            threadsPerWorker=2,
            s3Service=s3,
            objectStore=objectStore,
            sharedStateViewFactory=sharedStateViewFactory,
            machineIdHashSeed=machineIdHashSeed
            )

    def test_checkpointingCumulusClientRequestPathway(self):
        simulation = self.createSimulation()
        #give the simulation a couple of seconds to pick a scheduler
        t0 = time.time()
        while simulation.getGlobalScheduler() is None:
            time.sleep(0.01)
            self.assertTrue(time.time() - t0 < 2.0)

        simulation.getGlobalScheduler().setCheckpointStatusInterval(0.0001)
        count = 0

        try:
            simulation.submitComputation(simpleSumInLoopText)

            while time.time() - t0 < 2.0:
                result = simulation.getCurrentCheckpointStatistics(timeout = TIMEOUT)
                count = count + 1

            self.assertTrue(count > 10)

            print "Total roundtrips: ", count
        finally:
            simulation.teardown()

    def test_checkpointingSystemWritesToS3(self):
        simulation = self.createSimulation()
        self.assertTrue(len(simulation.objectStore.listValues()) == 0)

        try:
            #give the simulation a couple of seconds to pick a scheduler
            t0 = time.time()
            while simulation.getGlobalScheduler() is None:
                time.sleep(0.01)
                self.assertTrue(time.time() - t0 < TIMEOUT, "never got a scheduler")

            simulation.getGlobalScheduler().setCheckpointStatusInterval(0.01)

            simulation.submitComputation(simpleSumInLoopText)

            time.sleep(1.0)

            count = 0
            lastCheckpoint = None
            while time.time() - t0 < 20.0:
                simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()

                foundFullCheckpoint = False
                t1 = time.time()
                while time.time() - t1 < 10.0 and not foundFullCheckpoint:
                    statuses = simulation.getGlobalScheduler().currentOutstandingCheckpointStatuses(True, True)
                    if len(statuses):
                        (computation, (stats, checkpoint)) = statuses[0]
                        newCheckpoint = checkpoint.timestamp
                        if lastCheckpoint is None or newCheckpoint != lastCheckpoint:
                            lastCheckpoint = newCheckpoint
                            if checkpoint.writeToStorage and stats.checkpointSuccessful:
                                foundFullCheckpoint = True
                    time.sleep(.1)

                self.assertTrue(foundFullCheckpoint)

                count += 1
                logging.info(
                    "Total: %d after %s with %d files.",
                    count,
                    time.time() - t0,
                    len(simulation.objectStore.listValues())
                    )

                self.assertGreater(
                    len(simulation.objectStore.listValues()),
                    0)

                guids = simulation.getWorkerVdm(0).getPersistentCacheIndex().allCheckpointedComputationGuids()
                self.assertGreater(
                    len(guids),
                    0,
                    "Didn't write the checkpoint to the persistent cache"
                    )

                for item in simulation.objectStore.listValues():
                    simulation.objectStore.deleteValue(item[0])

        except:
            simulation.dumpSchedulerEventStreams()
            raise
        finally:
            simulation.teardown()

    def test_checkpointingRecoverySimpleSum(self):
        self.recoveryTest(simpleSumInLoopText)

    def test_checkpointingRecoveryVecLoop(self):
        for ix in range(3):
            self.recoveryTest(
                repeatedVecInLoop,
                interval1 = 1.0,
                interval2 = 1.0,
                interval3 = 1.0,
                initialWorkers = 4,
                workersToDrop1 = 3,
                workersToAdd1 = 3,
                machineIdHashSeed=str(ix)
                )

    def test_s3DatasetComputationHashesAreStable(self):
        for ix in range(2):
            s3 = InMemoryS3Interface.InMemoryS3InterfaceFactory()

            s3().setKeyValue(
                "bucketname",
                "key",
                "this is some data"
                )

            self.recoveryTest("""
                let data = cached(fun() { datasets.s3("bucketname","key") }())[0];

                cached(fun() { sum(0,10**12, fun(ix) { data[ix % size(data)] }) }())[0]
                """,
                s3Service = s3,
                interval1 = 1.0,
                interval2 = 1.0,
                interval3 = 1.0,
                interval4 = 1.0,
                workersToDrop1 = 3,
                workersToAdd1 = 3,
                workersToDrop2 = 3,
                workersToAdd2 = 3,
                machineIdHashSeed=str(ix)
                )

    def test_recoveryWithUnreadDatasetsS3(self):
        s3 = InMemoryS3Interface.InMemoryS3InterfaceFactory()

        s3().setKeyValue(
            "bucketname",
            "key",
            "this is some data"
            )

        simulation = self.createSimulation(s3Service = s3)

        try:
            #give the simulation a couple of seconds to pick a scheduler
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

            simulation.submitComputation("""
                let data = datasets.s3("bucketname","key")

                let res = sum(0,10**12)

                data[res % 2]
                """)

            time.sleep(1.0)

            simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()
            self.waitForAllCheckpointsToClear(simulation)
        finally:
            simulation.teardown()


    def recoveryTest(self,
                text,
                interval1 = 5.0,
                interval2 = 1.0,
                interval3 = 1.0,
                interval4 = 1.0,
                initialWorkers = 4,
                workersToDrop1 = 1,
                workersToAdd1 = 0,
                workersToDrop2 = 1,
                workersToAdd2 = 0,
                machineIdHashSeed=None,
                s3Service=None
                ):
        simulation = self.createSimulation(machineIdHashSeed=machineIdHashSeed, workerCount = initialWorkers, s3Service = s3Service)

        try:
            self.assertTrue(len(simulation.objectStore.listValues()) == 0)

            #give the simulation a couple of seconds to pick a scheduler
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

            simulation.submitComputation(text)

            time.sleep(interval1)

            simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()
            self.waitForAllCheckpointsToClear(simulation)
            checkpointSecondsElapsed = self.totalTimeElapsedOfMostRecentCheckpoints(simulation, onlyUnfinished=False, onlyCommitted=True)

            time.sleep(interval2)

            for _ in range(workersToDrop1):
                simulation.dropTopWorker()
            for _ in range(workersToAdd1):
                simulation.addWorker()

            self.assertTrue(simulation.waitForHandshake())

            time.sleep(interval3)

            simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()
            self.waitForAllCheckpointsToClear(simulation)

            checkpointSecondsElapsed2 = self.totalTimeElapsedOfMostRecentCheckpoints(simulation, onlyUnfinished=False, onlyCommitted=True)

            self.assertTrue(
                checkpointSecondsElapsed2 > checkpointSecondsElapsed,
                "Expected %s to be bigger than %s" % (checkpointSecondsElapsed2, checkpointSecondsElapsed)
                )

            for _ in range(workersToDrop2):
                simulation.dropBottomWorker()
            for _ in range(workersToAdd2):
                simulation.addWorker()

            self.assertTrue(simulation.waitForHandshake())

            time.sleep(interval4)

            simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()
            self.waitForAllCheckpointsToClear(simulation)

            checkpointSecondsElapsed3 = self.totalTimeElapsedOfMostRecentCheckpoints(simulation, onlyUnfinished=False, onlyCommitted=True)

            self.assertTrue(
                checkpointSecondsElapsed3 > checkpointSecondsElapsed2,
                "Expected %s to be bigger than %s" % (checkpointSecondsElapsed3, checkpointSecondsElapsed2)
                )
        finally:
            simulation.teardown()

    def test_checkpointingRecoveryFromCorruptedCacheStatePage(self):
        self.checkpointingRecoveryFromCorruptedCacheState("_page")

    def test_checkpointingRecoveryFromCorruptedCacheStateSummary(self):
        self.checkpointingRecoveryFromCorruptedCacheState("_summary")

    def test_checkpointingRecoveryFromCorruptedCacheStateSlice(self):
        self.checkpointingRecoveryFromCorruptedCacheState("_slice")

    def test_checkpointingRecoveryFromCorruptedCacheStateBigvec(self):
        self.checkpointingRecoveryFromCorruptedCacheState("_bigvec")

    def test_checkpointingRecoveryFromCorruptedCacheStateAll(self):
        self.checkpointingRecoveryFromCorruptedCacheState("")

    def checkpointingRecoveryFromCorruptedCacheState(self, suffix):
        simulation = self.createSimulation(workerCount=1)
        self.assertTrue(len(simulation.objectStore.listValues()) == 0)

        viewFactory = simulation.sharedStateViewFactory
        objectStore = simulation.objectStore

        try:
            #give the simulation a couple of seconds to pick a scheduler
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

            simulation.submitComputation(simpleSumInLoopText)

            time.sleep(5.0)

            simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()

            self.waitForAllCheckpointsToClear(simulation)

            elapsed = sum(self.timeElapsedOfMostRecentCheckpoints(simulation).values(),0.0)

            print "Total seconds elapsed in computing: ", elapsed

            simulation.dropTopWorker()

            #delete everything in the object store
            keys = simulation.objectStore.listValues()

            for k in keys:
                if k[0].endswith(suffix):
                    logging.info("Deleting %s", k[0])
                    simulation.objectStore.deleteValue(k[0])
    
            #give the simulation a couple of seconds to pick a scheduler
            simulation.addWorker()

            self.assertTrue(simulation.waitForHandshake(), "timed out trying to handshake")
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))
            simulation.getGlobalScheduler().setCheckpointStatusInterval(0.1)

            currentRegime = simulation.currentRegime()

            #submit the computation which should trigger an invalid-checkpoint
            #simulation.submitComputation(simpleSumInLoopText)

            #wait until we start making progress again
            time.sleep(5)

            cacheIndex = simulation.getWorkerVdm(0).getPersistentCacheIndex()

            self.assertTrue(len(cacheIndex.allCheckpointedComputationGuids()) == 0)
            self.assertTrue(cacheIndex.totalBytesInCache() == 0)
        finally:
            simulation.teardown()



    def DISABLEDtest_checkpointingAlwaysSuccessful(self):
        simulation = self.createSimulation()
        self.assertTrue(len(simulation.objectStore.listValues()) == 0)

        try:
            #give the simulation a couple of seconds to pick a scheduler
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

            simulation.submitComputation(sumInLoopText)

            t0 = time.time()

            timestamp = None

            simulation.waitForGlobalScheduler()
            time.sleep(1.0)

            count = 0
            timeOfLastCheckpoint = time.time()
            while time.time() - t0 < 10.0:
                simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()

                timestamp = self.waitForFullCheckpoint(simulation, priorCheckpoint = timestamp, checkInterval = 0.01)

                statuses = simulation.getGlobalScheduler().currentOutstandingCheckpointStatuses(True, True)

                wasSuccessful = statuses[0][1][0].checkpointSuccessful

                self.assertTrue(wasSuccessful)

                for item in simulation.objectStore.listValues():
                    simulation.objectStore.deleteValue(item[0])

                timeOfLastCheckpoint = time.time()

                count += 1

            #make sure we didn't stall either
            self.assertTrue(timeOfLastCheckpoint - time.time() < 5.0)

            print "completed ", count, " checkpoints in 10 seconds."
        finally:
            simulation.teardown()


    def test_checkpointingGarbageCollection(self):
        simulation = self.createSimulation()
        self.assertTrue(len(simulation.objectStore.listValues()) == 0)

        try:
            #give the simulation a couple of seconds to pick a scheduler
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

            simulation.submitComputation(sumInLoopText)

            t0 = time.time()

            timestamp = None

            simulation.waitForGlobalScheduler()
            time.sleep(1.0)

            count = 0
            timeOfLastCheckpoint = time.time()
            while count < 10:
                simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()
                self.waitForAllCheckpointsToClear(simulation)
                count += 1

            cache = simulation.getWorkerVdm(0).getPersistentCacheIndex()

            computations = cache.allCheckpointedComputations()

            self.assertTrue(len(computations) == 1)

            checkpoints = cache.checkpointsForComputation(computations[0])

            maxSeconds = 0.0
            for c in checkpoints:
                maxSeconds = max(maxSeconds, cache.checkpointSecondsOfCompute(c))

            time.sleep(2.0)
            simulation.triggerCheckpointGarbageCollection(False)
            time.sleep(2.0)

            checkpoints2 = cache.checkpointsForComputation(computations[0])

            maxSeconds2 = 0.0
            for c in checkpoints2:
                maxSeconds2 = max(maxSeconds, cache.checkpointSecondsOfCompute(c))

            self.assertTrue(len(checkpoints) > 2)
            self.assertEqual(len(checkpoints2), 2)

            #verify we didn't lose the farthest-ahead checkpoint
            self.assertEqual(maxSeconds, maxSeconds2)

        finally:
            simulation.teardown()

    def test_checkpointingGarbageCollectionGraph(self):
        simulation = self.createSimulation()
        self.assertTrue(len(simulation.objectStore.listValues()) == 0)

        try:
            #give the simulation a couple of seconds to pick a scheduler
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

            compId1 = simulation.submitComputation(expensiveChildCachecalls(0))
            compId2 = simulation.submitComputation(expensiveChildCachecalls(1))

            simulation.waitForGlobalScheduler()

            time.sleep(1.0)

            simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()
            self.waitForAllCheckpointsToClear(simulation)

            cache = simulation.getWorkerVdm(0).getPersistentCacheIndex()

            self.assertTrue(cache.totalComputationsInCache() > 0)
            self.assertEqual(cache.totalReachableComputationsInCache(), 0)

            cache.setScriptDependencies("script", HashNative.ImmutableTreeSetOfHash() + compId1.computationHash)

            self.assertTrue(cache.totalReachableComputationsInCache() > 0)
            self.assertTrue(cache.totalReachableComputationsInCache() < cache.totalComputationsInCache())

            cache.setScriptDependencies("script2", HashNative.ImmutableTreeSetOfHash() + compId2.computationHash)

            self.assertEqual(cache.totalComputationsInCache(), cache.totalReachableComputationsInCache())
        finally:
            simulation.teardown()

    def test_checkpointingGarbageCollectionCleansUpUnusedFiles(self):

        s3 = InMemoryS3Interface.InMemoryS3InterfaceFactory()

        simulation = self.createSimulation(s3Service = s3)

        prefix = simulation.objectStore.prefix
        s3().setKeyValue(
            simulation.objectStore.bucketName,
            simulation.objectStore.prefix + "a_file",
            "this is some data"
            )

        s3().setKeyValue(
            "not_the_simulation_object_store_bucket",
            simulation.objectStore.prefix + "a_file",
            "this is some data"
            )

        try:
            #give the simulation a couple of seconds to pick a scheduler
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

            simulation.submitComputation(expensiveChildCachecalls(0))

            time.sleep(1.0)

            simulation.triggerCheckpointGarbageCollection(False)

            time.sleep(1.0)

            simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()

            self.waitForAllCheckpointsToClear(simulation)

            self.assertFalse(
                s3().keyExists(
                    simulation.objectStore.bucketName,
                    simulation.objectStore.prefix + "a_file"
                    )
                )

            self.assertTrue(
                s3().keyExists(
                    "not_the_simulation_object_store_bucket",
                    simulation.objectStore.prefix + "a_file"
                    )
                )
        finally:
            simulation.teardown()

    def test_checkpointingAndGarbageCollectionSimultaneously(self):
        simulation = self.createSimulation()
        self.assertTrue(len(simulation.objectStore.listValues()) == 0)

        try:
            #give the simulation a couple of seconds to pick a scheduler
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

            simulation.submitComputation(sumInLoopText)

            def injector(*args):
                return .5
            simulation.s3Service.setDelayAfterWriteInjector(injector)

            t0 = time.time()

            timestamp = None

            simulation.waitForGlobalScheduler()
            time.sleep(1.0)

            count = 0
            timeOfLastCheckpoint = time.time()
            for passIx in range(5):
                simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()
                time.sleep(random.random() * random.random())
                simulation.triggerCheckpointGarbageCollection(False)

                self.waitForAllCheckpointsToClear(simulation)

                totalTime0 = self.totalTimeElapsedOfMostRecentCheckpoints(simulation)

                for ix in range(4):
                    simulation.dropTopWorker()
                for ix in range(4):
                    simulation.addWorker()

                simulation.waitForGlobalScheduler()
                simulation.getGlobalScheduler().setCheckpointStatusInterval(0.01)

                while len(self.timeElapsedOfMostRecentCheckpoints(simulation)) == 0:
                    time.sleep(0.01)

                totalTime1 = self.totalTimeElapsedOfMostRecentCheckpoints(simulation)

                self.assertTrue(totalTime1 >= totalTime0)
        finally:
            simulation.teardown()

    def test_checkpointingTimestampOnlyIncreases(self):
        simulation = self.createSimulation()
        
        try:
            #give the simulation a couple of seconds to pick a scheduler
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

            simulation.submitComputation(sumInLoopText)

            simulation.waitForGlobalScheduler()
            time.sleep(1.0)

            priorTotalTime = 0.0
            t0 = time.time()

            simulation.getGlobalScheduler().setCheckpointStatusInterval(0.01)
            while time.time() - t0 < 10.0:
                totalTime = self.totalTimeElapsedOfMostRecentCheckpoints(simulation)
                time.sleep(.1)

                self.assertTrue(totalTime >= priorTotalTime, (totalTime, priorTotalTime))
                priorTotalTime = totalTime
        finally:
            simulation.teardown()

    def test_checkpointingWithFaultyWrites(self):
        for injectionType in ["page","bigvec","checkpoint_slice","checkpoint_summary"]:
            print "starting test for injection type ", injectionType

            simulation = self.createSimulation(useHdfsObjectStore=False)
            self.assertTrue(len(simulation.objectStore.listValues()) == 0)

            injectedAnyFailures = [False]
            maxTimesToSee = {'page': 20, 'bigvec': 4, 'checkpoint_slice': 10, 'checkpoint_summary': 1}
            timesToSeeRemaining = [int(maxTimesToSee[injectionType] * random.random())]

            def injector(bucketname, keyname):
                if keyname.endswith(injectionType):
                    if timesToSeeRemaining[0] == 0:
                        timesToSeeRemaining[0] -= 1
                        injectedAnyFailures[0] = True

                        print "injecting a failure in ", injectionType
                        return True
                    else:
                        timesToSeeRemaining[0] -= 1

                return False

            simulation.s3Service.setWriteFailureInjector(injector)

            try:
                #give the simulation a couple of seconds to pick a scheduler
                self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

                simulation.submitComputation(vecOfVecCalcText)

                time.sleep(5.0)

                timestamp = None

                attempt = 0
                hasFailedCheckpoint = False

                while attempt < 10 and not hasFailedCheckpoint:
                    attempt += 1

                    simulation.getGlobalScheduler().triggerFullCheckpointsOnOutstandingComputations()

                    timestamp = self.waitForFullCheckpoint(simulation, priorCheckpoint=timestamp, onlySuccessful = False)

                    statuses = simulation.getGlobalScheduler().currentOutstandingCheckpointStatuses(True, True)

                    wasSuccessful = statuses[0][1][0].checkpointSuccessful

                    if not wasSuccessful:
                        hasFailedCheckpoint = True

                    if injectedAnyFailures[0]:
                        self.assertFalse(wasSuccessful)
                    else:
                        self.assertTrue(wasSuccessful)

                self.assertTrue(hasFailedCheckpoint)

            finally:
                simulation.teardown()

    def test_checkpointStatusSystem(self):
        simulation = self.createSimulation()

        try:
            self.assertTrue(simulation.waitForGlobalScheduler(timeout=2.0))

            simulation.getGlobalScheduler().setCheckpointStatusInterval(0.0001)

            simulation.submitComputation(simpleSumInLoopText)

            t0 = time.time()
            complete = False
            lastTimestamp = None
            timestamps = []

            while time.time() < t0 + 10 and not complete:
                if simulation.getAnyResult() is not None:
                    complete = True

                time.sleep(.001)

