original_docstring: 
      Main loop to submit a new transfer primitive to a transfertool.
      
docstring: Main loop to submit a new transfer primitive to a transfertool.
docstring_tokens: Main loop to submit a new transfer primitive to a transfertool .
code: def stager(once=False, rses=None, mock=False, bulk=100, group_bulk=1, group_policy='rule',
             fts_source_strategy='auto', activities=None, sleep_time=600, retry_other_fts=False):
      """
      Main loop to submit a new transfer primitive to a transfertool.
      """
  
      try:
          scheme = config_get('conveyor', 'scheme')
      except NoOptionError:
          scheme = None
  
      try:
          failover_scheme = config_get('conveyor', 'failover_scheme')
      except NoOptionError:
          failover_scheme = None
  
      try:
          bring_online = config_get('conveyor', 'bring_online')
      except NoOptionError:
          bring_online = 43200
  
      try:
          max_time_in_queue = {}
          timelife_conf = config_get('conveyor', 'max_time_in_queue')
          timelife_confs = timelife_conf.split(",")
          for conf in timelife_confs:
              act, timelife = conf.split(":")
              max_time_in_queue[act.strip()] = int(timelife.strip())
      except NoOptionError:
          max_time_in_queue = {}
      if 'default' not in max_time_in_queue:
          max_time_in_queue['default'] = 168
      logging.debug("Maximum time in queue for different activities: %s" % max_time_in_queue)
  
      activity_next_exe_time = defaultdict(time.time)
      executable = ' '.join(sys.argv)
      hostname = socket.getfqdn()
      pid = os.getpid()
      hb_thread = threading.current_thread()
      heartbeat.sanity_check(executable=executable, hostname=hostname)
      heart_beat = heartbeat.live(executable, hostname, pid, hb_thread)
      prepend_str = 'Thread [%i/%i] : ' % (heart_beat['assign_thread'] + 1, heart_beat['nr_threads'])
      logging.info(prepend_str + 'Stager starting with bring_online %s seconds' % (bring_online))
  
      time.sleep(10)  # To prevent running on the same partition if all the poller restart at the same time
      heart_beat = heartbeat.live(executable, hostname, pid, hb_thread)
      prepend_str = 'Thread [%i/%i] : ' % (heart_beat['assign_thread'] + 1, heart_beat['nr_threads'])
      logging.info(prepend_str + 'Stager started')
  
      while not graceful_stop.is_set():
  
          try:
              heart_beat = heartbeat.live(executable, hostname, pid, hb_thread)
              prepend_str = 'Thread [%i/%i] : ' % (heart_beat['assign_thread'] + 1, heart_beat['nr_threads'])
  
              if activities is None:
                  activities = [None]
              if rses:
                  rse_ids = [rse['id'] for rse in rses]
              else:
                  rse_ids = None
  
              for activity in activities:
                  if activity_next_exe_time[activity] > time.time():
                      graceful_stop.wait(1)
                      continue
  
                  logging.info(prepend_str + 'Starting to get stagein transfers for %s' % (activity))
                  start_time = time.time()
                  transfers = __get_stagein_transfers(total_workers=heart_beat['nr_threads'] - 1,
                                                      worker_number=heart_beat['assign_thread'],
                                                      failover_schemes=failover_scheme,
                                                      limit=bulk,
                                                      activity=activity,
                                                      rses=rse_ids,
                                                      mock=mock,
                                                      schemes=scheme,
                                                      bring_online=bring_online,
                                                      retry_other_fts=retry_other_fts)
                  record_timer('daemons.conveyor.stager.get_stagein_transfers.per_transfer', (time.time() - start_time) * 1000 / (len(transfers) if transfers else 1))
                  record_counter('daemons.conveyor.stager.get_stagein_transfers', len(transfers))
                  record_timer('daemons.conveyor.stager.get_stagein_transfers.transfers', len(transfers))
                  logging.info(prepend_str + 'Got %s stagein transfers for %s' % (len(transfers), activity))
  
                  # group transfers
                  logging.info(prepend_str + 'Starting to group transfers for %s' % (activity))
                  start_time = time.time()
                  grouped_jobs = bulk_group_transfer(transfers, group_policy, group_bulk, fts_source_strategy, max_time_in_queue)
                  record_timer('daemons.conveyor.stager.bulk_group_transfer', (time.time() - start_time) * 1000 / (len(transfers) if transfers else 1))
  
                  logging.info(prepend_str + 'Starting to submit transfers for %s' % (activity))
                  # submit transfers
                  for external_host in grouped_jobs:
                      for job in grouped_jobs[external_host]:
                          # submit transfers
                          submit_transfer(external_host=external_host, job=job, submitter='transfer_submitter', logging_prepend_str=prepend_str)
  
                  if len(transfers) < group_bulk:
                      logging.info(prepend_str + 'Only %s transfers for %s which is less than group bulk %s, sleep %s seconds' % (len(transfers), activity, group_bulk, sleep_time))
                      if activity_next_exe_time[activity] < time.time():
                          activity_next_exe_time[activity] = time.time() + sleep_time
          except Exception:
              logging.critical(prepend_str + '%s' % (traceback.format_exc()))
  
          if once:
              break
  
      logging.info(prepend_str + 'Graceful stop requested')
  
      heartbeat.die(executable, hostname, pid, hb_thread)
  
      logging.info(prepend_str + 'Graceful stop done')
