Status quo caching extension

lapis/cli/simulate.py

@@ -9,6 +9,7 @@
 from lapis.pool import StaticPool, Pool
 from lapis.pool_io.htcondor import htcondor_pool_reader
 from lapis.job_io.swf import swf_job_reader
+from lapis.storage_io.storage import storage_reader
 
 from lapis.scheduler import CondorJobScheduler
 from lapis.simulator import Simulator
@@ -25,6 +26,8 @@
 
 pool_import_mapper = {"htcondor": htcondor_pool_reader}
 
+storage_import_mapper = {"standard": storage_reader}
+
 
 @click.group()
 @click.option("--seed", type=int, default=1234)
@@ -71,8 +74,17 @@ def cli(ctx, seed, until, log_tcp, log_file, log_telegraf):
     type=(click.File("r"), click.Choice(list(pool_import_mapper.keys()))),
     multiple=True,
 )
+@click.option(
+    "--storage-files",
+    "storage_files",
+    type=(
+        click.File("r"),
+        click.File("r"),
+        click.Choice(list(storage_import_mapper.keys())),
+    ),
+)
 @click.pass_context
-def static(ctx, job_file, pool_file):
+def static(ctx, job_file, pool_file, storage_files):
     click.echo("starting static environment")
     simulator = Simulator(seed=ctx.obj["seed"])
     file, file_type = job_file
@@ -87,6 +99,12 @@ def static(ctx, job_file, pool_file):
             pool_reader=pool_import_mapper[pool_file_type],
             pool_type=StaticPool,
         )
+    storage_file, storage_content_file, storage_type = storage_files
+    simulator.create_storage(
+        storage_input=storage_file,
+        storage_content_input=storage_content_file,
+        storage_reader=storage_import_mapper[storage_type],
+    )
     simulator.run(until=ctx.obj["until"])
 
 

lapis/drone.py

@@ -15,6 +15,7 @@ def __init__(
         pool_resources: dict,
         scheduling_duration: float,
         ignore_resources: list = None,
+        sitename: str = None,
     ):
         """
         :param scheduler:
@@ -23,6 +24,7 @@ def __init__(
         """
         super(Drone, self).__init__()
         self.scheduler = scheduler
+        self.sitename = sitename
         self.pool_resources = pool_resources
         self.resources = Capacities(**pool_resources)
         # shadowing requested resources to determine jobs to be killed
@@ -140,6 +142,11 @@ async def start_job(self, job: Job, kill: bool = False):
                                 pass
                     self.scheduler.update_drone(self)
                     await job_execution.done
+                    print(
+                        "finished job {} on drone {} @ {}".format(
+                            repr(job), repr(self), time.now
+                        )
+                    )
             except ResourcesUnavailable:
                 await instant
                 job_execution.cancel()

lapis/file_provider.py

@@ -0,0 +1,37 @@
+from lapis.storage import Storage
+
+
+class FileProvider(object):
+
+    __slots__ = ("storages",)
+
+    def __init__(self):
+        self.storages = dict()
+
+    def add_storage_element(self, storage_element: Storage):
+        try:
+            self.storages[storage_element.sitename].append(storage_element)
+        except KeyError:
+            self.storages[storage_element.sitename] = [storage_element]
+
+    def provides_all_files(self, job):
+        """
+        Dummy implementation, to be replaced: if a part of every inputfile of the job is
+        provided by a storage element located on the same site as the drone the job
+        is running on this function returns True
+        :param job:
+        :return:
+        """
+        provided_storages = self.storages.get(job.drone.sitename, None)
+        if provided_storages:
+            for inputfilename, inputfilespecs in job.used_inputfiles.items():
+                provides_inputfile = 0
+                for storage in provided_storages:
+                    provides_inputfile += storage.provides_file(
+                        (inputfilename, inputfilespecs)
+                    )
+                if not provides_inputfile:
+                    return False
+            return True
+        else:
+            return False

lapis/job.py

@@ -5,6 +5,7 @@
 from usim import CancelTask
 
 from lapis.monitor import sampling_required
+from lapis.utilities.walltime_models import walltime_models
 
 if TYPE_CHECKING:
     from lapis.drone import Drone
