base.py 17.9 KB
Newer Older
1 2 3
#!/usr/bin/env python
# vim: set fileencoding=utf-8 :

4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34
###################################################################################
#                                                                                 #
# Copyright (c) 2019 Idiap Research Institute, http://www.idiap.ch/               #
# Contact: beat.support@idiap.ch                                                  #
#                                                                                 #
# Redistribution and use in source and binary forms, with or without              #
# modification, are permitted provided that the following conditions are met:     #
#                                                                                 #
# 1. Redistributions of source code must retain the above copyright notice, this  #
# list of conditions and the following disclaimer.                                #
#                                                                                 #
# 2. Redistributions in binary form must reproduce the above copyright notice,    #
# this list of conditions and the following disclaimer in the documentation       #
# and/or other materials provided with the distribution.                          #
#                                                                                 #
# 3. Neither the name of the copyright holder nor the names of its contributors   #
# may be used to endorse or promote products derived from this software without   #
# specific prior written permission.                                              #
#                                                                                 #
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND #
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED   #
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE          #
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE    #
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL      #
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR      #
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER      #
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,   #
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE   #
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.            #
#                                                                                 #
###################################################################################
35 36


37 38 39 40 41 42 43
"""
====
base
====

Execution utilities
"""
44
import collections
Samuel GAIST's avatar
Samuel GAIST committed
45
import glob
46
import logging
Samuel GAIST's avatar
Samuel GAIST committed
47
import os
48

Samuel GAIST's avatar
Samuel GAIST committed
49
import simplejson as json
50 51 52

from beat.backend.python.helpers import convert_experiment_configuration_to_container

Samuel GAIST's avatar
Samuel GAIST committed
53 54 55 56
from .. import algorithm
from .. import database
from .. import schema
from .. import stats
57

58 59 60
logger = logging.getLogger(__name__)


