def update_member_list(request, **kwargs):
    """Update the list of members by adding or removing the necessary members.

    """
    data = request.DATA
    loadbalancer_id = data.get('loadbalancer_id')
    pool_id = kwargs.get('pool_id')
    existing_members = kwargs.get('existing_members')
    members_to_add = kwargs.get('members_to_add')
    members_to_delete = kwargs.get('members_to_delete')

    if members_to_delete:
        kwargs = {'existing_members': existing_members,
                  'members_to_add': members_to_add,
                  'members_to_delete': members_to_delete,
                  'pool_id': pool_id}
        remove_member(request, **kwargs)
    elif members_to_add:
        kwargs = {'existing_members': existing_members,
                  'members_to_add': members_to_add,
                  'members_to_delete': members_to_delete,
                  'pool_id': pool_id}
        add_member(request, **kwargs)
    elif data.get('monitor'):
        args = (request, loadbalancer_id, update_monitor)
        thread.start_new_thread(poll_loadbalancer_status, args)

    def setUp(self):
        # start the server
        self.exit = False
        def server():
            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            s.bind(LOCALPORT); s.listen(1)
            while 1:
                c, a = s.accept()
                if self.exit: c.close(); break
                ending_compat = '\r\n\r\n' if not six.PY3 else b'\r\n\r\n'
                while not c.recv(4096).endswith(ending_compat): pass
                http_compat = 'HTTP/1.1 %d %s\r\n' % self.reply
                c.sendall(http_compat if not six.PY3 else http_compat.encode('utf-8'))
                if self.content is not None:
                    cont_length_compat = 'Content-Length: %d\r\n\r\n' % len(self.content)
                    c.sendall(cont_length_compat if not six.PY3 else cont_length_compat.encode('utf-8'))
                    c.sendall(self.content if not six.PY3 else self.content.encode('utf-8'))
                c.close()
            s.close()
            self.exit = False
        thread.start_new_thread(server, ())

        # create grabber and mirror group objects
        def failure(obj):
            self.code = getattr(obj.exception, 'code', None)
            return {}
        self.g  = URLGrabber()
        self.mg = MirrorGroup(self.g, ['http://%s:%d' % LOCALPORT],
                              failure_callback = failure)

def image_create(request, **kwargs):
    """Create image.

    :param kwargs:
        * copy_from: URL from which Glance server should immediately copy
            the data and store it in its configured image store.
        * data: Form data posted from client.
        * location: URL where the data for this image already resides.

    In the case of 'copy_from' and 'location', the Glance server
    will give us a immediate response from create and handle the data
    asynchronously.

    In the case of 'data' the process of uploading the data may take
    some time and is handed off to a seperate thread.
    """
    data = kwargs.pop("data", None)

    image = glanceclient(request).images.create(**kwargs)

    if data:
        if isinstance(data, TemporaryUploadedFile):
            # Hack to fool Django, so we can keep file open in the new thread.
            data.file.close_called = True
        if isinstance(data, InMemoryUploadedFile):
            # Clone a new file for InMemeoryUploadedFile.
            # Because the old one will be closed by Django.
            data = SimpleUploadedFile(data.name, data.read(), data.content_type)
        thread.start_new_thread(image_update, (request, image.id), {"data": data, "purge_props": False})

    return image

