docstr: Job scheduling by optimization of resource usage by solving ILP using pulp
docstr_tok: ["Job","scheduling","by","optimization","of","resource","usage","by","solving","ILP","using","pulp"]
code: def job_selector_ilp(self, jobs):
      """
      Job scheduling by optimization of resource usage by solving ILP using pulp
      """
      import pulp
      from pulp import lpSum
      from stopit import ThreadingTimeout as Timeout, TimeoutException
  
      if len(jobs) == 1:
          logger.debug(
              "Using greedy selector because only single job has to be scheduled."
          )
          return self.job_selector_greedy(jobs)
  
      with self._lock:
          if not self.resources["_cores"]:
              return set()
  
          # assert self.resources["_cores"] > 0
          scheduled_jobs = {
              job: pulp.LpVariable(
                  f"job_{idx}",
                  lowBound=0,
                  upBound=1,
                  cat=pulp.LpInteger,
              )
              for idx, job in enumerate(jobs)
          }
  
          temp_files = {
              temp_file
              for job in jobs
              for temp_file in self.workflow.dag.temp_input(job)
          }
  
          async def get_temp_sizes_gb():
              return {f: (await f.size()) / 1e9 for f in temp_files}
  
          temp_sizes_gb = (
              defaultdict(int)
              if self.touch
              else async_run(get_temp_sizes_gb())
          )
  
          temp_job_improvement = {
              temp_file: pulp.LpVariable(
                  f"temp_file_{idx}",
                  lowBound=0,
                  upBound=1,
                  cat="Continuous",
              )
              for idx, temp_file in enumerate(temp_files)
          }
  
          temp_file_deletable = {
              temp_file: pulp.LpVariable(
                  f"deletable_{idx}",
                  lowBound=0,
                  upBound=1,
                  cat=pulp.LpInteger,
              )
              for idx, temp_file in enumerate(temp_files)
          }
          prob = pulp.LpProblem("JobScheduler", pulp.LpMaximize)
  
          total_temp_size = max(
              sum(
                  [temp_sizes_gb[temp_file] for temp_file in temp_files]
              ),
              1,
          )
          total_core_requirement = sum(
              [
                  max(job.scheduler_resources.get("_cores", 1), 1)
                  for job in jobs
              ]
          )
          # Objective function
          # Job priority > Core load
          # Core load > temp file removal
          # Instant removal > temp size
          prob += (
              2
              * total_core_requirement
              * 2
              * total_temp_size
              * lpSum(
                  [job.priority * scheduled_jobs[job] for job in jobs]
              )
              + 2
              * total_temp_size
              * lpSum(
                  [
                      max(job.scheduler_resources.get("_cores", 1), 1)
                      * scheduled_jobs[job]
                      for job in jobs
                  ]
              )
              + total_temp_size
              * lpSum(
                  [
                      temp_file_deletable[temp_file]
                      * temp_sizes_gb[temp_file]
                      for temp_file in temp_files
                  ]
              )
              + lpSum(
                  [
                      temp_job_improvement[temp_file]
                      * temp_sizes_gb[temp_file]
                      for temp_file in temp_files
                  ]
              )
          )
  
          # Constraints:
          for name in self.workflow.global_resources:
              prob += (
                  lpSum(
                      [
                          scheduled_jobs[job]
                          * job.scheduler_resources.get(name, 0)
                          for job in jobs
                      ]
                  )
                  <= self.resources[name]
              )
  
          # Choose jobs that lead to "fastest" (minimum steps) removal of existing temp file
          remaining_jobs = self.remaining_jobs
          for temp_file in temp_files:
              prob += temp_job_improvement[temp_file] <= lpSum(
                  [
                      scheduled_jobs[job]
                      * self.required_by_job(temp_file, job)
                      for job in jobs
                  ]
              ) / lpSum(
                  [
                      self.required_by_job(temp_file, job)
                      for job in remaining_jobs
                  ]
              )
  
              prob += (
                  temp_file_deletable[temp_file]
                  <= temp_job_improvement[temp_file]
              )
  
      try:
          with Timeout(10, swallow_exc=False):
              self._solve_ilp(prob)
      except TimeoutException as e:
          logger.warning(
              "Failed to solve scheduling problem with ILP solver in time (10s). "
              "Falling back to greedy solver."
          )
          return self.job_selector_greedy(jobs)
      except pulp.apis.core.PulpSolverError as e:
          logger.warning(
              "Failed to solve scheduling problem with ILP solver. Falling back to greedy solver. "
              "Run Snakemake with --verbose to see the full solver output for debugging the problem."
          )
          return self.job_selector_greedy(jobs)
  
      selected_jobs = set(
          job
          for job, variable in scheduled_jobs.items()
          if math.isclose(variable.value(), 1.0)
      )
  
      if not selected_jobs:
          # No selected jobs. This could be due to insufficient resources or a failure in the ILP solver
          # Hence, we silently fall back to the greedy solver to make sure that we don't miss anything.
          return self.job_selector_greedy(jobs)
  
      for name in self.workflow.global_resources:
          self.resources[name] -= sum(
              [
                  job.scheduler_resources.get(name, 0)
                  for job in selected_jobs
              ]
          )
      return selected_jobs
  
imports: import asyncio
  from collections import defaultdict
  import math
  import os, signal, sys
  import threading
  from functools import partial
  from itertools import chain, accumulate
  from contextlib import ContextDecorator
  from snakemake_interface_executor_plugins.scheduler import JobSchedulerExecutorInterface
  from snakemake_interface_executor_plugins.registry import ExecutorPluginRegistry
  from snakemake_interface_executor_plugins.registry import Plugin
  from snakemake_interface_executor_plugins.settings import ExecMode
  from snakemake.common import async_run
  from snakemake.exceptions import RuleException, WorkflowError, print_exception
  from snakemake.logging import logger
  from fractions import Fraction
label_desc: Use image node_id 17 for calling the JobScheduler obj's underlying member method code with example usage: obj.job_selector_ilp(jobs) and returns: selected_jobs, self, self, set, unknown, self, self