61
class BaseExecutor(object):
Philip ABBET's avatar
Philip ABBET committed
62
    """Executors runs the code given an execution block information
63 64


Philip ABBET's avatar
Philip ABBET committed
65
    Parameters:
66

Philip ABBET's avatar
Philip ABBET committed
67
      prefix (str): Establishes the prefix of your installation.
68

Philip ABBET's avatar
Philip ABBET committed
69 70 71 72
      data (dict, str): The piece of data representing the block to be executed.
        It must validate against the schema defined for execution blocks. If a
        string is passed, it is supposed to be a fully qualified absolute path to
        a JSON file containing the block execution information.
73

André Anjos's avatar
André Anjos committed
74
      cache (:py:class:`str`, Optional): If your cache is not located under
Philip ABBET's avatar
Philip ABBET committed
75 76
        ``<prefix>/cache``, then specify a full path here. It will be used
        instead.
77

André Anjos's avatar
André Anjos committed
78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104
      dataformat_cache (:py:class:`dict`, Optional): A dictionary mapping
        dataformat names to loaded dataformats. This parameter is optional and,
        if passed, may greatly speed-up database loading times as dataformats
        that are already loaded may be re-used. If you use this parameter, you
        must guarantee that the cache is refreshed as appropriate in case the
        underlying dataformats change.

      database_cache (:py:class:`dict`, Optional): A dictionary mapping
        database names to loaded databases. This parameter is optional and, if
        passed, may greatly speed-up database loading times as databases that
        are already loaded may be re-used. If you use this parameter, you must
        guarantee that the cache is refreshed as appropriate in case the
        underlying databases change.

      algorithm_cache (:py:class:`dict`, Optional): A dictionary mapping
        algorithm names to loaded algorithms. This parameter is optional and,
        if passed, may greatly speed-up database loading times as algorithms
        that are already loaded may be re-used. If you use this parameter, you
        must guarantee that the cache is refreshed as appropriate in case the
        underlying algorithms change.

      library_cache (:py:class:`dict`, Optional): A dictionary mapping library
        names to loaded libraries. This parameter is optional and, if passed,
        may greatly speed-up library loading times as libraries that are
        already loaded may be re-used. If you use this parameter, you must
        guarantee that the cache is refreshed as appropriate in case the
        underlying libraries change.
105 106


Philip ABBET's avatar
Philip ABBET committed
107
    Attributes:
108

Philip ABBET's avatar
Philip ABBET committed
109
      cache (str): The path to the cache currently being used
110

Philip ABBET's avatar
Philip ABBET committed
111 112
      errors (list): A list containing errors found while loading this execution
        block.
113

Philip ABBET's avatar
Philip ABBET committed
114 115
      data (dict): The original data for this executor, as loaded by our JSON
        decoder.
116

Philip ABBET's avatar
Philip ABBET committed
117 118
      algorithm (beat.core.algorithm.Algorithm): An object representing the
        algorithm to be run.
119

Philip ABBET's avatar
Philip ABBET committed
120
      databases (dict): A dictionary in which keys are strings with database
André Anjos's avatar
André Anjos committed
121
        names and values are :py:class:`.database.Database`, representing the
Philip ABBET's avatar
Philip ABBET committed
122 123
        databases required for running this block. The dictionary may be empty
        in case all inputs are taken from the file cache.
124

Philip ABBET's avatar
Philip ABBET committed
125 126 127 128
      views (dict): A dictionary in which the keys are tuples pointing to the
        ``(<database-name>, <protocol>, <set>)`` and the value is a setup view
        for that particular combination of details. The dictionary may be empty
        in case all inputs are taken from the file cache.
129

130 131
      input_list (beat.backend.python.inputs.InputList): A list of inputs that
        will be served to the algorithm.
132

133 134
      output_list (beat.backend.python.outputs.OutputList): A list of outputs
        that the algorithm will produce.
135

Philip ABBET's avatar
Philip ABBET committed
136 137
      data_sources (list): A list with all data-sources created by our execution
        loader.
138

Philip ABBET's avatar
Philip ABBET committed
139 140
      data_sinks (list): A list with all data-sinks created by our execution
        loader. These are useful for clean-up actions in case of problems.
141

Philip ABBET's avatar
Philip ABBET committed
142
    """
143

144 145 146 147 148 149 150 151 152 153 154
    def __init__(
        self,
        prefix,
        data,
        cache=None,
        dataformat_cache=None,
        database_cache=None,
        algorithm_cache=None,
        library_cache=None,
        custom_root_folders=None,
    ):
Philip ABBET's avatar
Philip ABBET committed
155 156 157

        # Initialisations
        self.prefix = prefix
158
        self.cache = cache or os.path.join(self.prefix, "cache")
Philip ABBET's avatar
Philip ABBET committed
159
        self.algorithm = None
160
        self.loop_algorithm = None
Philip ABBET's avatar
Philip ABBET committed
161 162
        self.databases = {}
        self.input_list = None
163
        self.data_loaders = None
Philip ABBET's avatar
Philip ABBET committed
164 165 166 167
        self.output_list = None
        self.data_sinks = []
        self.errors = []
        self.data = data
168
        self.debug = False
169

Philip ABBET's avatar
Philip ABBET committed
170 171 172
        # Check that the cache path exists
        if not os.path.exists(self.cache):
            raise IOError("Cache path `%s' does not exist" % self.cache)
173

Philip ABBET's avatar
Philip ABBET committed
174 175 176 177 178 179
        # Check the custom root folders
        if custom_root_folders is not None:
            if not isinstance(custom_root_folders, collections.Mapping):
                raise TypeError("The custom root folders must be in dictionary format")
        else:
            custom_root_folders = {}
180

Philip ABBET's avatar
Philip ABBET committed
181 182 183 184 185
        # Temporary caches, if the user has not set them, for performance
        database_cache = database_cache if database_cache is not None else {}
        dataformat_cache = dataformat_cache if dataformat_cache is not None else {}
        algorithm_cache = algorithm_cache if algorithm_cache is not None else {}
        library_cache = library_cache if library_cache is not None else {}
