def __init__(self):
        callbackSchedulerFactory = CallbackScheduler.createSimpleCallbackSchedulerFactory()
        self.callbackScheduler = callbackSchedulerFactory.createScheduler("Simulator", 1)

        self.uforaPath = os.path.abspath(os.path.join(os.path.dirname(__file__), '../'))

        self.sharedStatePath = os.path.join(self.uforaPath, 'distributed/SharedState')
        self.sharedStateMainline = os.path.join(self.sharedStatePath, 'sharedStateMainline.py')

        self.gatewayServiceMainline = os.path.join(self.uforaPath, 'scripts/init/ufora-gateway.py')

        self.webPath = os.path.join(self.uforaPath, 'web/relay')
        self.relayScript = os.path.join(self.webPath, 'server.coffee')

        self.relayPort = Setup.config().relayPort
        self.relayHttpsPort = Setup.config().relayHttpsPort
        self.sharedStatePort = Setup.config().sharedStatePort
        self.restApiPort = Setup.config().restApiPort
        self.subscribableWebObjectsPort = Setup.config().subscribableWebObjectsPort

        #create an OutOfProcessDownloader so we can execute commands like 'forever'
        #from there, instead of forking from the main process (which can run out of memory)
        self.processPool = OutOfProcessDownloader.OutOfProcessDownloaderPool(1)


        self.desirePublisher = None
        self._connectionManager = None

def constructVDM(
        callbackScheduler,
        vectorRamCacheBytes = None,
        maxRamCacheBytes = None,
        maxVectorChunkSize = None
        ):
    if vectorRamCacheBytes is None:
        vectorRamCacheBytes = Setup.config().cumulusVectorRamCacheMB * 1024 * 1024

    if maxRamCacheBytes is None:
        maxRamCacheBytes = Setup.config().cumulusMaxRamCacheMB * 1024 * 1024

    if maxVectorChunkSize is None:
        maxVectorChunkSize = Setup.config().maxPageSizeInBytes

        if maxVectorChunkSize > vectorRamCacheBytes / 32:
            logging.info(
                "VDM constructor specified a chunk size of %s MB " +
                "and a memory size of %s MB. Reducing the chunk size because its too large",
                vectorRamCacheBytes / 1024.0 / 1024.0,
                maxVectorChunkSize / 1024.0 / 1024.0
                )

            maxVectorChunkSize = vectorRamCacheBytes / 32

    logging.info("Creating a VDM with %s MB of memory and %s max vector size",
        vectorRamCacheBytes / 1024.0 / 1024.0,
        maxVectorChunkSize / 1024.0 / 1024.0
        )

    VDM = FORANative.VectorDataManager(callbackScheduler, maxVectorChunkSize)
    VDM.setMemoryLimit(vectorRamCacheBytes, maxRamCacheBytes)

    return VDM

    def createServiceAndServiceThread(self):
        config = Setup.config()
        config.cumulusMaxRamCacheMB = self.cumulusMaxRamCacheSizeOverride
        config.cumulusVectorRamCacheMB = self.cumulusVectorRamCacheSizeOverride
        config.cumulusServiceThreadCount = self.cumulusThreadCountOverride
        config.cumulusDiskCacheStorageSubdirectory = str(uuid.uuid4())

        ownAddress = str(uuid.uuid4())
        callbackScheduler = self.callbackSchedulerFactory.createScheduler(
            "InMemoryClusterChild",
            1)
        channelListener = self.createMultiChannelListener(
            callbackScheduler,
            [Setup.config().cumulusControlPort, Setup.config().cumulusDataPort],
            ownAddress)
        service = CumulusService.CumulusService(
            ownAddress=ownAddress,
            channelListener=channelListener,
            channelFactory=self.channelManager.createChannelFactory(),
            eventHandler=CumulusNative.CumulusWorkerHoldEventsInMemoryEventHandler(),
            callbackScheduler=callbackScheduler,
            diagnosticsDir=None,
            config=config,
            viewFactory=self.sharedStateViewFactory
            )
        service.startService(lambda: None)
        return service

