def _finalize_manager(process, address, authkey, state, _Client):
     try:
         SyncManager._finalize_manager(process, address, authkey, state,
                                       _Client)
     except WindowsError:
         # http://stackoverflow.com/questions/17076679/windowserror-access-is-denied-on-calling-process-terminate
         pass

   def dconnect(self):
      """
      Attempts to connect to the DFS on the host/port for which the Node was initialized for.
      If no connection can be made, Node will keep attempting to connect until a connection
      can be established. Once a connection can be established the remove methods requested
      will be registered.
      """

      # remove connection from cache:
      # BaseProxy class has thread local storage which caches the connection
      # which is reused for future connections causing "borken pipe" errors on
      # creating new manager.
      if self.dfs in BaseProxy._address_to_local:
         if hasattr(BaseProxy._address_to_local[self.dfs][0], 'connection'):
            del BaseProxy._address_to_local[self.dfs][0].connection

      # register handlers
      SyncManager.register("get_nodes")

      print "connecting to dfs", self.dfs
      while self.alive:

         try:
            self.impd= SyncManager(address= self.dfs, authkey= self.dauthkey)
            self.impd.connect()
            print "connected to dfs", self.dfs
            break
         except (EOFError, IOError, SocketError) as e:
            print "could not connect ...trying again", str(e)
            sleep(1)

def main_proc():

    pid = os.getpid()
    # initialize manager
    mgr = SyncManager()
    mgr.start(mgr_init)
    
    try:
        # Create share object between processes
        shared_queue = mgr.Queue()

        # Create subprocesses
        put_proc = Process(target=put_data_proc, args=(shared_queue,))
        put_proc_1 = Process(target=put_data_proc_1, args=(shared_queue,))
        get_proc = Process(target=get_data_proc, args=(shared_queue,))

        # Start the processes
        put_proc.start()
        put_proc_1.start()
        get_proc.start()

        # Join the processes until they finished
        put_proc.join()
        put_proc_1.join()
        get_proc.join()

    except KeyboardInterrupt:
        print "Main process (pid=%s) was interruptted" % pid
    finally:
        mgr.shutdown()

 def _run_server(cls, *args):
     # make sure that the server ignores SIGINT (KeyboardInterrupt)
     signal.signal(signal.SIGINT, signal.SIG_IGN)
     # prevent connection errors to trigger exceptions
     try:
         SyncManager._run_server(*args)
     except socket.error:
         pass

 def __init__(self, args):
     # Init print management
     Print.init(log=args.log, debug=args.debug, all=args.all, cmd=args.prog)
     # Print command-line
     Print.command()
     self._process_color_arg(args)
     # Get project and related configuration
     self.project = args.project
     self.config_dir = args.i
     self.processes = args.max_processes if args.max_processes <= cpu_count() else cpu_count()
     self.use_pool = (self.processes != 1)
     self.lock = Lock()
     self.nbfiles = 0
     self.nbskip = 0
     self.nberrors = 0
     self.file_filter = []
     if args.include_file:
         self.file_filter.extend([(f, True) for f in args.include_file])
     else:
         # Default includes netCDF only
         self.file_filter.append(('^.*\.nc$', True))
     if args.exclude_file:
         # Default exclude hidden files
         self.file_filter.extend([(f, False) for f in args.exclude_file])
     else:
         self.file_filter.append(('^\..*$', False))
     self.dir_filter = args.ignore_dir
     # Init process manager
     if self.use_pool:
         manager = SyncManager()
         manager.start()
         Print.BUFFER = manager.Value(c_char_p, '')
         self.progress = manager.Value('i', 0)
     else:
         self.progress = Value('i', 0)
     self.tunits_default = None
     if self.project in DEFAULT_TIME_UNITS.keys():
         self.tunits_default = DEFAULT_TIME_UNITS[self.project]
     # Change frequency increment
     if args.set_inc:
         for table, frequency, increment, units in args.set_inc:
             if table not in set(zip(*FREQ_INC.keys())[0]):
                 raise InvalidTable(table)
             if frequency not in set(zip(*FREQ_INC.keys())[1]):
                 raise InvalidFrequency(frequency)
             keys = [(table, frequency)]
             if table == 'all':
                 keys = [k for k in FREQ_INC.keys() if k[1] == frequency]
             if frequency == 'all':
                 keys = [k for k in FREQ_INC.keys() if k[0] == table]
             for key in keys:
                 FREQ_INC[key] = [float(increment), str(units)]
     # Get reference time properties if submitted
     # Default is to deduce them from first file scanned
     self.ref_calendar = args.calendar
     self.ref_units = args.units
     # Init collector
     self.sources = None

    def __init__(self, queue_size=None):
        #print "instantiating process manager"
        SyncManager.__init__(self)

        self.start()

        self.ff = None
        self.input_queue = self.Queue(queue_size)
        self.output_queue = self.Queue(queue_size)
        self.worker = None

 def __init__(self, *args, **kwargs):
     # init ingestor process, create tweet queue
     manager = SyncManager()
     manager.start(mgr_init)
     self.tweet_queue = manager.Queue()
     self.ingestion_process = multiprocessing.Process(target=do_ingestion, args=(self.tweet_queue,))
     self.ingestion_process.start()
     
     # call superclass init
     tweepy.StreamListener.__init__(self, *args, **kwargs)

   def connect(self):
      """
      Attempts to connect to the Queue on the host/port for which the DFS was initialized for.
      If no connection can be made, DFS will keep attempting to connect until a connection
      can be established.  One connection is established the remove methods requested will be
      registered.
      """

      # remove connection from cache:
      # BaseProxy class has thread local storage which caches the connection
      # which is reused for future connections causing "borken pipe" errors on
      # creating new manager.
      if self.queue in BaseProxy._address_to_local:
         del BaseProxy._address_to_local[self.queue][0].connection

      # register handlers
      SyncManager.register("get_streams")
      SyncManager.register("get_queue")
      SyncManager.register("get_store")
      SyncManager.register("get_properties")

      print "connecting to queue", self.queue
      while self.alive:

         try:
            self.impq= SyncManager(address= self.queue, authkey= self.qauthkey)
            self.impq.connect()
            break
         except (EOFError, IOError, SocketError) as e:
            print "could not connect ...trying again", str(e)
            sleep(1)