def create_listener(request, **kwargs):
    """Create a new listener.

    """
    data = request.DATA
    listenerSpec = {
        'protocol': data['listener']['protocol'],
        'protocol_port': data['listener']['port'],
        'loadbalancer_id': kwargs['loadbalancer_id']
    }
    if data['listener'].get('name'):
        listenerSpec['name'] = data['listener']['name']
    if data['listener'].get('description'):
        listenerSpec['description'] = data['listener']['description']
    if data.get('certificates'):
        listenerSpec['default_tls_container_ref'] = data['certificates'][0]
        listenerSpec['sni_container_refs'] = data['certificates']

    listener = neutronclient(request).create_listener(
        {'listener': listenerSpec}).get('listener')

    if data.get('pool'):
        args = (request, kwargs['loadbalancer_id'], create_pool)
        kwargs = {'callback_kwargs': {'listener_id': listener['id']}}
        thread.start_new_thread(poll_loadbalancer_status, args, kwargs)

    return listener

 def run(self):
     self._plugins.init_plugins()
     self._dispatcher.start()
     self._client.rtm_connect()
     _thread.start_new_thread(self._keepactive, tuple())
     logger.info('connected to slack RTM api')
     self._dispatcher.loop()

  def log_run_info(self, model_name, dataset_name, run_params, test_id=None):
    """Collect most of the TF runtime information for the local env.

    The schema of the run info follows official/benchmark/datastore/schema.

    Args:
      model_name: string, the name of the model.
      dataset_name: string, the name of dataset for training and evaluation.
      run_params: dict, the dictionary of parameters for the run, it could
        include hyperparameters or other params that are important for the run.
      test_id: string, the unique name of the test run by the combination of key
        parameters, eg batch size, num of GPU. It is hardware independent.
    """
    run_info = _gather_run_info(model_name, dataset_name, run_params, test_id)
    # Starting new thread for bigquery upload in case it might take long time
    # and impact the benchmark and performance measurement. Starting a new
    # thread might have potential performance impact for model that run on CPU.
    thread.start_new_thread(
        self._bigquery_uploader.upload_benchmark_run_json,
        (self._bigquery_data_set,
         self._bigquery_run_table,
         self._run_id,
         run_info))
    thread.start_new_thread(
        self._bigquery_uploader.insert_run_status,
        (self._bigquery_data_set,
         self._bigquery_run_status_table,
         self._run_id,
         RUN_STATUS_RUNNING))

 def on_finish(self, status):
   thread.start_new_thread(
       self._bigquery_uploader.update_run_status,
       (self._bigquery_data_set,
        self._bigquery_run_status_table,
        self._run_id,
        status))

def create_pool(request, **kwargs):
    """Create a new pool.

    """
    data = request.DATA
    poolSpec = {
        'protocol': data['pool']['protocol'],
        'lb_algorithm': data['pool']['method'],
        'listener_id': kwargs['listener_id']
    }
    if data['pool'].get('name'):
        poolSpec['name'] = data['pool']['name']
    if data['pool'].get('description'):
        poolSpec['description'] = data['pool']['description']
    pool = neutronclient(request).create_lbaas_pool(
        {'pool': poolSpec}).get('pool')

    if data.get('members'):
        args = (request, kwargs['loadbalancer_id'], add_member)
        kwargs = {'callback_kwargs': {'pool_id': pool['id'],
                                      'index': 0}}
        thread.start_new_thread(poll_loadbalancer_status, args, kwargs)
    elif data.get('monitor'):
        args = (request, kwargs['loadbalancer_id'], create_health_monitor)
        kwargs = {'callback_kwargs': {'pool_id': pool['id']}}
        thread.start_new_thread(poll_loadbalancer_status, args, kwargs)

    return pool

def testConNeg():
    _thread.start_new_thread(runHttpServer, tuple())
    # hang on a second while server starts
    time.sleep(1)
    graph=Graph()
    graph.parse("http://localhost:12345/foo", format="xml")
    graph.parse("http://localhost:12345/foo", format="n3")
    graph.parse("http://localhost:12345/foo", format="nt")

 def _parallel_split(obj, eng, calls):
     lock = thread.allocate_lock()
     i = 0
     eng.setVar('lock', lock)
     for func in calls:
         new_eng = duplicate_engine_instance(eng)
         new_eng.setWorkflow([lambda o, e: e.setVar('lock', lock), func])
         thread.start_new_thread(new_eng.process, ([obj], ))