def UserFacingMainline(main, argv, modulesToInitialize=None, parser=None):
    """Helper function that initializes some modules and then calls main.

    Used to centralize error handling for common initialization routines and to set up the
    initial component hosts.
    """
    if parser is None:
        parser = Setup.defaultParser()

    setup = Setup.defaultSetup()

    parsedArguments = parser.parse_args(argv[1:])
    setup.processArgs(parsedArguments)

    setup.config.configureLoggingForUserProgram()

    with Setup.PushSetup(setup):
        initializeModules(modulesToInitialize)

        result = main(parsedArguments)

        if result is None:
            result = 0

        sys.stdout.flush()
        sys.stderr.flush()

        os._exit(result)

    def startSharedState(self):
        cacheDir = Setup.config().getConfigValue(
            "SHARED_STATE_CACHE",
            os.path.join(Setup.config().fakeAwsBaseDir, 'ss_cache')
            )

        logging.info("Starting shared state with cache dir '%s' and log file '%s'",
                     cacheDir,
                     self.sharedStateLogFile)

        with DirectoryScope.DirectoryScope(self.sharedStatePath):
            args = ['forever',
                    '--killSignal', 'SIGTERM',
                    '-l', self.sharedStateLogFile,
                    'start',
                    '-c', 'python', self.sharedStateMainline,
                    '--cacheDir', cacheDir,
                    '--logging', 'info'
                   ]

            def sharedStateStdout(msg):
                logging.info("SHARED STATE OUT> %s", msg)
            def sharedStateStderr(msg):
                logging.info("SHARED STATE ERR> %s", msg)

            startSharedState = SubprocessRunner.SubprocessRunner(
                args,
                sharedStateStdout,
                sharedStateStderr,
                dict(os.environ)
                )
            startSharedState.start()
            startSharedState.wait(60.0)
            startSharedState.stop()

    def __init__(self, vdm, offlineCache):
        Stoppable.Stoppable.__init__(self)
        self.dependencies_ = TwoWaySetMap.TwoWaySetMap()
        self.vdm_ = vdm
        self.offlineCache_ = offlineCache
        self.finishedValues_ = {}
        self.intermediates_ = {}
        self.lock_ = threading.RLock()
        self.completable_ = Queue.Queue()
        self.timesComputed = 0
        self.computingContexts_ = {}
        self.computingContexts_t0_ = {}
        self.isSplit_ = set()
        self.watchers_ = {}
        self.contexts_ = []

        self.inProcessDownloader = (
            OutOfProcessDownloader.OutOfProcessDownloaderPool(
                Setup.config().cumulusServiceThreadCount,
                actuallyRunOutOfProcess = False
                )
            )

        self.threads_ = []
        self.isActive = True
        #setup the primary cache object, and set its worker threads going
        for threadIx in range(Setup.config().cumulusServiceThreadCount):
            workerThread = ManagedThread.ManagedThread(target = self.threadWorker)
            workerThread.start()
            self.threads_.append(workerThread)

    def __init__(self, callbackScheduler, cachePathOverride=None, port=None):
        self.callbackScheduler = callbackScheduler
        port = Setup.config().sharedStatePort
        logging.info("Initializing SharedStateService with port = %s", port)

        self.cachePath = cachePathOverride if cachePathOverride is not None else \
                         Setup.config().sharedStateCache

        if self.cachePath != '' and not os.path.exists(self.cachePath):
            os.makedirs(self.cachePath)

        CloudService.Service.__init__(self)
        self.socketServer = SimpleServer.SimpleServer(port)
        self.keyspaceManager = KeyspaceManager(
            0,
            1,
            pingInterval=120,
            cachePathOverride=cachePathOverride
            )


        self.socketServer._onConnect = self.onConnect
        self.socketServerThread = ManagedThread.ManagedThread(target=self.socketServer.start)
        self.logfilePruneThread = ManagedThread.ManagedThread(target=self.logFilePruner)

        self.stoppedFlag = threading.Event()

 def generateTestConfigFileBody_(self):
     return ("ROOT_DATA_DIR = %s\n"
             "BASE_PORT = %s\n"
             "FORA_MAX_MEM_MB = %s\n"
             ) % (
             Setup.config().rootDataDir,
             Setup.config().basePort,
             "10000" if multiprocessing.cpu_count() <= 8 else "60000"
             )

def runPythonUnitTests_(args, testFilter, testGroupName, testFiles):
    testArgs = ["dummy"]

    if args.testHarnessVerbose or args.list:
        testArgs.append('--nocaptureall')

    testArgs.append('--verbosity=0')

    if not args.list:
        print "Executing %s unit tests." % testGroupName

    Setup.config().configureLoggingForUserProgram()

    parser = PythonTestArgumentParser()
    filterActions = parser.parse_args(args.remainder)

    bsaRootDir = os.path.split(ufora.__file__)[0]

    testCasesToRun = []

    plugins = nose.plugins.manager.PluginManager([OutputCaptureNosePlugin()])

    config = nose.config.Config(plugins=plugins)
    config.configure(testArgs)
    for i in range(args.copies):
        testCases = UnitTestCommon.loadTestCases(config, testFiles, bsaRootDir, 'ufora')
        if filterActions:
            testCases = applyFilterActions(filterActions, testCases)

        testCasesToRun += testCases

    if testFilter is not None:
        testCasesToRun = testFilter(testCasesToRun)

    if args.list:
        for test in testCasesToRun:
            print test.id()

        os._exit(0)

    if args.random:
        import random
        random.shuffle(testCasesToRun)

    if args.pythreadcheck:
        results = {}
        for test in testCasesToRun:
            results[test] = runPyTestSuite(config, None, unittest.TestSuite([test]), testArgs)

        return True in results.values()
    else:
        testFiles = '.'
        return runPyTestSuite(config, None, testCasesToRun, testArgs)