def Manager():
    '''
    Returns a manager associated with a running server process

    The managers methods such as `Lock()`, `Condition()` and `Queue()`
    can be used to create shared objects.
    '''
    from multiprocessing.managers import SyncManager
    m = SyncManager()
    m.start()
    return m

    def __init__(self, servo_id):
        self.servo_id = servo_id
        self.angle = Value('f', 0.0)
        self.stop_signal = Value('b', False)

        # http://jtushman.github.io/blog/2014/01/14/python-|-multiprocessing-and-interrupts/
        manager = SyncManager() # instead of regular Manager because we want to ignore kb interrupt
        manager.start(Servo.init_mgr) # start the manager explicitly
        self.command_queue = manager.list([])
        self.current_command = manager.dict()

        self.finished = Value('b', False)

  def _finalize_manager(process, *args, **kwargs):
    """Shutdown the manager process."""

    def _join(functor, *args, **kwargs):
      timeout = kwargs.get('timeout')
      if not timeout is None and timeout < 1:
        kwargs['timeout'] = 1

      functor(*args, **kwargs)

    process.join = functools.partial(_join, process.join)
    SyncManager._finalize_manager(process, *args, **kwargs)

class DataSender:

    def __init__(self,phantfile):
        try:
            self.phant = json.load(open(phantfile, 'r'))
        except IOError:
            raise ValueError("Invalid phantfile location")
        self.running = True
        self._manager = SyncManager()


    def start(self):
        self._manager.start(self._mgr_init)
        self._que = self._manager.Queue()
        self._process = Process(target=self.up, args=(self._que,))
        self._process.start()

    def _mgr_init(self):
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        print("initialized manager")

    def up(self,que):
        
        def stop(val,val2):
            print "process SIGINT stopping"
            self.running = False

        signal.signal(signal.SIGINT, stop)
        print('datauploader started')
        while self.running or not que.empty():
            item = json.loads(que.get(True))
            print("handling item={0}".format(item))
            self.httpsend(item)
            que.task_done()
            time.sleep(2)
        print("datauploader process terminating...")

    def send(self, data):
        self._que.put(data)

    def httpsend(self, data):
        postdata = urllib.urlencode(data)
        headers = {'Phant-Private-Key': self.phant['privateKey'] }
        req = urllib2.Request(self.phant['inputUrl'], postdata, headers)
        res = urllib2.urlopen(req)
        content = res.read()
        print("response: {0}".format(content))
    
    def stop(self):
        print("shutting down sender")
        self.running = False
        self._que.join()
        self._process.terminate()

def get_binary_matrix_from_service(q):
    global matrix_service
    if matrix_service == None:
        matrices = dict()
        matrix_service = SyncManager(address=("localhost", 50000), authkey="")
        SyncManager.register("get_matrix", lambda q: get_matrix(q, matrices))
        Process(target=lambda: matrix_service.get_server().serve_forever()).start()
    SyncManager.register("get_matrix")
    matrix_service.connect()
    return matrix_service.get_matrix(q)

   def connect(self):

      # register with Queue
      SyncManager.register('getPipeline')
      SyncManager.register('getStore')
      self.qInstance= self.opts.qInstance
      (self.qHost, self.qPort, self.qKey)= self.opts.queue.split(':')
      queue= SyncManager(address= (self.qHost, int(self.qPort)), authkey= self.qKey)
      queue.connect()
      self.pipeline= queue.getPipeline()
      self.store= queue.getStore()

def init_good_sync_manager():
    from multiprocessing.managers import SyncManager
    #handle SIGINT from SyncManager object
    def mgr_sig_handler(signal, frame):
        print 'not closing the mgr'

    #initilizer for SyncManager
    def mgr_init():
        import signal
        signal.signal(signal.SIGINT, mgr_sig_handler)
        print 'initialized mananger'

    #using syncmanager directly instead of letting Manager() do it for me
    manager = SyncManager()
    manager.start(mgr_init)

