parameters: (self, args, etcd_client)
docstring: 0 group mode, be aware of healthy status of other workers
          1 decouple mode, check own status only
docstring_summary: 0 group mode, be aware of healthy status of other workers
          1 decouple mode, check own status only
docstring_tokens: 0 group mode be aware of healthy status of other workers 1 decouple mode check own status only
function: def __init__(self, args, etcd_client):
  
          self.args = args
          server = args.elastic_server or os.getenv('PADDLE_ELASTIC_SERVER')
          name = args.job_id or os.getenv('PADDLE_ELASTIC_JOB_ID')
          self.min_np, self.max_np = self._parse_np(args.np)
          host = args.host or os.getenv('POD_IP')
          scale = args.scale or int(os.getenv('PADDLE_ELASTIC_SCALE', 0))
          force = args.force or os.getenv('PADDLE_ELASTIC_FORCE')
  
          self.host = host if host else self._get_host()
  
          (self.device_mode,
           self.devices_per_proc) = launch_utils.get_device_proc_info(args)
  
          self.elastic_timeout = int(
              os.getenv('PADDLE_ELASTIC_TIMEOUT', ELASTIC_TIMEOUT))
          elastic_ttl = int(os.getenv('PADDLE_ELASTIC_TTL', ELASTIC_TTL))
  
          self.start_port = None
          if cloud_utils.use_paddlecloud():
              self.trainers = os.getenv('PADDLE_TRAINERS', '')
              self.np = len(self.trainers.split(","))
              self.start_port = int(os.getenv("PADDLE_PORT", "6170"))
              self.dist_endpoints = os.getenv('DISTRIBUTED_TRAINER_ENDPOINTS', '')
              trainer_endpoints = os.getenv('PADDLE_TRAINER_ENDPOINTS', '')
              self.trainer_endpoints_list = trainer_endpoints.split(",")
          else:
              self.trainers = args.ips or os.getenv('PADDLE_TRAINERS', '')
              node_ips = self.trainers.split(",")
              self.np = len(node_ips)
              self.start_port = int(os.getenv("FLAGS_START_PORT", "6170"))
              self.dist_endpoints = self._host_to_endpoints(
                  node_ips, self.devices_per_proc, self.start_port)
              self.trainer_endpoints_list = [
                  "%s:%d" % (ip, self.start_port) for ip in node_ips
              ]
  
          self.curr_host = "%s:%d" % (self.host, self.start_port)
          logger.info(f'start job with np={self.np}')
          logger.info(
              f"trainers={self.trainers}, trainer_endpoints_list={self.trainer_endpoints_list}"
          )
  
          # auto correct the value of elastic_level
          # 1: Fault tolerant, 2: Elastic
          self.elastic_level = int(
              os.getenv('PADDLE_ELASTIC_FAULT_TOLERANC_LEVEL',
                        ElasticLevel.FAULT_TOLERANCE))
          if self.min_np == self.max_np or \
                  (self.min_np > 0 and self.max_np == 0):
              self.elastic_level = ElasticLevel.FAULT_TOLERANCE
              logger.info(f'start job with ElasticLevel.FAULT_TOLERANCE')
          if self.min_np > 0 and self.max_np > self.min_np:
              self.elastic_level = ElasticLevel.ELASTIC
              logger.info(f'start job with ElasticLevel.ELASTIC')
  
          # compatible with kuberntes service discovery
          if not server and os.getenv(
                  'PADDLE_ELASTIC_ETCD_SERVICE_HOST') and os.getenv(
                      'PADDLE_ELASTIC_ETCD_SERVICE_PORT'):
              server = '{}:{}'.format(
                  os.getenv('PADDLE_ELASTIC_ETCD_SERVICE_HOST'),
                  os.getenv('PADDLE_ELASTIC_ETCD_SERVICE_PORT'))
  
          logger.debug('init with server {} host {}'.format(server, host))
  
          self.hosts = []
          self.stopped = False
  
          self.sigint = 0
          self.need_sync = False
  
          self.elastic_startup_time = None
  
          if not server or ':' not in server or not name or not self.np:
              logger.info(
                  'Elastic is not enabled with server {} name {} and np {}'.
                  format(server, name, self.np))
              self.enable = False
              return
          else:
              self.enable = True
  
          self.etcd = etcd_client
  
          # etcd data
          self.prefix = "/paddle/" + name
          self.node_prefix = self.prefix + '/nodes'
          self.np_path = self.prefix + '/np'
          self.endpoints_path = self.prefix + '/endpoints'
  
          node_tag = ''.join(
              random.choice('abcdefghijklmnopqrstuvwxyz') for _ in range(6))
          self.host_path = '{}/{}{}'.format(self.node_prefix, node_tag,
                                            time.time())
          '''
          0 group mode, be aware of healthy status of other workers
          1 decouple mode, check own status only
          '''
          self.etcd.put(self.prefix, b'0')
  
          # register callback
          def host_call_back(event):
              self.hosts = [
                  six.ensure_str(i[0])
                  for i in self.etcd.get_prefix(self.node_prefix)
              ]
              self.hosts = list(set(self.hosts)) if self.hosts else self.hosts
              logger.info(
                  f"host_call_back curr_host={self.curr_host}, hosts:{self.hosts}")
              self.need_sync = True
              self.elastic_startup_time = None
  
          host_watch = self.etcd.add_watch_prefix_callback(self.node_prefix,
                                                           host_call_back)
          host_lease = self.etcd.lease(elastic_ttl)
  
          # register etcd lease heartbeat
          def lease_heartbeat():
              while True:
                  try:
                      host_lease.refresh()
  
                      hosts = [
                          six.ensure_str(i[0])
                          for i in self.etcd.get_prefix(self.node_prefix)
                      ]
                      hosts = list(set(hosts)) if hosts else hosts
                      logger.info(
                          f"[lease_heartbeat] curr_host={self.curr_host}, hosts={hosts}"
                      )
                      if self.curr_host not in hosts:
                          logger.info(
                              f"[lease_heartbeat] register host={self.curr_host}")
                          self.etcd.put(self.host_path,
                                        six.b(self.curr_host),
                                        lease=host_lease)
                  except Exception as e:
                      logger.error("[lease_heartbeat] internal error:{} {}".
                                   format(e, traceback.format_exc()))
                      break
                  time.sleep(elastic_ttl / 3)
  
          keepalived_thread = threading.Thread(
              name='lease_heartbeat', target=lease_heartbeat, daemon=True)
          keepalived_thread.start()
  
          self.etcd.put(self.host_path, six.b(self.curr_host), lease=host_lease)
  
          # endpoints handle DISTRIBUTED_TRAINER_ENDPOINTS and PADDLE_TRAINERS
          self.etcd.put(self.endpoints_path,
                        six.b('{}|{}'.format(self.dist_endpoints, self.trainers)))
  
          def endpoints_call_back(event):
              if not self.dist_endpoints:
                  return
              edps = six.ensure_str(self.etcd.get(self.endpoints_path)[0] or '')
              self.dist_endpoints, self.trainers = edps.split('|')
              logger.info("set DISTRIBUTED_TRAINER_ENDPOINTS {} ".format(
                  self.dist_endpoints))
              logger.info("set PADDLE_TRAINERS {} ".format(self.trainers))
  
          endpoints_watch = self.etcd.add_watch_callback(self.endpoints_path,
                                                         endpoints_call_back)
  
          self.watches = [host_watch, endpoints_watch]
          self.launcher = None
