def queue_callback(self, zh, rc, data):
   if zookeeper.OK == rc:
     try:
       for child in sorted(data):
         action = self.actionNode + '/' + child
         workLog = self.actionNode + '/' + child + '/worklog'
         statusLog = self.statusNode + '/status-'
         """ Launch the task if the task has not been executed """
         if zookeeper.exists(zh, workLog, None) == None:
           self.launch(zh, workLog, action, statusLog)
         else:
           """ If task has been previous launched, check for partial execution """
           buffer = zookeeper.get(zh, workLog, 0)
           state = simplejson.loads(buffer[0])
           """ If task is incompleted in execution, launch again """
           if 'status' in state and state['status'] == 'STARTING':
             logger.info('Relaunch '+child)
             self.launch(zh, workLog, action, statusLog)
           else:
             """ If the task has been launched, and completed, update status queue """
             if zookeeper.exists(zh, statusLog, None) == None:
               logger.info('Update status.')
               self.update(zh, statusLog, state)
     except NoNodeException, err:
       """ Skip no node exception """
     except Exception, err:
       logger.exception(err)

  def notifyFailedTransaction(self, app_id, txid):
    """ Notify failed transaction id.

    This method will add the transaction id into black list.
    After this call, the transaction becomes invalid.
    """
    self.__waitForConnect()
    self.checkTransaction(app_id, txid)
    print "notify failed transaction app:%s, txid:%d" % (app_id, txid)
    txpath = self.__getTransactionPath(app_id, txid)
    lockpath = None
    try:
      lockpath = zookeeper.get(self.handle, PATH_SEPARATOR.join([txpath, TX_LOCK_PATH]), None)[0]
    except zookeeper.NoNodeException:
      # there is no lock. it means there is no need to rollback.
      pass

    if lockpath:
      # add transacion id to black list.
      now = str(time.time())
      broot = self.__getBlacklistRootPath(app_id)
      if not zookeeper.exists(self.handle, broot):
        self.__forceCreatePath(broot)
      zookeeper.acreate(self.handle, PATH_SEPARATOR.join([broot, str(txid)]), now, ZOO_ACL_OPEN)
      # update local cache before notification
      if app_id in self.blacklistcache:
        with self.blacklistcv:
          self.blacklistcache[app_id].add(str(txid))
      # copy valid transaction id for each updated key into valid list.
      for child in zookeeper.get_children(self.handle, txpath):
        if re.match("^" + TX_UPDATEDKEY_PREFIX, child):
          value = zookeeper.get(self.handle, PATH_SEPARATOR.join([txpath, child]), None)[0]
          valuelist = value.split(PATH_SEPARATOR)
          key = urllib.unquote_plus(valuelist[0])
          vid = valuelist[1]
          vtxroot = self.__getValidTransactionRootPath(app_id)
          if not zookeeper.exists(self.handle, vtxroot):
            self.__forceCreatePath(vtxroot)
          vtxpath = self.__getValidTransactionPath(app_id, key)
          zookeeper.acreate(self.handle, vtxpath, vid, ZOO_ACL_OPEN)
      # release the lock
      try:
        zookeeper.adelete(self.handle, lockpath)
      except zookeeper.NoNodeException:
        # this should be retry.
        pass

    # just remove transaction node
    try:
      for item in zookeeper.get_children(self.handle, txpath):
        zookeeper.adelete(self.handle, PATH_SEPARATOR.join([txpath, item]))
      zookeeper.adelete(self.handle, txpath)
    except zookeeper.NoNodeException:
      # something wrong. next GC will take care of it.
      return False

    return True

    def write(self, path, contents, ephemeral=False, exclusive=False):
        """ 
        Writes the contents to the path in zookeeper. It will create the path in
        zookeeper if it does not already exist.

        This method will return True if the value is written, False otherwise.
        (The value will not be written if the exclusive is True and the node
        already exists.)
        """
        partial_path = ''

        # We start from the second element because we do not want to inclued
        # the initial empty string before the first "/" because all paths begin
        # with "/". We also don't want to include the final node because that
        # is dealt with later.

        for path_part in path.split("/")[1:-1]:
            partial_path = partial_path + "/" + path_part
            if not(zookeeper.exists(self.handle, partial_path)):
                try:
                    zookeeper.create(self.handle, partial_path, '', [self.acl], 0)
                except zookeeper.NodeExistsException:
                    pass

        exists = zookeeper.exists(self.handle, path)

        # Don't create it if we're exclusive.
        if exists and exclusive:
            return False

        # We make sure that we have the creation flags for ephemeral nodes,
        # otherwise they could be associated with previous connections that
        # have not yet timed out.
        if ephemeral and exists:
            try:
                zookeeper.delete(self.handle, path)
            except zookeeper.NoNodeException:
                pass
            exists = False

        if exists:
            zookeeper.set(self.handle, path, contents)
            return True
        else:
            flags = (ephemeral and zookeeper.EPHEMERAL or 0)
            try:
                zookeeper.create(self.handle, path, contents, [self.acl], flags)
                return True
            except zookeeper.NodeExistsException:
                if not(exclusive):
                    # Woah, something happened between the top and here.
                    # We just restart and retry the whole routine.
                    self.write(path, contents, ephemeral=ephemeral)
                    return True
                else:
                    return False

    def initialize_node_structure(self):
        if not zk.exists(self.zh, self.ROOT):
            self.create(self.ROOT)

        if not zk.exists(self.zh, self.NODE_SERVERS):
            self.create(self.NODE_SERVERS)

        self.NODE_GJOBS = self.ROOT + '/jobs'
        if not zk.exists(self.zh, self.NODE_GJOBS):
            self.create(self.NODE_GJOBS)

    def _write(self, path, contents, ephemeral, exclusive, sequential, mustexist):
        # We start from the second element because we do not want to inclued
        # the initial empty string before the first "/" because all paths begin
        # with "/". We also don't want to include the final node because that
        # is dealt with later.
        partial_path = ''
        for path_part in path.split("/")[1:-1]:
            partial_path = partial_path + "/" + path_part
            if not(zookeeper.exists(self.handle, partial_path)):
                try:
                    zookeeper.create(self.handle, partial_path, '', [self.acl], 0)
                except zookeeper.NodeExistsException:
                    pass

        if sequential:
            exists = False
        else:
            exists = zookeeper.exists(self.handle, path)

        # Don't create it if we're exclusive.
        if exists and exclusive:
            return False

        # Check if we require the node to exist.
        if not exists and mustexist:
            return False

        # We make sure that we have the creation flags for ephemeral nodes,
        # otherwise they could be associated with previous connections that
        # have not yet timed out.
        if ephemeral and exists:
            try:
                zookeeper.delete(self.handle, path)
            except zookeeper.NoNodeException:
                pass
            exists = False

        if exists:
            zookeeper.set(self.handle, path, contents)
            return path
        else:
            flags = 0
            if ephemeral:
                flags = flags | zookeeper.EPHEMERAL
            if sequential:
                flags = flags | zookeeper.SEQUENCE

            # NOTE: We return the final path created.
            return zookeeper.create(self.handle, path, contents, [self.acl], flags)