@@ -14,7 +15,7 @@ class Job(object):
     __slots__ = (
         "resources",
         "used_resources",
-        "walltime",
+        "_walltime",
         "requested_walltime",
         "queue_date",
         "requested_inputfiles",
@@ -23,6 +24,7 @@ class Job(object):
         "in_queue_until",
         "_name",
         "drone",
+        "fileprovider",
         "_success",
     )
 
@@ -59,7 +61,7 @@ def __init__(
                     self.used_resources[key],
                 )
                 self.resources[key] = self.used_resources[key]
-        self.walltime = used_resources.pop("walltime")
+        self._walltime = used_resources.pop("walltime")
         self.requested_walltime = resources.pop("walltime", None)
         self.requested_inputfiles = resources.pop("inputfiles", None)
         self.used_inputfiles = used_resources.pop("inputfiles", None)
@@ -68,6 +70,7 @@ def __init__(
         self.in_queue_since = in_queue_since
         self.in_queue_until = None
         self.drone = drone
+        self.fileprovider = None
         self._name = name
         self._success: Optional[bool] = None
 
@@ -91,10 +94,23 @@ def waiting_time(self) -> float:
             return self.in_queue_until - self.in_queue_since
         return float("Inf")
 
+    @property
+    def walltime(self):
+        if self.fileprovider and self.fileprovider.provides_all_files(self):
+            return walltime_models["maxeff"](self, self._walltime)
+        else:
+            return self._walltime
+
     async def run(self):
         self.in_queue_until = time.now
         self._success = None
         await sampling_required.put(self)
+        if self.drone:
+            print(
+                "running job {} on site {} in drone {}".format(
+                    repr(self), self.drone.sitename, repr(self.drone)
+                )
+            )
         try:
             await (time + self.walltime)
         except CancelTask:

lapis/pool_io/htcondor.py

@@ -48,5 +48,6 @@ def htcondor_pool_reader(
                     for key, value in resource_name_mapping.items()
                 },
                 ignore_resources=["disk"],
+                sitename=row.get("sitename", None),
             ),
         )

lapis/scheduler.py

@@ -1,5 +1,5 @@
 from typing import Dict
-from usim import Scope, interval
+from usim import Scope, interval, time
 
 from lapis.drone import Drone
 from lapis.monitor import sampling_required