function_tokens: def __init__ ( self , args , etcd_client ) : self . args = args server = args . elastic_server or os . getenv ( 'PADDLE_ELASTIC_SERVER' ) name = args . job_id or os . getenv ( 'PADDLE_ELASTIC_JOB_ID' ) self . min_np , self . max_np = self . _parse_np ( args . np ) host = args . host or os . getenv ( 'POD_IP' ) scale = args . scale or int ( os . getenv ( 'PADDLE_ELASTIC_SCALE' , 0 ) ) force = args . force or os . getenv ( 'PADDLE_ELASTIC_FORCE' ) self . host = host if host else self . _get_host ( ) ( self . device_mode , self . devices_per_proc ) = launch_utils . get_device_proc_info ( args ) self . elastic_timeout = int ( os . getenv ( 'PADDLE_ELASTIC_TIMEOUT' , ELASTIC_TIMEOUT ) ) elastic_ttl = int ( os . getenv ( 'PADDLE_ELASTIC_TTL' , ELASTIC_TTL ) ) self . start_port = None if cloud_utils . use_paddlecloud ( ) : self . trainers = os . getenv ( 'PADDLE_TRAINERS' , '' ) self . np = len ( self . trainers . split ( "," ) ) self . start_port = int ( os . getenv ( "PADDLE_PORT" , "6170" ) ) self . dist_endpoints = os . getenv ( 'DISTRIBUTED_TRAINER_ENDPOINTS' , '' ) trainer_endpoints = os . getenv ( 'PADDLE_TRAINER_ENDPOINTS' , '' ) self . trainer_endpoints_list = trainer_endpoints . split ( "," ) else : self . trainers = args . ips or os . getenv ( 'PADDLE_TRAINERS' , '' ) node_ips = self . trainers . split ( "," ) self . np = len ( node_ips ) self . start_port = int ( os . getenv ( "FLAGS_START_PORT" , "6170" ) ) self . dist_endpoints = self . _host_to_endpoints ( node_ips , self . devices_per_proc , self . start_port ) self . trainer_endpoints_list = [ "%s:%d" % ( ip , self . start_port ) for ip in node_ips ] self . curr_host = "%s:%d" % ( self . host , self . start_port ) logger . info ( f'start job with np={self.np}' ) logger . info ( f"trainers={self.trainers}, trainer_endpoints_list={self.trainer_endpoints_list}" ) # auto correct the value of elastic_level # 1: Fault tolerant, 2: Elastic self . elastic_level = int ( os . getenv ( 'PADDLE_ELASTIC_FAULT_TOLERANC_LEVEL' , ElasticLevel . FAULT_TOLERANCE ) ) if self . min_np == self . max_np or ( self . min_np > 0 and self . max_np == 0 ) : self . elastic_level = ElasticLevel . FAULT_TOLERANCE logger . info ( f'start job with ElasticLevel.FAULT_TOLERANCE' ) if self . min_np > 0 and self . max_np > self . min_np : self . elastic_level = ElasticLevel . ELASTIC logger . info ( f'start job with ElasticLevel.ELASTIC' ) # compatible with kuberntes service discovery if not server and os . getenv ( 'PADDLE_ELASTIC_ETCD_SERVICE_HOST' ) and os . getenv ( 'PADDLE_ELASTIC_ETCD_SERVICE_PORT' ) : server = '{}:{}' . format ( os . getenv ( 'PADDLE_ELASTIC_ETCD_SERVICE_HOST' ) , os . getenv ( 'PADDLE_ELASTIC_ETCD_SERVICE_PORT' ) ) logger . debug ( 'init with server {} host {}' . format ( server , host ) ) self . hosts = [ ] self . stopped = False self . sigint = 0 self . need_sync = False self . elastic_startup_time = None if not server or ':' not in server or not name or not self . np : logger . info ( 'Elastic is not enabled with server {} name {} and np {}' . format ( server , name , self . np ) ) self . enable = False return else : self . enable = True self . etcd = etcd_client # etcd data self . prefix = "/paddle/" + name self . node_prefix = self . prefix + '/nodes' self . np_path = self . prefix + '/np' self . endpoints_path = self . prefix + '/endpoints' node_tag = '' . join ( random . choice ( 'abcdefghijklmnopqrstuvwxyz' ) for _ in range ( 6 ) ) self . host_path = '{}/{}{}' . format ( self . node_prefix , node_tag , time . time ( ) ) self . etcd . put ( self . prefix , b'0' ) # register callback def host_call_back ( event ) : self . hosts = [ six . ensure_str ( i [ 0 ] ) for i in self . etcd . get_prefix ( self . node_prefix ) ] self . hosts = list ( set ( self . hosts ) ) if self . hosts else self . hosts logger . info ( f"host_call_back curr_host={self.curr_host}, hosts:{self.hosts}" ) self . need_sync = True self . elastic_startup_time = None host_watch = self . etcd . add_watch_prefix_callback ( self . node_prefix , host_call_back ) host_lease = self . etcd . lease ( elastic_ttl ) # register etcd lease heartbeat def lease_heartbeat ( ) : while True : try : host_lease . refresh ( ) hosts = [ six . ensure_str ( i [ 0 ] ) for i in self . etcd . get_prefix ( self . node_prefix ) ] hosts = list ( set ( hosts ) ) if hosts else hosts logger . info ( f"[lease_heartbeat] curr_host={self.curr_host}, hosts={hosts}" ) if self . curr_host not in hosts : logger . info ( f"[lease_heartbeat] register host={self.curr_host}" ) self . etcd . put ( self . host_path , six . b ( self . curr_host ) , lease = host_lease ) except Exception as e : logger . error ( "[lease_heartbeat] internal error:{} {}" . format ( e , traceback . format_exc ( ) ) ) break time . sleep ( elastic_ttl / 3 ) keepalived_thread = threading . Thread ( name = 'lease_heartbeat' , target = lease_heartbeat , daemon = True ) keepalived_thread . start ( ) self . etcd . put ( self . host_path , six . b ( self . curr_host ) , lease = host_lease ) # endpoints handle DISTRIBUTED_TRAINER_ENDPOINTS and PADDLE_TRAINERS self . etcd . put ( self . endpoints_path , six . b ( '{}|{}' . format ( self . dist_endpoints , self . trainers ) ) ) def endpoints_call_back ( event ) : if not self . dist_endpoints : return edps = six . ensure_str ( self . etcd . get ( self . endpoints_path ) [ 0 ] or '' ) self . dist_endpoints , self . trainers = edps . split ( '|' ) logger . info ( "set DISTRIBUTED_TRAINER_ENDPOINTS {} " . format ( self . dist_endpoints ) ) logger . info ( "set PADDLE_TRAINERS {} " . format ( self . trainers ) ) endpoints_watch = self . etcd . add_watch_callback ( self . endpoints_path , endpoints_call_back ) self . watches = [ host_watch , endpoints_watch ] self . launcher = None