Caching integration

docs/source/topics/monitoring.rst

@@ -77,9 +77,9 @@ COBalD-specific Monitoring
 Caching-specific Monitoring
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-.. TODO::
-
-    Will be added as soon as the caching branch is merged.
+.. autofunction:: lapis.monitor.caching.storage_status
+.. autofunction:: lapis.monitor.caching.storage_connection
+.. autofunction:: lapis.monitor.caching.remote_connection
 
 Telegraf
 --------

lapis/cachealgorithm.py

@@ -0,0 +1,57 @@
+from typing import Optional, Callable, Tuple
+
+from lapis.files import RequestedFile, StoredFile
+from lapis.storage import Storage
+from lapis.utilities.cache_cleanup_implementations import sort_files_by_cachedsince
+
+
+def check_size(file: RequestedFile, storage: Storage):
+    return storage.size >= file.filesize
+
+
+def check_relevance(file: RequestedFile, storage: Storage):
+    return True
+
+
+def delete_oldest(
+    file: RequestedFile, storage: Storage
+) -> Tuple[bool, Tuple[StoredFile]]:
+    deletable_files = []
+    currently_free = storage.free_space
+    if currently_free < storage.free_space:
+        sorted_files = sort_files_by_cachedsince(storage.files.items())
+        while currently_free < file.filesize:
+            deletable_files.append(next(sorted_files))
+            currently_free += deletable_files[-1].filesize
+    return True, tuple(deletable_files)
+
+
+def delete_oldest_few_used(
+    file: RequestedFile, storage: Storage
+) -> Tuple[bool, Optional[Tuple[StoredFile]]]:
+    deletable_files = []
+    currently_free = storage.free_space
+    if currently_free < storage.free_space:
+        sorted_files = sort_files_by_cachedsince(storage.files.items())
+        for current_file in sorted_files:
+            if current_file.numberofaccesses < 3:
+                deletable_files.append(current_file)
+                currently_free += deletable_files[-1].filesize
+                if currently_free >= file.filesize:
+                    return True, tuple(deletable_files)
+    return False, None
+
+
+class CacheAlgorithm(object):
+    def __init__(self, caching_strategy: Callable, deletion_strategy: Callable):
+        self._caching_strategy = lambda file, storage: check_size(
+            file, storage
+        ) and check_relevance(file, storage)
+        self._deletion_strategy = lambda file, storage: delete_oldest(file, storage)
+
+    def consider(
+        self, file: RequestedFile, storage: Storage
+    ) -> Tuple[bool, Optional[Tuple[StoredFile]]]:
+        if self._caching_strategy(file, storage):
+            return self._deletion_strategy(file, storage)
+        return False, None

lapis/cli/simulate.py

@@ -1,3 +1,5 @@
+from functools import partial
+
 import click
 import logging.handlers
 
@@ -9,6 +11,8 @@
 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 import Storage, HitrateStorage
+from lapis.storage_io.storage import storage_reader
 
 from lapis.scheduler import CondorJobScheduler
 from lapis.simulator import Simulator
@@ -25,18 +29,22 @@
 
 pool_import_mapper = {"htcondor": htcondor_pool_reader}
 
+storage_import_mapper = {"standard": storage_reader}
+
 
 @click.group()
 @click.option("--seed", type=int, default=1234)
 @click.option("--until", type=float)
 @click.option("--log-tcp", "log_tcp", is_flag=True)
 @click.option("--log-file", "log_file", type=click.File("w"))
 @click.option("--log-telegraf", "log_telegraf", is_flag=True)
+@click.option("--calculation-efficiency", type=float)
 @click.pass_context
-def cli(ctx, seed, until, log_tcp, log_file, log_telegraf):
+def cli(ctx, seed, until, log_tcp, log_file, log_telegraf, calculation_efficiency):
     ctx.ensure_object(dict)
     ctx.obj["seed"] = seed
     ctx.obj["until"] = until
+    ctx.obj["calculation_efficiency"] = calculation_efficiency
     monitoring_logger = logging.getLogger()
     monitoring_logger.setLevel(logging.DEBUG)
     time_filter = SimulationTimeFilter()
@@ -71,22 +79,51 @@ 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())),
+    ),
+    default=(None, None, None),
+)
+@click.option("--remote-throughput", "remote_throughput", type=float, default=10)
+@click.option("--cache-hitrate", "cache_hitrate", type=float, default=None)
 @click.pass_context