186

Philip ABBET's avatar
Philip ABBET committed
187 188 189
        # Basic validation of the data declaration, including JSON loading if required
        if not isinstance(data, dict):
            if not os.path.exists(data):
190
                self.errors.append("File not found: %s" % data)
Philip ABBET's avatar
Philip ABBET committed
191
                return
192

193
        self.data, self.errors = schema.validate("execution", data)
Philip ABBET's avatar
Philip ABBET committed
194 195
        if self.errors:
            return
196

Philip ABBET's avatar
Philip ABBET committed
197
        # Load the algorithm (using the algorithm cache if possible)
198 199
        if self.data["algorithm"] in algorithm_cache:
            self.algorithm = algorithm_cache[self.data["algorithm"]]
Philip ABBET's avatar
Philip ABBET committed
200
        else:
201 202 203
            self.algorithm = algorithm.Algorithm(
                self.prefix, self.data["algorithm"], dataformat_cache, library_cache
            )
Philip ABBET's avatar
Philip ABBET committed
204
            algorithm_cache[self.algorithm.name] = self.algorithm
205

Philip ABBET's avatar
Philip ABBET committed
206 207 208
        if not self.algorithm.valid:
            self.errors += self.algorithm.errors
            return
209

210 211 212 213
        if "loop" in self.data:
            loop = self.data["loop"]
            if loop["algorithm"] in algorithm_cache:
                self.loop_algorithm = algorithm_cache[loop["algorithm"]]
214
            else:
215 216 217
                self.loop_algorithm = algorithm.Algorithm(
                    self.prefix, loop["algorithm"], dataformat_cache, library_cache
                )
218 219
                algorithm_cache[self.loop_algorithm.name] = self.loop_algorithm

220 221 222 223
                if len(loop["inputs"]) != len(self.loop_algorithm.input_map):
                    self.errors.append(
                        "The number of inputs of the loop algorithm doesn't correspond"
                    )
224

225
                for name in self.data["inputs"].keys():
226
                    if name not in self.algorithm.input_map.keys():
227 228 229
                        self.errors.append(
                            "The input '%s' doesn't exist in the loop algorithm" % name
                        )
230

231 232 233 234 235 236 237 238 239 240 241
                if len(loop["outputs"]) != len(self.loop_algorithm.output_map):
                    self.errors.append(
                        "The number of outputs of the loop algorithm doesn't correspond"
                    )

                for name in self.data["outputs"].keys():
                    if name not in self.algorithm.output_map.keys():
                        self.errors.append(
                            "The output '%s' doesn't exist in the loop algorithm" % name
                        )

Philip ABBET's avatar
Philip ABBET committed
242
        # Check that the mapping in coherent
243 244 245 246 247 248 249 250 251 252 253 254 255
        if len(self.data["inputs"]) != len(self.algorithm.input_map):
            self.errors.append(
                "The number of inputs of the algorithm doesn't correspond"
            )

        if "outputs" in self.data and (
            len(self.data["outputs"]) != len(self.algorithm.output_map)
        ):
            self.errors.append(
                "The number of outputs of the algorithm doesn't correspond"
            )

        for name in self.data["inputs"].keys():
Philip ABBET's avatar
Philip ABBET committed
256
            if name not in self.algorithm.input_map.keys():
257 258 259
                self.errors.append(
                    "The input '%s' doesn't exist in the algorithm" % name
                )
260

261 262
        if "outputs" in self.data:
            for name in self.data["outputs"].keys():
Philip ABBET's avatar
Philip ABBET committed
263
                if name not in self.algorithm.output_map.keys():
264 265 266
                    self.errors.append(
                        "The output '%s' doesn't exist in the algorithm" % name
                    )
267

268 269
        if "loop" in self.data:
            for name in ["request", "answer"]:
270
                if name not in self.algorithm.loop_map.keys():
271 272 273
                    self.errors.append(
                        "The loop '%s' doesn't exist in the algorithm" % name
                    )