def createService(args):
    callbackSchedulerFactory = CallbackScheduler.createSimpleCallbackSchedulerFactory()
    callbackScheduler = callbackSchedulerFactory.createScheduler('ufora-worker', 1)
    channelListener = MultiChannelListener(callbackScheduler,
                                           [args.base_port, args.base_port + 1])

    sharedStateViewFactory = ViewFactory.ViewFactory.TcpViewFactory(
        callbackSchedulerFactory.createScheduler('SharedState', 1),
        args.manager_address,
        int(args.manager_port)
        )

    channelFactory = TcpChannelFactory.TcpStringChannelFactory(callbackScheduler)

    diagnostics_dir = os.getenv("UFORA_WORKER_DIAGNOSTICS_DIR")
    eventHandler = diagnostics_dir and createEventHandler(
        diagnostics_dir,
        callbackSchedulerFactory.createScheduler("ufora-worker-event-handler", 1)
        )

    own_address = args.own_address or get_own_ip()
    print "Listening on", own_address, "ports:", args.base_port, "and", args.base_port+1

    config = Setup.config()
    print "RAM cache of %d / %d MB and %d threads. Track tcmalloc: %s" % (
        config.cumulusVectorRamCacheMB,
        config.cumulusMaxRamCacheMB,
        config.cumulusServiceThreadCount,
        config.cumulusTrackTcmalloc
        )

    print "Ufora store at %s:%s" % (args.manager_address, args.manager_port)

    s3InterfaceFactory = ActualS3Interface.ActualS3InterfaceFactory()
    print "PythonIoTasks threads: %d. Out of process: %s" % (
        config.externalDatasetLoaderServiceThreads,
        s3InterfaceFactory.isCompatibleWithOutOfProcessDownloadPool
        )

    return CumulusService.CumulusService(
        own_address,
        channelListener,
        channelFactory,
        eventHandler,
        callbackScheduler,
        diagnostics_dir,
        Setup.config(),
        viewFactory=sharedStateViewFactory,
        s3InterfaceFactory=s3InterfaceFactory,
        objectStore=NullObjectStore.NullObjectStore()
        )

def parseS3Dataset(s3InterfaceFactory, s3Dataset):
    """Log in to amazon S3 and return an appropriate s3Interface and a bucket/keypair"""
    if s3Dataset.isInternal():
        #use the internal login. This should have access only to our one internal bucket
        return (
            s3InterfaceFactory(),
            Setup.config().userDataS3Bucket,
            s3Dataset.asInternal.keyname
            )

    elif s3Dataset.isExternal():
        asE = s3Dataset.asExternal

        if asE.awsAccessKey != "":
            try:
                interface = s3InterfaceFactory(
                    asE.awsAccessKey,
                    asE.awsSecretKey
                    )
            except:
                raise InvalidDatasetException("Failed to log into S3 with given credentials")
        else:
            interface = s3InterfaceFactory()

        return (interface, asE.bucket, asE.key)
    else:
        raise DatasetLoadException("Unknown dataset type")

    def test_teardown(self):
        harness = SharedStateTestHarness.SharedStateTestHarness(False,
                                                                port=Setup.config().sharedStatePort)

        view = harness.newView()
        view.teardown()
        harness.teardown()

    def test_teardown_simple(self):
        vdm = FORANative.VectorDataManager(callbackScheduler, Setup.config().maxPageSizeInBytes)
        context = ExecutionContext.ExecutionContext(dataManager = vdm)

        context.evaluate(
            FORA.extractImplValContainer(
                FORA.eval("fun(){nothing}")
                ),
            FORANative.symbol_Call
            )

        context.getFinishedResult()

        toEval = FORA.extractImplValContainer(
            FORA.eval(
                """fun() {
                    let f = fun() { };
                    let v = [1, [3]];
                    cached(f())
                    }"""
                )
            )

        context.evaluate(toEval, FORANative.symbol_Call)
        while not context.isCacheRequest():
            context.resume()

        context.teardown(True)

    def test_teardown_during_vector_load(self):
        vdm = FORANative.VectorDataManager(callbackScheduler, Setup.config().maxPageSizeInBytes)
        context = ExecutionContext.ExecutionContext(dataManager = vdm)

        context.evaluate(
                FORA.extractImplValContainer(
                    FORA.eval("fun() { let v = [1,2,3].paged; fun() { v[1] } }")
                    ),
                FORANative.symbol_Call
                )
        vdm.unloadAllPossible()

        pagedVecAccessFun = context.getFinishedResult().asResult.result

        context.teardown()

        context.evaluate(
            pagedVecAccessFun,
            FORANative.symbol_Call
            )

        self.assertFalse(context.isInterrupted())
        self.assertTrue(context.isVectorLoad())

        context.teardown()

    def test_serialize_while_holding_interior_vector(self):
        vdm = FORANative.VectorDataManager(callbackScheduler, Setup.config().maxPageSizeInBytes)
        context = ExecutionContext.ExecutionContext(dataManager = vdm, allowInterpreterTracing=False)

        context.evaluate(
            FORA.extractImplValContainer(
                FORA.eval("""
                    fun() {
                        let v = [[1].paged].paged;
                        let v2 = v[0]

                        `TriggerInterruptForTesting()

                        1+2+3+v+v2
                        }"""
                    )
                ),
            FORANative.symbol_Call
            )

        self.assertTrue(context.isInterrupted())

        serialized = context.serialize()

        context = None

