Set Engine's dry_run_mode true to override all target_id of worker sections
324430
def dry_run
      begin
        Fluent::Engine.dry_run_mode = true
        change_privilege
        init_engine
        run_configure
      rescue Fluent::ConfigError => e
        $log.error "config error", file: @config_path, error: e
        $log.debug_backtrace
        exit!(1)
      ensure
        Fluent::Engine.dry_run_mode = false
      end
    end
604069
def set_count_auto(self, count=None):
        
        count = count or self._section.vars.CPU_CORES

        self._set('workers', count)

        return self._section
1369510
def set_dryrun(dryrun):
    
    r = _get_runez()
    old = r.DRYRUN
    r.DRYRUN = bool(dryrun)
    return old
583888
def run_on_change(paths)
      if @options[:dry_run]
        paths.each { |path| UI.info "Dry run: #{path}" }
        return
      end

      total    = 0
      failures = 0
      duration = 0
      run_command("mix test #{paths.join(' ')}") do |line|
        puts line
        if /Finished in ([0-9.]+) seconds/.match(line)
          duration = Regexp::last_match[1]
        elsif /([0-9]+) tests, ([0-9]+) failures/.match(line)
          total = Regexp::last_match[1].to_i
          failures = Regexp::last_match[2].to_i
        end
      end

      Notifier.notify(
        guard_message(total, failures, duration),
        :title => "Elixir results",
        :image => guard_image(failures),
        :priority => guard_priority(failures)
      )
    end
92563
def ensure_dry_running(state, project)
      dry_running = project.dry_running?
      has_finished = !project.version.nil?

      if (state == :is_in_progress && dry_running) ||
         (state == :is_not_in_progress && !dry_running) ||
         (state == :has_finished && !dry_running && has_finished)
        yield
      else
        if (state == :is_in_progress) ||
           (state == :has_finished && !dry_running && !has_finished)
          puts "Please start a dry-run first, use \"tetra dry-run\""
        elsif (state == :is_not_in_progress) ||
              (state == :has_finished && dry_running)
          puts "There is a dry-run in progress, please finish it (^D) or abort it (^C^D)"
        end
      end
    end
861644
def _set_worker_thread_level(self):
        
        bthread_logger = logging.getLogger(
            'google.cloud.logging.handlers.transports.background_thread')
        if self.debug_thread_worker:
            bthread_logger.setLevel(logging.DEBUG)
        else:
            bthread_logger.setLevel(logging.INFO)
220429
def set_host_finished(self, scan_id, target, host):
        
        finished_hosts = self.scans_table[scan_id]['finished_hosts']
        finished_hosts[target].extend(host)
        self.scans_table[scan_id]['finished_hosts'] = finished_hosts
45138
def upsert(version, stack, ini, dryrun, yaml, no_poll, work_directory):
    
    ini_data = read_config_info(ini)
    if 'environment' not in ini_data:
        print('[environment] section is required in the INI file')
        sys.exit(1)

    if version:
        ini_data['codeVersion'] = version
    else:
        ini_data['codeVersion'] = str(int(time.time()))

    if 'region' not in ini_data['environment']:
        ini_data['environment']['region'] = find_myself()

    ini_data['yaml'] = bool(yaml)
    ini_data['no_poll'] = bool(no_poll)
    ini_data['dryrun'] = bool(dryrun)

    if stack:
        ini_data['environment']['stack_name'] = stack

    if work_directory:
        try:
            os.chdir(work_directory)
        except Exception as wtf:
            logging.error(wtf)
            sys.exit(2)

    print(json.dumps(ini_data, indent=2))
    start_upsert(ini_data)
986634
public void markAllNewWorkers(Map<Integer, String> workerIds) {
        int startTime = TimeUtils.current_time_secs();

        for (Entry<Integer, String> entry : workerIds.entrySet()) {
            String oldWorkerIds = portToWorkerId.get(entry.getKey());
            if (oldWorkerIds != null) {
                workerIdToStartTimeAndPort.remove(oldWorkerIds);
                // update portToWorkerId
                LOG.info("A port is still occupied by an old worker, remove useless "
                        + oldWorkerIds + " from workerIdToStartTimeAndPort");
            }
            portToWorkerId.put(entry.getKey(), entry.getValue());
            workerIdToStartTimeAndPort.put(entry.getValue(), new Pair<>(startTime, entry.getKey()));
        }
    }
338228
def on_train_end(self, **kwargs: Any) -> None:  
        ""
        self.client.log_artifact(run_id=self.run, local_path=self.nb_path)
        self.client.set_terminated(run_id=self.run)
194197
def clean_global_runtime_state(reset_subsystem=False):
  
  if reset_subsystem:
    # Reset subsystem state.
    Subsystem.reset()

  # Reset Goals and Tasks.
  Goal.clear()

  # Reset global plugin state.
  BuildConfigInitializer.reset()
560030
def patch_all():
    
    global _patched

    if _patched:
        return
    _patched = True

    patch_default_retcodes()
    patch_worker_run_task()
    patch_worker_factory()
    patch_keepalive_run()
    patch_cmdline_parser()

    logger.debug("applied law-specific luigi patches")
153775
def run (self):
        """"""
        if self.all:
            # remove share directory
            directory = os.path.join("build", "share")
            if os.path.exists(directory):
                remove_tree(directory, dry_run=self.dry_run)
            else:
                log.warn("'%s' does not exist -- can't clean it", directory)
        clean.run(self)
246596
def cli(*, worker_settings, burst, check, watch, verbose):
    
    sys.path.append(os.getcwd())
    worker_settings = import_string(worker_settings)
    logging.config.dictConfig(default_log_config(verbose))

    if check:
        exit(check_health(worker_settings))
    else:
        kwargs = {} if burst is None else {'burst': burst}
        if watch:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(watch_reload(watch, worker_settings, loop))
        else:
            run_worker(worker_settings, **kwargs)
258330
def run_bulk(t, first, last, incr = 1)
                retry_until { DRMAA.run_bulk_jobs(t.ptr, first, last, incr) }
            end
749083
def configure(cls: typing.Type["ThreadPooled"], max_workers: typing.Optional[int] = None) -> None:
        
        if isinstance(cls.__executor, ThreadPoolExecutor):
            if cls.__executor.max_workers == max_workers:
                return
            cls.__executor.shutdown()

        cls.__executor = ThreadPoolExecutor(max_workers=max_workers)
228251
def bulk_record_workunits(self, engine_workunits):
    """"""
    for workunit in engine_workunits:
      duration = workunit['end_timestamp'] - workunit['start_timestamp']

      span = zipkin_span(
        service_name="pants",
        span_name=workunit['name'],
        duration=duration,
        span_storage=self.span_storage,
      )
      span.zipkin_attrs = ZipkinAttrs(
        trace_id=self.trace_id,
        span_id=workunit['span_id'],
        # TODO change it when we properly pass parent_id to the v2 engine Nodes
        # TODO Pass parent_id with ExecutionRequest when v2 engine is called by a workunit
        # TODO pass parent_id when v2 engine Node is called by another v2 engine Node
        parent_span_id=workunit.get("parent_id", self.parent_id),
        flags='0', # flags: stores flags header. Currently unused
        is_sampled=True,
      )
      span.start()
      span.start_timestamp = workunit['start_timestamp']
      span.stop()