274

Philip ABBET's avatar
Philip ABBET committed
275 276
        if self.errors:
            return
277

Philip ABBET's avatar
Philip ABBET committed
278
        # Load the databases (if any is required)
279 280 281 282 283 284 285 286 287 288 289
        self._update_db_cache(
            self.data["inputs"], custom_root_folders, database_cache, dataformat_cache
        )

        if "loop" in self.data:
            self._update_db_cache(
                self.data["loop"]["inputs"],
                custom_root_folders,
                database_cache,
                dataformat_cache,
            )
290

Philip ABBET's avatar
Philip ABBET committed
291 292
    def __enter__(self):
        """Prepares inputs and outputs for the processing task
293

Philip ABBET's avatar
Philip ABBET committed
294
        Raises:
295

Philip ABBET's avatar
Philip ABBET committed
296
          IOError: in case something cannot be properly setup
297

Philip ABBET's avatar
Philip ABBET committed
298
        """
299

300 301
        logger.info("Start the execution of '%s'", self.algorithm.name)

Philip ABBET's avatar
Philip ABBET committed
302
        return self
303

Philip ABBET's avatar
Philip ABBET committed
304 305 306
    def __exit__(self, exc_type, exc_value, traceback):
        """Closes all sinks and disconnects inputs and outputs
        """
307

Philip ABBET's avatar
Philip ABBET committed
308 309 310 311 312
        for sink in self.data_sinks:
            # we save the output only if no valid error has been thrown
            # n.b.: a system exit will raise SystemExit which is not an Exception
            if not isinstance(exc_type, Exception):
                sink.close()
313

Philip ABBET's avatar
Philip ABBET committed
314
        self.input_list = None
315
        self.data_loaders = []
Philip ABBET's avatar
Philip ABBET committed
316 317
        self.output_list = None
        self.data_sinks = []
318

319 320 321
    def _update_db_cache(
        self, inputs, custom_root_folders, database_cache, dataformat_cache
    ):
322 323 324
        """ Update the database cache based on the input list given"""

        for name, details in inputs.items():
325
            if "database" in details:
326

327
                if details["database"] not in self.databases:
328

329 330
                    if details["database"] in database_cache:
                        db = database_cache[details["database"]]
331
                    else:
332 333 334
                        db = database.Database(
                            self.prefix, details["database"], dataformat_cache
                        )
335 336 337

                        name = "database/%s" % db.name
                        if name in custom_root_folders:
338
                            db.data["root_folder"] = custom_root_folders[name]
339 340 341 342 343 344 345 346

                        database_cache[db.name] = db

                    self.databases[db.name] = db

                    if not db.valid:
                        self.errors += db.errors

Philip ABBET's avatar
Philip ABBET committed
347 348
    def _prepare_inputs(self):
        """Prepares all input required by the execution."""
349

Philip ABBET's avatar
Philip ABBET committed
350
        raise NotImplementedError()
351

Philip ABBET's avatar
Philip ABBET committed
352 353
    def _prepare_outputs(self):
        """Prepares all output required by the execution."""
354

Philip ABBET's avatar
Philip ABBET committed
355
        raise NotImplementedError()
356

357 358 359
    def process(
        self, virtual_memory_in_megabytes=0, max_cpu_percent=0, timeout_in_minutes=0
    ):