-def static(ctx, job_file, pool_file):
+def static(ctx, job_file, pool_file, storage_files, remote_throughput, cache_hitrate):
     click.echo("starting static environment")
     simulator = Simulator(seed=ctx.obj["seed"])
     file, file_type = job_file
     simulator.create_job_generator(
-        job_input=file, job_reader=job_import_mapper[file_type]
+        job_input=file,
+        job_reader=partial(
+            job_import_mapper[file_type],
+            calculation_efficiency=ctx.obj["calculation_efficiency"],
+        ),
     )
     simulator.create_scheduler(scheduler_type=CondorJobScheduler)
+
+    if all(storage_files):
+        simulator.create_connection_module(remote_throughput)
+        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],
+            storage_type=partial(HitrateStorage, cache_hitrate)
+            if cache_hitrate is not None
+            else Storage,
+        )
     for current_pool in pool_file:
         pool_file, pool_file_type = current_pool
         simulator.create_pools(
             pool_input=pool_file,
             pool_reader=pool_import_mapper[pool_file_type],
             pool_type=StaticPool,
         )
+    simulator.enable_monitoring()
     simulator.run(until=ctx.obj["until"])
 
 
@@ -108,7 +145,11 @@ def dynamic(ctx, job_file, pool_file):
     simulator = Simulator(seed=ctx.obj["seed"])
     file, file_type = job_file
     simulator.create_job_generator(
-        job_input=file, job_reader=job_import_mapper[file_type]
+        job_input=file,
+        job_reader=partial(
+            job_import_mapper[file_type],
+            calculation_efficiency=ctx.obj["calculation_efficiency"],
+        ),
     )
     simulator.create_scheduler(scheduler_type=CondorJobScheduler)
     for current_pool in pool_file:
@@ -119,6 +160,7 @@ def dynamic(ctx, job_file, pool_file):
             pool_type=Pool,
             controller=SimulatedLinearController,
         )
+    simulator.enable_monitoring()
     simulator.run(until=ctx.obj["until"])
 
 
@@ -146,7 +188,11 @@ def hybrid(ctx, job_file, static_pool_file, dynamic_pool_file):
     simulator = Simulator(seed=ctx.obj["seed"])
     file, file_type = job_file
     simulator.create_job_generator(
-        job_input=file, job_reader=job_import_mapper[file_type]
+        job_input=file,
+        job_reader=partial(
+            job_import_mapper[file_type],
+            calculation_efficiency=ctx.obj["calculation_efficiency"],
+        ),
     )
     simulator.create_scheduler(scheduler_type=CondorJobScheduler)
     for current_pool in static_pool_file:
@@ -164,6 +210,7 @@ def hybrid(ctx, job_file, static_pool_file, dynamic_pool_file):
             pool_type=Pool,
             controller=SimulatedLinearController,
         )
+    simulator.enable_monitoring()
     simulator.run(until=ctx.obj["until"])
 
 

lapis/connection.py