@@ -26,9 +26,10 @@ class CondorJobScheduler(object):
     :return:
     """
 
-    def __init__(self, job_queue):
+    def __init__(self, job_queue, fileprovider):
         self._stream_queue = job_queue
         self.drone_cluster = []
+        self.fileprovider = fileprovider
         self.interval = 60
         self.job_queue = JobQueue()
         self._collecting = True
@@ -90,6 +91,12 @@ async def run(self):
                 for job in self.job_queue:
                     best_match = self._schedule_job(job)
                     if best_match:
+                        job.fileprovider = self.fileprovider
+                        print(
+                            "start job {} on drone {} @ {}".format(
+                                repr(job), repr(best_match), time.now
+                            )
+                        )
                         scope.do(best_match.start_job(job))
                         self.job_queue.remove(job)
                         await sampling_required.put(self.job_queue)

lapis/simulator.py

@@ -6,6 +6,7 @@
 
 from lapis.drone import Drone
 from lapis.job import job_to_queue_scheduler
+from lapis.file_provider import FileProvider
 from lapis.monitor.general import (
     user_demand,
     job_statistics,
@@ -26,6 +27,8 @@ def __init__(self, seed=1234):
         random.seed(seed)
         self.job_queue = Queue()
         self.pools = []
+        self.storage_list = []
+        self.fileprovider = FileProvider()
         self.controllers = []
         self.job_scheduler = None
         self.job_generator = None
@@ -59,8 +62,17 @@ def create_pools(self, pool_input, pool_reader, pool_type, controller=None):
             if controller:
                 self.controllers.append(controller(target=pool, rate=1))
 
+    def create_storage(self, storage_input, storage_content_input, storage_reader):
+        for storage in storage_reader(
+            storage=storage_input, storage_content=storage_content_input
+        ):
+            self.storage_list.append(storage)
+            self.fileprovider.add_storage_element(storage)
+
     def create_scheduler(self, scheduler_type):
-        self.job_scheduler = scheduler_type(job_queue=self.job_queue)
+        self.job_scheduler = scheduler_type(
+            job_queue=self.job_queue, fileprovider=self.fileprovider
+        )
 
     def run(self, until=None):
         print(f"running until {until}")

lapis/storage.py

@@ -0,0 +1,81 @@
+from usim import time
+
+from typing import Optional
+
+from lapis.utilities.cache_algorithm_implementations import cache_algorithm
+from lapis.utilities.cache_cleanup_implementations import cache_cleanup
+
+
+class Storage(object):
+
+    __slots__ = ("name", "sitename", "storagesize", "usedstorage", "content")
+
+    def __init__(
+        self, name: str, sitename: str, storagesize: int, content: Optional[dict] = None
+    ):
+        self.name = name
+        self.sitename = sitename
+        self.storagesize = storagesize
+        self.content = content
+        self.usedstorage = self.get_used_storage()
+        self.describe_state()
+
+    def get_used_storage(self):
+        return sum(subdict["usedsize"] for subdict in self.content.values())
+
+    def free_space(self):
+        return self.storagesize - self.usedstorage
+
+    def place_new_file(self, filerequest: tuple):
+        filename, filespecs = filerequest
+        if self.free_space() - filespecs["usedsize"] < 0:
+            self.make_room(self.free_space() - filespecs["usedsize"])
+        self.content.update({filename: filespecs})
+        self.content[filename].update(
+            cachedsince=time.now, lastaccessed=time.now, numberofaccesses=0
+        )
+        self.usedstorage = self.get_used_storage()
+
+    def update_file(self, filerequest: tuple):
+        filename, filespecs = filerequest
+        requested_file = filename
+        filesize_difference = (
+            filespecs["usedsize"] - self.content[requested_file]["usedsize"]
+        )
+        if filesize_difference > 0:
+            self.make_room(filesize_difference)
+            self.content[requested_file]["usedsize"] += filesize_difference
+        self.content[requested_file]["lastaccessed"] = time.now
+        self.content[requested_file]["numberofaccesses"] += 1
+        self.usedstorage = self.get_used_storage()
+
+    def make_room(self, filesize_difference: int):
+        if self.free_space() - filesize_difference < 0:
+            cache_cleanup["fifo"](filesize_difference, self)
+
+    def provides_file(self, filerequest: dict):
+        filename, filespecs = filerequest
+        if filename in self.content.keys():
+            self.update_file(filerequest)
+            return True
+        else:
+            if self.cache_file():
+                self.place_new_file(filerequest)
+            return False
+
+    def cache_file(self):
+        # cache everything, test different implementations
+        return cache_algorithm["standard"]()
+
+    def describe_state(self):
+        print(
+            "{name} on site {site}: {used}MB of {tot}MB used ({div} %), contains "
+            "files {filelist}".format(
+                name=self.name,
+                site=self.sitename,
+                used=self.usedstorage,
+                tot=self.storagesize,
+                div=100.0 * self.usedstorage / self.storagesize,
+                filelist=", ".join(self.content.keys()),
+            )
+        )

lapis/storage_io/storage.py

@@ -0,0 +1,34 @@
+import csv
+from lapis.storage import Storage
+
+
+def storage_reader(storage, storage_content):
+    storage_content = storage_content_reader(storage_content)
+    reader = csv.DictReader(storage, delimiter=" ", quotechar="'")
+    for row in reader:
+        yield Storage(
+            name=row["name"],
+            sitename=row["sitename"],
+            storagesize=int(row["cachesizeMB"]),
+            content=storage_content[row["name"]],
+        )
+
+
+def storage_content_reader(file_name):
+    reader = csv.DictReader(file_name, delimiter=" ", quotechar="'")
+    cache_information = dict()
+    for row in reader:
+        if row["cachename"] not in cache_information.keys():
+            cache_information[row["cachename"]] = dict()
+        cache_information[row["cachename"]][row["filename"]] = dict()
+        for key in [
+            "filesize",
+            "usedsize",
+            "cachedsince",
+            "lastaccessed",
+            "numberofaccesses",
+        ]:
+            cache_information[row["cachename"]][row["filename"]][key] = int(row[key])
+    if not cache_information:
+        cache_information = None
+    return cache_information

lapis/utilities/cache_algorithm_implementations.py

@@ -0,0 +1,5 @@
+def cache_all():
+    return True
+
+
+cache_algorithm = {"standard": cache_all}