def Run(self):
    self.error = False
    thread_pool = self.thread_pool_factory(
        self.num_threads, queue_size=self.max_queue_size)

    progress_queue = self.progress_queue_factory(self.max_queue_size)
    self.request_manager = self.request_manager_factory(
        app_id=self.app_id,
        host_port=self.host_port,
        url_path=self.url_path,
        kind=self.kind,
        throttle=self.throttle,
        batch_size=self.batch_size,
        secure=self.secure,
        dry_run=self.dry_run,
        server=self.server,
        throttle_class=self.throttle_class,
        oauth2_parameters=self.oauth2_parameters)
    try:


      self.request_manager.Authenticate()
    except Exception, e:
      self.error = True


      if not isinstance(e, urllib2.HTTPError) or (
          e.code != 302 and e.code != 401):
        logger.exception('Exception during authentication')
      raise AuthenticationError()
    if (self.request_manager.auth_called and
        not self.request_manager.authenticated):


      self.error = True
      raise AuthenticationError('Authentication failed')

    kinds = self.RunPostAuthentication()

    for thread in thread_pool.Threads():
      self.throttle.Register(thread)

    self.progress_thread = self.progresstrackerthread_factory(
        progress_queue, self.progress_db)

    if self.progress_db.UseProgressData():
      logger.debug('Restarting upload using progress database')
      progress_generator_factory = self.progress_db.GetProgressStatusGenerator
    else:
      progress_generator_factory = None

    self.data_source_thread = (
        self.datasourcethread_factory(self.request_manager,
                                      kinds,
                                      thread_pool,
                                      progress_queue,
                                      self.input_generator_factory,
                                      progress_generator_factory))



    self.throttle.Register(self.data_source_thread)


    thread_local = threading.local()
    thread_local.shut_down = False

    def Interrupt(unused_signum, unused_frame):
      """Shutdown gracefully in response to a signal."""
      thread_local.shut_down = True
      self.error = True

    signal.signal(signal.SIGINT, Interrupt)


    self.progress_thread.start()
    self.data_source_thread.start()



    while not thread_local.shut_down:





      self.data_source_thread.join(timeout=0.25)


      if self.data_source_thread.isAlive():


        for thread in list(thread_pool.Threads()) + [self.progress_thread]:
          if not thread.isAlive():
            logger.info('Unexpected thread death: %s', thread.getName())
            thread_local.shut_down = True
            self.error = True
            break
      else:

        break

    def _Join(ob, msg):
      logger.debug('Waiting for %s...', msg)

      if isinstance(ob, threading.Thread):
        ob.join(timeout=3.0)
        if ob.isAlive():
          logger.debug('Joining %s failed', ob)
        else:
          logger.debug('... done.')
      elif isinstance(ob, (Queue.Queue, ReQueue)):
        if not InterruptibleQueueJoin(ob, thread_local, thread_pool):
          ShutdownThreads(self.data_source_thread, thread_pool)
      else:
        ob.join()
        logger.debug('... done.')


    if self.data_source_thread.error or thread_local.shut_down:
      ShutdownThreads(self.data_source_thread, thread_pool)
    else:

      _Join(thread_pool.requeue, 'worker threads to finish')

    thread_pool.Shutdown()

    thread_pool.JoinThreads()
    thread_pool.CheckErrors()
    print ''





    if self.progress_thread.isAlive():
      InterruptibleQueueJoin(progress_queue, thread_local, thread_pool,
                             check_workers=False)
    else:
      logger.warn('Progress thread exited prematurely')



    progress_queue.put(_THREAD_SHOULD_EXIT)
    _Join(self.progress_thread, 'progress_thread to terminate')
    self.progress_thread.CheckError()
    if not thread_local.shut_down:
      self.progress_thread.WorkFinished()


    self.data_source_thread.CheckError()

    return self.ReportStatus()
Perform the work of the BulkTransporterApp.
dataset/ETHPy150Open GoogleCloudPlatform/python-compat-runtime/appengine-compat/exported_appengine_sdk/google/appengine/tools/bulkloader.py/BulkTransporterApp.Run