def set_node(delurl,addurl):
	handler = zookeeper.init("127.0.0.1:2181")
	delnode = zookeeper.exists(handler,delurl)
	addnode = zookeeper.exists(handler,addurl)
	#找到旧的URL就删除
	if delnode:
		zookeeper.delete(handler,delurl)#删除旧的URL
	else:
		write_log(' zookeeper not found old url '+delurl)
	#如果新的URL不存在就创建
	if addnode == None:
		try:
			zookeeper.create(handler,addurl,'',[{"perms":0x1f,"scheme":"world","id":"anyone"}],0)#zookeeper重设URL
			write_log(' zookeeper url from '+delurl+' change to '+addurl+' success')
		except Exception, e:
			write_log(' zookeeper url from '+delurl+' change to '+addurl+' failed')

def exists(path):
    init_client()
    global handle
    try:
        return zookeeper.exists(handle,path) is not None
    except zookeeper.NoNodeException:
        return False

    def getReadLock(self, metadata):
        #Acquire read lock on file through ZooKeeper
        fn = metadata.filename
        print "Trying to acquire read lock for " + self.root + fn
        self.zpath = self.createLockNode(fn, 'r')
        myLock = self.zpath.rsplit('/',1)[1]
        while True:
            children = sorted(zookeeper.get_children(self.zh, self.root + fn))
            lastLock = ''
            for child in children:
                try:
                    if child == myLock:
                        break
                    elif zookeeper.get(self.zh, self.root+fn+'/'+child)[0] == 'w':
                        lastLock = child
                except:
                    pass
            if lastLock != '':
                def watcher (handle, type, state, path):
                    self.cv2.acquire()
                    self.cv2.notify()
                    self.cv2.release()

                self.cv2.acquire()
                if zookeeper.exists(self.zh, self.root+fn+'/'+lastLock, watcher):
                    self.cv2.wait()
                self.cv2.release()
            else:
                break
        print "Acquired read lock for " + self.root + fn

    def __init__(self, zkhosts, root=NODE_HQ_ROOT, alivenode='alive'):
        """zkhosts: a string or a list. list will be ','.join-ed into a string.
        root: root node path (any parents must exist, if any)
        """
        
        if not isinstance(zkhosts, basestring):
            zkhosts = ','.join(zkhosts)
        self.zkhosts = zkhosts
        self.ROOT = root
        self.alivenode = alivenode
        self.nodename = os.uname()[1]

        self.NODE_SERVERS = self.ROOT + '/servers'
        self.NODE_ME = self.NODE_SERVERS+'/'+self.nodename

        self.zh = zk.init(self.zkhosts, self.__watcher)

        self.jobs = {}

        self.initialize_node_structure()

        if not zk.exists(self.zh, self.NODE_ME):
            zk.create(self.zh, self.NODE_ME, '', PERM_WORLD)

        # setup notification
        zk.get_children(self.zh, self.NODE_SERVERS, self.__servers_watcher)
        #zk.get_children(self.zh, self.NODE_ME, self.__watcher)

        self.NODE_JOBS = self.NODE_ME+'/jobs'
        zk.acreate(self.zh, self.NODE_JOBS, '', PERM_WORLD)

  def __init__(self, service_name, connect_string, timeout=DEFAULT_TIMEOUT, default_port=6664):
    self.SERVICE_NODE = "/" + service_name
    self.AVAILABILITY_NODE = self.SERVICE_NODE + "/available"
    self.MEMBERSHIP_NODE = self.SERVICE_NODE + "/members"
    self.connected = False
    self.timeout = timeout
    self.default_port = default_port
    self.conn_cv = threading.Condition()
    self.conn_cv.acquire()
    self.handle = zookeeper.init(connect_string, self.connection_watcher, timeout)
    self.conn_cv.wait(timeout / 1000)
    self.conn_cv.release()
    self.watcher_lock = threading.Lock()
    self.logger = logging.getLogger("sincc")

    if not self.connected:
      raise SinClusterClientError("Unable to connect to %s" % connect_string)

    for path in [self.SERVICE_NODE, self.AVAILABILITY_NODE, self.MEMBERSHIP_NODE]:
      if not zookeeper.exists(self.handle, path):
        zookeeper.create(self.handle, path, "", [ZOO_OPEN_ACL_UNSAFE], 0)
    self.listeners = []
    # Start to watch both /members and /available
    zookeeper.get_children(self.handle, self.MEMBERSHIP_NODE, self.watcher)
    available = zookeeper.get_children(self.handle, self.AVAILABILITY_NODE, self.watcher)
    self.available_nodes = {}
    for node_id in available:
      self.available_nodes[int(node_id)] = Node(int(node_id),
                                                zookeeper.get(self.handle, self.AVAILABILITY_NODE + "/" + node_id)[0])

 def list_children(self, path):
     """
     Returns a list of all the children nodes in the path. None is returned if the path does
     not exist.
     """
     if zookeeper.exists(self.handle, path):
         value = zookeeper.get_children(self.handle, path)
         return value

    def read(self, path, default=None):
        """
        Returns the conents in the path. default is returned if the path does not exists.
        """
        value = default
        if zookeeper.exists(self.handle, path):
            value, timeinfo = zookeeper.get(self.handle, path)

        return value

    def __make_presence(self):
        if not zookeeper.exists(self.zk, '/present/%s' % self.queue):
            self.__create_node('/present/%s' % self.queue, '', perms, 0)

        self.__create_node('/present/%s/%d' % (self.queue, self.zk),
                              '',
                              perms,
                              zookeeper.EPHEMERAL)
        self.__log('Created presence node successfully')

  def registUpdatedKey(self, app_id, current_txid, target_txid, entity_key):
    """ Regist valid transaction id for entity.

    target_txid must be the latest valid transaction id for the entity.
    """
    self.__waitForConnect()
    vtxpath = self.__getValidTransactionPath(app_id, entity_key)
    if zookeeper.exists(self.handle, vtxpath):
      # just update transaction id for entity if there is valid transaction node.
      zookeeper.aset(self.handle, vtxpath, str(target_txid))
    else:
      # store the updated key info into current transaction node.
      value = PATH_SEPARATOR.join([urllib.quote_plus(entity_key), str(target_txid)])
      txpath = self.__getTransactionPath(app_id, current_txid)
      if zookeeper.exists(self.handle, txpath):
        zookeeper.acreate(self.handle, PATH_SEPARATOR.join([txpath, TX_UPDATEDKEY_PREFIX]), value, ZOO_ACL_OPEN, zookeeper.SEQUENCE)
      else:
        raise ZKTransactionException(ZKTransactionException.TYPE_INVALID, "Transaction %d is not valid." % current_txid)

 def __get_message(self):
     messages = self.__get_children('/queues/%s' % self.queue)
     self.__log('got %d children of the queue' % len(messages), level=logging.Logger.debug)
     for message in sorted(messages):
         if not zookeeper.exists(self.zk, '/queues/%s/%s/taken' % (self.queue, message)):
             self.__create_node('/queues/%s/%s/taken' % (self.queue, message), str(self.zk),
                                perms, zookeeper.EPHEMERAL)
             self._current_message = '/queues/%s/%s' % (self.queue, message)
             return self.__get_node('/queues/%s/%s' % (self.queue, message))
     return None

    def register_service(self, service, port=0):
        path = ""
        if service == "master":
            path = ZKConf.ZK_PATH_TEMP_MASTER
        elif service == "stats":
            path = ZKConf.ZK_PATH_TEMP_STATS
        elif service == "proxy":
            path = ZKConf.ZK_PATH_TEMP_PROXY
        
        if not path:
            raise Exception("service type '%s' doesn't exist" % (service,))

        node = zookeeper.exists(self._handler, path, None)
        if not node:
            if not zookeeper.exists(self._handler, ZKConf.ZK_PATH_ROOT, None):
                zookeeper.create(self._handler, ZKConf.ZK_PATH_ROOT, "", [ZKConf.ZOO_CREATOR_ALL_ACL], 0)
            zookeeper.create(self._handler, path, "", [ZKConf.ZOO_CREATOR_ALL_ACL], 0)

        ip = utils.getip()
        mepath = "".join([path, "/", ip, ":", str(port), "-"])
        if not port:
            mepath = "".join([path, "/", ip, "-"])

        childs = zookeeper.get_children(self._handler, path, None)
        is_created = False
        pathme = mepath.split("/")[-1]
        for child in childs:
            if child.startswith(pathme):
                is_created = child.split("-")[-1]
                break

        if is_created:
            meflag = mepath + is_created
            utils.log(utils.cur(), "%s none-callback %s" % ("*"*10, "*"*10), meflag)
        else:
            utils.log(utils.cur(), "%s %s %s" % ("*"*10, self.register_node_cb, "*"*10), mepath)
            meflag = zookeeper.create(self._handler,
                                      mepath,
                                      "",
                                      [ZKConf.ZOO_CREATOR_ALL_ACL],
                                      zookeeper.SEQUENCE|zookeeper.EPHEMERAL)
            utils.log(utils.cur(), "-"*10, meflag)

    def exists(self, path, watch=None):
        """
        Determine whether the ZooKeeper Node at `path` exists

        Arguments:
        - `path`: string

        Return: dict of stats or None
        Exceptions: None
        """
        return zookeeper.exists(self._zk, path, watch)

 def clean_tree(zh, root):
   """Recursively removes the zookeeper subtree underneath :root given zookeeper handle :zh."""
   try:
     if not zookeeper.exists(zh, root):
       return True
     for child in zookeeper.get_children(zh, root):
       if not ZookeeperUtil.clean_tree(zh, posixpath.join(root, child)):
         return False
     zookeeper.delete(zh, root)
   except zookeeper.ZooKeeperException as e:
     return False
   return True

    def post(self):
	request_dict = self.request.arguments
        cluster_name  = (request_dict['cluster_name'])[0]
        zk=zookeeper.init(self.zk_connect(cluster_name))
	new_node = (request_dict['New_post_node'])[0]
	new_value = (request_dict['new_node_value'])[0]
	if zookeeper.exists(zk,new_node):
	    zookeeper.close(zk)
	    self.write("此节点存在")
	else:
	    zookeeper.create(zk,new_node,new_value,[{"perms":0x1f,"scheme":"world","id":"anyone"}],0)	
	    zookeeper.close(zk)
	    self.write("增加成功")

    def cached_exists(self, path):
        cache = self._caches.setdefault('exists', dict())

        retval = cache.get(path, Ellipsis)
        if retval is not Ellipsis:
            return retval

        def invalidator(event):
            cache.pop(path, None)

        retval = zookeeper.exists(self.handle, path, self._wrap_watcher(invalidator))
        cache[path] = retval
        return retval