@@ -0,0 +1,137 @@
+import random
+
+from typing import Union
+from usim import Scope, time, Pipe
+
+from lapis.cachealgorithm import (
+    CacheAlgorithm,
+    check_size,
+    check_relevance,
+    delete_oldest_few_used,
+)
+from lapis.storage import Storage, RemoteStorage
+from lapis.files import RequestedFile
+from lapis.monitor import sampling_required
+
+
+class Connection(object):
+
+    __slots__ = ("storages", "remote_connection", "caching_algorithm")
+
+    def __init__(self, throughput=100):
+        self.storages = dict()
+        self.remote_connection = RemoteStorage(Pipe(throughput=throughput))
+        self.caching_algorithm = CacheAlgorithm(
+            caching_strategy=lambda file, storage: check_size(file, storage)
+            and check_relevance(file, storage),
+            deletion_strategy=lambda file, storage: delete_oldest_few_used(
+                file, storage
+            ),
+        )
+
+    def add_storage_element(self, storage_element: Storage):
+        """
+        Register storage element in Connetion module clustering storage elements by
+        sitename
+        :param storage_element:
+        :return:
+        """
+        storage_element.remote_connection = self.remote_connection
+        try:
+            self.storages[storage_element.sitename].append(storage_element)
+        except KeyError:
+            self.storages[storage_element.sitename] = [storage_element]
+
+    async def _determine_inputfile_source(
+        self, requested_file: RequestedFile, dronesite: str, job_repr: str
+    ) -> Union[Storage, RemoteStorage]:
+        """
+        Collects NamedTuples containing the amount of data of the requested file
+        cached in a storage element and the storage element for all reachable storage
+        objects on the drone's site. The tuples are sorted by amount of cached data
+        and the storage object where the biggest part of the file is cached is
+        returned. If the file is not cached in any storage object the connection module
+        remote connection is returned.
+        :param requested_file:
+        :param dronesite:
+        :param job_repr:
+        :return:
+        """
+        provided_storages = self.storages.get(dronesite, None)
+        if provided_storages is not None:
+            look_up_list = []
+            for storage in provided_storages:
+                look_up_list.append(storage.look_up_file(requested_file, job_repr))
+            storage_list = sorted(
+                [entry for entry in look_up_list], key=lambda x: x[0], reverse=True
+            )
+            for entry in storage_list:
+                # TODO: check should better check that size is bigger than requested
+                if entry.cached_filesize > 0:
+                    return entry.storage
+        return self.remote_connection
+
+    async def stream_file(self, requested_file: RequestedFile, dronesite, job_repr):
+        """
+        Determines which storage object is used to provide the requested file and
+        startes the files transfer. For files transfered via remote connection a
+        potential cache decides whether to cache the file and handles the caching
+        process.
+        :param requested_file:
+        :param dronesite:
+        :param job_repr:
+        :return:
+        """
+        used_connection = await self._determine_inputfile_source(
+            requested_file, dronesite, job_repr
+        )
+        await sampling_required.put(used_connection)
+        if used_connection == self.remote_connection and self.storages.get(
+            dronesite, None
+        ):
+            try:
+                potential_cache = random.choice(self.storages[dronesite])
+                cache_file, files_for_deletion = self.caching_algorithm.consider(
+                    file=requested_file, storage=potential_cache
+                )
+                if cache_file:
+                    for file in files_for_deletion:
+                        await potential_cache.remove(file, job_repr)
+                    await potential_cache.add(requested_file, job_repr)
+                else:
+                    print(
+                        f"APPLY CACHING DECISION: Job {job_repr}, File {requested_file.filename}: File wasnt "
+                        f"cached @ {time.now}"
+                    )
+            except KeyError:
+                pass
+        print(f"now transfering {requested_file.filesize} from {used_connection}")
+        await used_connection.transfer(requested_file, job_repr)
+        print(
+            "Job {}: finished transfering of file {}: {}GB @ {}".format(
+                job_repr, requested_file.filename, requested_file.filesize, time.now
+            )
+        )
+
+    async def transfer_files(self, drone, requested_files: dict, job_repr):
+        """
+        Converts dict information about requested files to RequestedFile object and
+        parallely launches streaming for all files
+        :param drone:
+        :param requested_files:
+        :param job_repr:
+        :return:
+        """
+        print("registered caches", self.storages)
+        start_time = time.now
+        async with Scope() as scope:
+            for inputfilename, inputfilespecs in requested_files.items():
+                requested_file = RequestedFile(
+                    inputfilename, inputfilespecs["usedsize"]
+                )
+                scope.do(self.stream_file(requested_file, drone.sitename, job_repr))
+        stream_time = time.now - start_time
+        print(
+            "STREAMED files {} in {}".format(list(requested_files.keys()), stream_time)
+        )
+        return stream_time

lapis/drone.py

@@ -1,7 +1,9 @@
 from cobald import interfaces
+from typing import Optional
 from usim import time, Scope, instant, Capacities, ResourcesUnavailable, Queue
 
 from lapis.job import Job
+from lapis.connection import Connection
 
 
 class ResourcesExceeded(Exception):
@@ -12,9 +14,11 @@ class Drone(interfaces.Pool):
     def __init__(
         self,
         scheduler,
-        pool_resources: dict,
-        scheduling_duration: float,
+        pool_resources: Optional[dict] = None,
+        scheduling_duration: Optional[float] = None,
         ignore_resources: list = None,
+        sitename: str = None,
+        connection: Connection = None,
     ):
         """
         :param scheduler:
@@ -23,6 +27,8 @@ def __init__(
         """
         super(Drone, self).__init__()
         self.scheduler = scheduler
+        self.connection = connection
+        self.sitename = sitename
         self.pool_resources = pool_resources
         self.resources = Capacities(**pool_resources)
         # shadowing requested resources to determine jobs to be killed
@@ -147,6 +153,11 @@ async def _run_job(self, job: Job, kill: bool):
                                 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()