<s> import os <EOL> import logging <EOL> import time <EOL> import socket <EOL> import shutil <EOL> import uuid <EOL> import zmq <EOL> from dpark import util <EOL> import dpark . conf as conf <EOL> logger = logging . getLogger ( __name__ ) <EOL> class DparkEnv : <EOL> environ = { } <EOL> @ classmethod <EOL> def register ( cls , name , value ) : <EOL> cls . environ [ name ] = value <EOL> @ classmethod <EOL> def get ( cls , name , default = None ) : <EOL> return cls . environ . get ( name , default ) <EOL> def __init__ ( self ) : <EOL> self . started = False <EOL> def start ( self , isMaster , environ = { } ) : <EOL> if self . started : <EOL> return <EOL> logger . debug ( "<STR_LIT>" , os . getpid ( ) , isMaster , environ ) <EOL> self . isMaster = isMaster <EOL> if isMaster : <EOL> roots = conf . DPARK_WORK_DIR <EOL> if isinstance ( roots , str ) : <EOL> roots = roots . split ( '<STR_LIT:U+002C>' ) <EOL> name = '<STR_LIT>' % ( socket . gethostname ( ) , uuid . uuid4 ( ) ) <EOL> self . workdir = [ os . path . join ( root , name ) for root in roots ] <EOL> try : <EOL> for d in self . workdir : <EOL> util . mkdir_p ( d ) <EOL> except OSError as e : <EOL> if environ . get ( '<STR_LIT>' , False ) : <EOL> raise e <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' + self . workdir [ <NUM_LIT:0> ] <EOL> self . environ [ '<STR_LIT>' ] = self . workdir <EOL> self . environ [ '<STR_LIT>' ] = util . COMPRESS <EOL> else : <EOL> self . environ . update ( environ ) <EOL> if self . environ [ '<STR_LIT>' ] != util . COMPRESS : <EOL> raise Exception ( "<STR_LIT>" % self . environ [ '<STR_LIT>' ] ) <EOL> self . ctx = zmq . Context ( ) <EOL> from dpark . tracker import TrackerServer , TrackerClient <EOL> if isMaster : <EOL> self . trackerServer = TrackerServer ( ) <EOL> self . trackerServer . start ( ) <EOL> addr = self . trackerServer . addr <EOL> env . register ( '<STR_LIT>' , addr ) <EOL> else : <EOL> addr = env . get ( '<STR_LIT>' ) <EOL> self . trackerClient = TrackerClient ( addr ) <EOL> from dpark . cache import CacheTracker <EOL> self . cacheTracker = CacheTracker ( ) <EOL> from dpark . shuffle import LocalFileShuffle , MapOutputTracker <EOL> LocalFileShuffle . initialize ( isMaster ) <EOL> self . mapOutputTracker = MapOutputTracker ( ) <EOL> from dpark . shuffle import ParallelShuffleFetcher <EOL> self . shuffleFetcher = ParallelShuffleFetcher ( <NUM_LIT:2> ) <EOL> from dpark . broadcast import start_manager <EOL> start_manager ( isMaster ) <EOL> self . started = True <EOL> logger . debug ( "<STR_LIT>" ) <EOL> def stop ( self ) : <EOL> if not getattr ( self , '<STR_LIT>' , False ) : <EOL> return <EOL> logger . debug ( "<STR_LIT>" , os . getpid ( ) ) <EOL> self . shuffleFetcher . stop ( ) <EOL> self . cacheTracker . stop ( ) <EOL> self . mapOutputTracker . stop ( ) <EOL> if self . isMaster : <EOL> self . trackerServer . stop ( ) <EOL> from dpark . broadcast import stop_manager <EOL> stop_manager ( ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> for d in self . workdir : <EOL> shutil . rmtree ( d , True ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> self . started = False <EOL> env = DparkEnv ( ) </s>