def createViewFactory():
    callbackSchedulerFactory = CallbackScheduler.createSimpleCallbackSchedulerFactory()
    return ViewFactory.ViewFactory.TcpViewFactory(
        callbackSchedulerFactory.createScheduler('fora-interpreter', 1),
        'localhost',
        Setup.config().sharedStatePort
        )

    def __init__(self, relayHostname, relayHttpsPort = None, messageDelayInSeconds = None):
        """Initialize a PipeTransport.

        messageDelayInSeconds - if not None, then all messages will be delayed by this many
            seconds before being pumped into the receiving channel. This can simulate
            delays talking over the internet.
        """
        self.onMessageReceived = None
        self.onDisconnected = None
        self.inputLoopThread = None
        self.isShuttingDown = False
        self.proxyProcess = None
        self.isConnected = False
        self.messageDelayInSeconds = messageDelayInSeconds
        self.messagePumpThread = None
        self.messagePumpQueue = Queue.Queue()

        self.relayHostname = relayHostname
        if relayHttpsPort:
            self.relayHttpsPort = relayHttpsPort
        else:
            self.relayHttpsPort = Setup.config().relayHttpsPort

        self.proxyStdIn = None
        self.proxyStdOut = None
        self.proxyStdErr = None
        self.proxyOutputThread = None

        logging.info("PipeTransport created for host %s:%s", self.relayHostname, self.relayHttpsPort)

    def test_refcountsInCompiledCode(self):
        vdm = FORANative.VectorDataManager(callbackScheduler, Setup.config().maxPageSizeInBytes)

        context = ExecutionContext.ExecutionContext(
            dataManager = vdm,
            allowInterpreterTracing = True,
            blockUntilTracesAreCompiled = True
            )

        text = """fun(){
        let f = fun(v, depth) {
            if (depth > 100)
                //this will trigger an interrupt since the data cannot exist in the VDM
                datasets.s3('','')
            else
                f(v, depth+1)
            }

        f([1,2,3,4,5], 0)
        }"""

        context.evaluate(
            FORA.extractImplValContainer(FORA.eval(text)),
            FORANative.symbol_Call
            )

        stacktraceText = context.extractCurrentTextStacktrace()

        self.assertTrue(stacktraceText.count("Vector") < 10)

def createService(args):
    callbackSchedulerFactory = CallbackScheduler.createSimpleCallbackSchedulerFactory()
    callbackScheduler = callbackSchedulerFactory.createScheduler('ufora-worker', 1)
    channelListener = MultiChannelListener(callbackScheduler,
                                           [args.base_port, args.base_port + 1])

    sharedStateViewFactory = ViewFactory.ViewFactory.TcpViewFactory(
        callbackSchedulerFactory.createScheduler('SharedState', 1),
        args.manager_address,
        int(args.manager_port)
        )

    channelFactory = TcpChannelFactory.TcpStringChannelFactory(callbackScheduler)

    diagnostics_dir = os.getenv("UFORA_WORKER_DIAGNOSTICS_DIR")
    eventHandler = diagnostics_dir and createEventHandler(
        diagnostics_dir,
        callbackSchedulerFactory.createScheduler("ufora-worker-event-handler", 1)
        )

    own_address = args.own_address or get_own_ip()
    print "Listening on", own_address, "ports:", args.base_port, "and", args.base_port+1

    return CumulusService.CumulusService(
        own_address,
        channelListener,
        channelFactory,
        eventHandler,
        callbackScheduler,
        diagnostics_dir,
        Setup.config(),
        viewFactory=sharedStateViewFactory
        )

def defaultLocalEvaluator(remoteEvaluator=None, vdmOverride=None):
    return LocalEvaluator(
        lambda vdm: None,
        Setup.config().cumulusVectorRamCacheMB * 1024 * 1024,
        remoteEvaluator,
        vdmOverride=vdmOverride
        )