Philip ABBET's avatar
Philip ABBET committed
360
        """Executes the user algorithm code
361

Philip ABBET's avatar
Philip ABBET committed
362
        Parameters:
363

André Anjos's avatar
André Anjos committed
364 365 366
          virtual_memory_in_megabytes (:py:class:`int`, Optional): The amount
            of virtual memory (in Megabytes) available for the job. If set to
            zero, no limit will be applied.
367

André Anjos's avatar
André Anjos committed
368 369 370 371 372
          max_cpu_percent (:py:class:`int`, Optional): The maximum amount of
            CPU usage allowed in a system. This number must be an integer
            number between 0 and ``100*number_of_cores`` in your system. For
            instance, if your system has 2 cores, this number can go between 0
            and 200. If it is <= 0, then we don't track CPU usage.
373

Philip ABBET's avatar
Philip ABBET committed
374 375
          timeout_in_minutes (int): The number of minutes to wait for the user
            process to execute. After this amount of time, the user process is
André Anjos's avatar
André Anjos committed
376 377
            killed with ``signal.SIGKILL``. If set to zero, no timeout will be
            applied.
378

Philip ABBET's avatar
Philip ABBET committed
379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396
        Returns:

          dict: A dictionary which is JSON formattable containing the summary of
            this block execution.

        """

        raise NotImplementedError()

    @property
    def valid(self):
        """A boolean that indicates if this executor is valid or not"""

        return not bool(self.errors)

    @property
    def analysis(self):
        """A boolean that indicates if the current block is an analysis block"""
397
        return "result" in self.data
Philip ABBET's avatar
Philip ABBET committed
398 399 400 401 402 403

    @property
    def outputs_exist(self):
        """Returns ``True`` if outputs this block is supposed to produce exists."""

        if self.analysis:
404 405 406
            path = os.path.join(self.cache, self.data["result"]["path"]) + "*"
            if not glob.glob(path):
                return False
Philip ABBET's avatar
Philip ABBET committed
407 408

        else:
409 410 411 412
            for name, details in self.data["outputs"].items():
                path = os.path.join(self.cache, details["path"]) + "*"
                if not glob.glob(path):
                    return False
Philip ABBET's avatar
Philip ABBET committed
413 414 415 416 417 418 419 420 421 422 423 424

        # if you get to this point all outputs already exist
        return True

    @property
    def io_statistics(self):
        """Summarize current I/O statistics looking at data sources and sinks, inputs and outputs

        Returns:

          dict: A dictionary summarizing current I/O statistics
        """
425

Philip ABBET's avatar
Philip ABBET committed
426
        return stats.io_statistics(self.data, self.input_list, self.output_list)
427

Philip ABBET's avatar
Philip ABBET committed
428
    def __str__(self):
429
        return json.dumps(self.data, indent=4)
430

Philip ABBET's avatar
Philip ABBET committed
431 432
    def write(self, path):
        """Writes contents to precise filesystem location"""
433

434
        with open(path, "wt") as f:
Philip ABBET's avatar
Philip ABBET committed
435
            f.write(str(self))
436

Philip ABBET's avatar
Philip ABBET committed
437 438
    def dump_runner_configuration(self, directory):
        """Exports contents useful for a backend runner to run the algorithm"""
439

440
        data = convert_experiment_configuration_to_container(self.data)
441

442
        with open(os.path.join(directory, "configuration.json"), "wb") as f:
443
            json_data = json.dumps(data, indent=2)
444
            f.write(json_data.encode("utf-8"))
445

446
        tmp_prefix = os.path.join(directory, "prefix")
447 448
        if not os.path.exists(tmp_prefix):
            os.makedirs(tmp_prefix)
449

Philip ABBET's avatar
Philip ABBET committed
450
        self.algorithm.export(tmp_prefix)
451

452 453 454
        if self.loop_algorithm:
            self.loop_algorithm.export(tmp_prefix)

Philip ABBET's avatar
Philip ABBET committed
455 456
    def dump_databases_provider_configuration(self, directory):
        """Exports contents useful for a backend runner to run the algorithm"""
457

458
        with open(os.path.join(directory, "configuration.json"), "wb") as f:
459
            json_data = json.dumps(self.data, indent=2)
460
            f.write(json_data.encode("utf-8"))
461

462 463 464
        tmp_prefix = os.path.join(directory, "prefix")
        if not os.path.exists(tmp_prefix):
            os.makedirs(tmp_prefix)
465

Philip ABBET's avatar
Philip ABBET committed
466 467
        for db in self.databases.values():
            db.export(tmp_prefix)