def image_create(request, **kwargs):
    """Create image.

    :param kwargs:
        * copy_from: URL from which Glance server should immediately copy
            the data and store it in its configured image store.
        * data: Form data posted from client.
        * location: URL where the data for this image already resides.

    In the case of 'copy_from' and 'location', the Glance server
    will give us a immediate response from create and handle the data
    asynchronously.

    In the case of 'data' the process of uploading the data may take
    some time and is handed off to a separate thread.
    """
    data = kwargs.pop('data', None)
    location = None
    if VERSIONS.active >= 2:
        location = kwargs.pop('location', None)

    image = glanceclient(request).images.create(**kwargs)
    if location is not None:
        glanceclient(request).images.add_location(image.id, location, {})

    if data:
        if isinstance(data, six.string_types):
            # The image data is meant to be uploaded externally, return a
            # special wrapper to bypass the web server in a subsequent upload
            return ExternallyUploadedImage(image, request)
        elif isinstance(data, TemporaryUploadedFile):
            # Hack to fool Django, so we can keep file open in the new thread.
            data.file.close_called = True
        elif isinstance(data, InMemoryUploadedFile):
            # Clone a new file for InMemeoryUploadedFile.
            # Because the old one will be closed by Django.
            data = SimpleUploadedFile(data.name,
                                      data.read(),
                                      data.content_type)
        if VERSIONS.active < 2:
            thread.start_new_thread(image_update,
                                    (request, image.id),
                                    {'data': data})
        else:
            def upload():
                try:
                    return glanceclient(request).images.upload(image.id, data)
                finally:
                    filename = str(data.file.name)
                    try:
                        os.remove(filename)
                    except OSError as e:
                        LOG.warning('Failed to remove temporary image file '
                                    '%(file)s (%(e)s)',
                                    {'file': filename, 'e': e})
            thread.start_new_thread(upload, ())

    return Image(image)

 def _parallel_split(obj, eng, calls):
     lock = thread.allocate_lock()
     eng.store['lock'] = lock
     for func in calls:
         new_eng = eng.duplicate()
         new_eng.setWorkflow(
             [lambda o, e: e.store.update({'lock': lock}), func]
         )
         thread.start_new_thread(new_eng.process, ([obj], ))

    def _rtm_connect(self):
        r = self.slacker.rtm.start().body
        self.driver_username = r['self']['name']
        self.driver_userid = r['self']['id']

        self.users = {u['name']: u['id'] for u in r['users']}
        self.testbot_userid = self.users[self.testbot_username]

        self._websocket = websocket.create_connection(r['url'])
        self._websocket.sock.setblocking(0)
        _thread.start_new_thread(self._rtm_read_forever, tuple())

    def subscribe(self, notification, cb, data=None):

        np_data = {
            "running": True,
            "notification": notification,
            "callback": cb,
            "userdata": data,
        }

        thread.start_new_thread( self.notifier, ("NotificationProxyNotifier_"+notification, np_data, ) )
        while(1):
            time.sleep(1)

def image_create(request, **kwargs):
    copy_from = kwargs.pop("copy_from", None)
    data = kwargs.pop("data", None)

    image = glanceclient(request).images.create(**kwargs)

    if data:
        thread.start_new_thread(image_update, (request, image.id), {"data": data, "purge_props": False})
    elif copy_from:
        thread.start_new_thread(image_update, (request, image.id), {"copy_from": copy_from, "purge_props": False})

    return image

 def _process(self, event):
     """ 基于多线程的处理事件 """
     if event.route not in self._routes:
         log.warning("事件%s 没有被处理" % event.route)
         return
     for handler in self._routes[event.route]:
         try:
             log.debug("处理事件%s" % event.route)
             _thread.start_new_thread(handler, (event,))
             #handler(event)
         except Exception as e:
             log.error(e)

    def run(self):
        """Start a new thread for the player.

        Call this function before trying to play any music with
        play_file() or play().
        """

        # If we don't use the MainLoop, messages are never sent.

        def start():
            loop = GLib.MainLoop()
            loop.run()

        _thread.start_new_thread(start, ())

    def notifier(self, name, args=None):

        if args == None:
            return None

        self.observe_notification(args.get("notification"))

        while args.get("running") == True:
            np_name = self.get_notification(args.get("notification"))
            if np_name:
                userdata = args.get("userdata")
                try:
                    thread.start_new_thread( args.get("callback") , (np_name, userdata, ) )
                except:
                    self.logger.error("Error: unable to start thread")

def start_xmlrpc_server(addr='localhost', port=8000):        
    server = SimpleXMLRPCServer((addr, port))
    functions   = [
        getRootNodePath,
        getitem,
        getChildNodes,
        getMethods,
        getMethodDoc,
        getRepr,
        callMethod
    ]
    for function in functions:
        server.register_function(function, function.__name__)
    thread.start_new_thread(server.serve_forever, tuple())
    return server
        

def image_create(request, **kwargs):
    copy_from = kwargs.pop('copy_from', None)
    data = kwargs.pop('data', None)

    image = glanceclient(request).images.create(**kwargs)

    if data:
        thread.start_new_thread(image_update,
                                (request, image.id),
                                {'data': data,
                                 'purge_props': False})
    elif copy_from:
        thread.start_new_thread(image_update,
                                (request, image.id),
                                {'copy_from': copy_from,
                                 'purge_props': False})

    return image