code_tokens: def stager ( once = False , rses = None , mock = False , bulk = 100 , group_bulk = 1 , group_policy = 'rule' , fts_source_strategy = 'auto' , activities = None , sleep_time = 600 , retry_other_fts = False ) : try : scheme = config_get ( 'conveyor' , 'scheme' ) except NoOptionError : scheme = None try : failover_scheme = config_get ( 'conveyor' , 'failover_scheme' ) except NoOptionError : failover_scheme = None try : bring_online = config_get ( 'conveyor' , 'bring_online' ) except NoOptionError : bring_online = 43200 try : max_time_in_queue = { } timelife_conf = config_get ( 'conveyor' , 'max_time_in_queue' ) timelife_confs = timelife_conf . split ( "," ) for conf in timelife_confs : act , timelife = conf . split ( ":" ) max_time_in_queue [ act . strip ( ) ] = int ( timelife . strip ( ) ) except NoOptionError : max_time_in_queue = { } if 'default' not in max_time_in_queue : max_time_in_queue [ 'default' ] = 168 logging . debug ( "Maximum time in queue for different activities: %s" % max_time_in_queue ) activity_next_exe_time = defaultdict ( time . time ) executable = ' ' . join ( sys . argv ) hostname = socket . getfqdn ( ) pid = os . getpid ( ) hb_thread = threading . current_thread ( ) heartbeat . sanity_check ( executable = executable , hostname = hostname ) heart_beat = heartbeat . live ( executable , hostname , pid , hb_thread ) prepend_str = 'Thread [%i/%i] : ' % ( heart_beat [ 'assign_thread' ] + 1 , heart_beat [ 'nr_threads' ] ) logging . info ( prepend_str + 'Stager starting with bring_online %s seconds' % ( bring_online ) ) time . sleep ( 10 ) heart_beat = heartbeat . live ( executable , hostname , pid , hb_thread ) prepend_str = 'Thread [%i/%i] : ' % ( heart_beat [ 'assign_thread' ] + 1 , heart_beat [ 'nr_threads' ] ) logging . info ( prepend_str + 'Stager started' ) while not graceful_stop . is_set ( ) : try : heart_beat = heartbeat . live ( executable , hostname , pid , hb_thread ) prepend_str = 'Thread [%i/%i] : ' % ( heart_beat [ 'assign_thread' ] + 1 , heart_beat [ 'nr_threads' ] ) if activities is None : activities = [ None ] if rses : rse_ids = [ rse [ 'id' ] for rse in rses ] else : rse_ids = None for activity in activities : if activity_next_exe_time [ activity ] > time . time ( ) : graceful_stop . wait ( 1 ) continue logging . info ( prepend_str + 'Starting to get stagein transfers for %s' % ( activity ) ) start_time = time . time ( ) transfers = __get_stagein_transfers ( total_workers = heart_beat [ 'nr_threads' ] - 1 , worker_number = heart_beat [ 'assign_thread' ] , failover_schemes = failover_scheme , limit = bulk , activity = activity , rses = rse_ids , mock = mock , schemes = scheme , bring_online = bring_online , retry_other_fts = retry_other_fts ) record_timer ( 'daemons.conveyor.stager.get_stagein_transfers.per_transfer' , ( time . time ( ) - start_time ) * 1000 / ( len ( transfers ) if transfers else 1 ) ) record_counter ( 'daemons.conveyor.stager.get_stagein_transfers' , len ( transfers ) ) record_timer ( 'daemons.conveyor.stager.get_stagein_transfers.transfers' , len ( transfers ) ) logging . info ( prepend_str + 'Got %s stagein transfers for %s' % ( len ( transfers ) , activity ) ) logging . info ( prepend_str + 'Starting to group transfers for %s' % ( activity ) ) start_time = time . time ( ) grouped_jobs = bulk_group_transfer ( transfers , group_policy , group_bulk , fts_source_strategy , max_time_in_queue ) record_timer ( 'daemons.conveyor.stager.bulk_group_transfer' , ( time . time ( ) - start_time ) * 1000 / ( len ( transfers ) if transfers else 1 ) ) logging . info ( prepend_str + 'Starting to submit transfers for %s' % ( activity ) ) for external_host in grouped_jobs : for job in grouped_jobs [ external_host ] : submit_transfer ( external_host = external_host , job = job , submitter = 'transfer_submitter' , logging_prepend_str = prepend_str ) if len ( transfers ) < group_bulk : logging . info ( prepend_str + 'Only %s transfers for %s which is less than group bulk %s, sleep %s seconds' % ( len ( transfers ) , activity , group_bulk , sleep_time ) ) if activity_next_exe_time [ activity ] < time . time ( ) : activity_next_exe_time [ activity ] = time . time ( ) + sleep_time except Exception : logging . critical ( prepend_str + '%s' % ( traceback . format_exc ( ) ) ) if once : break logging . info ( prepend_str + 'Graceful stop requested' ) heartbeat . die ( executable , hostname , pid , hb_thread ) logging . info ( prepend_str + 'Graceful stop done' )
short_docstring: Main loop to submit a new transfer primitive to a transfertool.
short_docstring_tokens: Main loop to submit a new transfer primitive to a transfertool .
comment: """
      Main loop to submit a new transfer primitive to a transfertool.
      """$SEP$# To prevent running on the same partition if all the poller restart at the same time$SEP$# group transfers$SEP$# submit transfers$SEP$# submit transfers