def get_server_queue():
    #FIXME: some OSX users were getting "Can't assign requested address" errors
    # if we use socket.gethostname() for the address. Changing it to
    # 'localhost' seems to fix the issue, but I don't know why. We had to
    # use socket.gethostname() in order to get our benchmark tests to run
    # using qsub on a linux cluster, so with this 'fix', testflo benchmark tests
    # will likely not work on a cluster of OSX machines.
    if sys.platform == 'darwin':
        addr = 'localhost'
    else:
        addr = socket.gethostname()

    manager = SyncManager(address=(addr, 0), authkey=_testflo_authkey)
    manager.start()
    return manager, manager.Queue()

class Downloader(object):
    def __init__(self, timeout=30, retries=100, wait=1):
        self.timeout = timeout
        self.retries = retries
        self.wait = wait
        
        self.manager = SyncManager()
        self.manager.start()
        
    def retry_fetch_data(self, url):
        market_data = self.fetch_data(url)
        
        retries = 1
        while not market_data and retries < self.retries:
            print "Retry #%s..." % str(retries)
            market_data = self.fetch_data(url)
            if market_data:
                print "Fetched: " + str(len(market_data))
            else:
                print "Fetched nothing!"
            retries += 1
        
        return market_data
    
    def fetch_data(self, url):
        limit = 60
        msg = "Downloading " + url[0: min(limit, len(url))] 
        if len(url) > limit:
            msg += "(+" + str(len(url) - limit) + ")"
        print msg
            
        return_dict = self.manager.dict()
        self.job = Process(target=get_page_data, args=(url, return_dict))
        self.job.start()
        
        self.job.join(self.timeout)
        if self.job.is_alive():
            self.job.terminate()
        self.job = None
        
        market_data = None
        if 'page' in return_dict:
            market_data = return_dict['page']
        
        if self.wait > 0:
            time.sleep(self.wait)
        
        return market_data

    def __int__(self, name='default', address=None, authkey=None):
        """
        This is the default constructor for the class.

        :param name:    The manager name
        :param address: The address of the server
        :param authkey: The auth key
        :return:
        """

        # Set internals
        self.__name     = name

        # Override the manager
        SyncManager.__init__(address, authkey)
        return

   def __init__(self, address, authkey, taskdir= "tasks", id= None, **properties):
      """Creates a stream and retrieves the streams priority queue and data-store."""

      self.id= id if id else str(uuid1())
      self.ipaddress= getipaddress()

      self.address= address
      self.taskdir= path.join(taskdir, self.id)
      self.properties= properties

      self.impq= SyncManager(address= self.address, authkey= authkey)
      self.impq.register("get_streams")
      self.impq.register("create_stream")
      self.impq.register("delete_stream")
      self.impq.register("get_store")
      self.impq.register("get_queue")
      self.impq.connect()

      self.jobs= []
      self.impq.create_stream(id= self.id, ipaddress= self.ipaddress, **properties)
      self.store= self.impq.get_store(id= self.id)
      self.queue= self.impq.get_queue(id= self.id)
      self.alive= True
      self._current_thread= None
      self._lock= Lock()
      self.threads= []
      self.errors= {}
      self.ready= {}
      self._progress= {}


      try:
         makedirs(self.taskdir)
      except:
         pass

	def __init__(self, cookie_file, url_queue_size, pg_queue_size, nr_downloadprocess, nr_parserprocess):
		super(SpiderEngine, self).__init__()

		self.logger = logging.getLogger(self.__class__.__name__)

		self.multiprocess_manager = SyncManager()#SyncManager(('',58585))
		self.multiprocess_manager.start()

		self.lck4urlq=self.multiprocess_manager.Lock()
		self.lck4pageq=self.multiprocess_manager.Lock()
		# event for suprocess to initiative exit.
		self.shutdown=self.multiprocess_manager.Event()

		self.url_queue=Queue(url_queue_size)
		self.page_queue=Queue(pg_queue_size)
		self.url_hist=self.multiprocess_manager.dict()
		self.urls= UrlScheduler(self.url_queue, self.url_hist, self.lck4urlq)

		# init multiprocess log
		self.mlog=get_logger()
		mhandler=logging.StreamHandler()
		mhandler.setFormatter(logging.Formatter('%(processName)s %(funcName)s() | %(message)s', '%H:%M:%S'))
		self.mlog.addHandler(mhandler)
		self.mlog.setLevel(logging.INFO)

		self.pages= PageScheduler(self.urls, self.page_queue, self.lck4pageq)
		self.downloader= PageDownloader(cookie_file, self.urls, self.pages, self.shutdown, self.multiprocess_manager, nr_downloadprocess, self.mlog)
		self.parser=PageParser(self.urls, self.pages, self.shutdown, self.multiprocess_manager, nr_parserprocess, self.mlog)