base.py 16.3 KB
Newer Older
1
2
3
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
#!/usr/bin/env python
# vim: set fileencoding=utf-8 :

###############################################################################
#                                                                             #
# Copyright (c) 2017 Idiap Research Institute, http://www.idiap.ch/           #
# Contact: beat.support@idiap.ch                                              #
#                                                                             #
# This file is part of the beat.core module of the BEAT platform.             #
#                                                                             #
# Commercial License Usage                                                    #
# Licensees holding valid commercial BEAT licenses may use this file in       #
# accordance with the terms contained in a written agreement between you      #
# and Idiap. For further information contact tto@idiap.ch                     #
#                                                                             #
# Alternatively, this file may be used under the terms of the GNU Affero      #
# Public License version 3 as published by the Free Software and appearing    #
# in the file LICENSE.AGPL included in the packaging of this file.            #
# The BEAT platform is distributed in the hope that it will be useful, but    #
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY  #
# or FITNESS FOR A PARTICULAR PURPOSE.                                        #
#                                                                             #
# You should have received a copy of the GNU Affero Public License along      #
# with the BEAT platform. If not, see http://www.gnu.org/licenses/.           #
#                                                                             #
###############################################################################


29
30
31
32
33
34
35
36
37
"""
====
base
====

Execution utilities
"""


38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

import os
import sys
import glob
import collections

import logging
logger = logging.getLogger(__name__)

import simplejson

from .. import schema
from .. import database
from .. import algorithm
from .. import stats

from beat.backend.python.helpers import convert_experiment_configuration_to_container


class BaseExecutor(object):
Philip ABBET's avatar
Philip ABBET committed
58
    """Executors runs the code given an execution block information
59
60


Philip ABBET's avatar
Philip ABBET committed
61
    Parameters:
62

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

Philip ABBET's avatar
Philip ABBET committed
65
66
67
68
      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.
69

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

André Anjos's avatar
André Anjos committed
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
      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.
101
102


Philip ABBET's avatar
Philip ABBET committed
103
    Attributes:
104

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

Philip ABBET's avatar
Philip ABBET committed
107
108
      errors (list): A list containing errors found while loading this execution
        block.
109

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

Philip ABBET's avatar
Philip ABBET committed
113
114
      algorithm (beat.core.algorithm.Algorithm): An object representing the
        algorithm to be run.
115

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

Philip ABBET's avatar
Philip ABBET committed
121
122
123
124
      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.
125

Philip ABBET's avatar
Philip ABBET committed
126
127
      input_list (beat.core.inputs.InputList): A list of inputs that will be
        served to the algorithm.
128

Philip ABBET's avatar
Philip ABBET committed
129
130
      output_list (beat.core.outputs.OutputList): A list of outputs that the
        algorithm will produce.
131

Philip ABBET's avatar
Philip ABBET committed
132
133
      data_sources (list): A list with all data-sources created by our execution
        loader.
134

Philip ABBET's avatar
Philip ABBET committed
135
136
      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.
137

Philip ABBET's avatar
Philip ABBET committed
138
    """
139

Philip ABBET's avatar
Philip ABBET committed
140
141
142
143
144
145
146
147
    def __init__(self, prefix, data, cache=None, dataformat_cache=None,
            database_cache=None, algorithm_cache=None, library_cache=None,
            custom_root_folders=None):

        # Initialisations
        self.prefix = prefix
        self.cache = cache or os.path.join(self.prefix, 'cache')
        self.algorithm = None
148
        self.loop_algorithm = None
Philip ABBET's avatar
Philip ABBET committed
149
150
        self.databases = {}
        self.input_list = None
151
        self.data_loaders = None
Philip ABBET's avatar
Philip ABBET committed
152
153
154
155
        self.output_list = None
        self.data_sinks = []
        self.errors = []
        self.data = data
156
        self.debug = False
157

Philip ABBET's avatar
Philip ABBET committed
158
159
160
        # Check that the cache path exists
        if not os.path.exists(self.cache):
            raise IOError("Cache path `%s' does not exist" % self.cache)
161

Philip ABBET's avatar
Philip ABBET committed
162
163
164
165
166
167
        # 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 = {}
168

Philip ABBET's avatar
Philip ABBET committed
169
170
171
172
173
        # 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 {}
174

Philip ABBET's avatar
Philip ABBET committed
175
176
177
178
179
        # Basic validation of the data declaration, including JSON loading if required
        if not isinstance(data, dict):
            if not os.path.exists(data):
                self.errors.append('File not found: %s' % data)
                return
180

Philip ABBET's avatar
Philip ABBET committed
181
182
183
        self.data, self.errors = schema.validate('execution', data)
        if self.errors:
            return
184

Philip ABBET's avatar
Philip ABBET committed
185
186
187
188
        # Load the algorithm (using the algorithm cache if possible)
        if self.data['algorithm'] in algorithm_cache:
            self.algorithm = algorithm_cache[self.data['algorithm']]
        else:
189
190
191
192
            self.algorithm = algorithm.Algorithm(self.prefix,
                                                 self.data['algorithm'],
                                                 dataformat_cache,
                                                 library_cache)
Philip ABBET's avatar
Philip ABBET committed
193
            algorithm_cache[self.algorithm.name] = self.algorithm
194

Philip ABBET's avatar
Philip ABBET committed
195
196
197
        if not self.algorithm.valid:
            self.errors += self.algorithm.errors
            return
198

199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
        if 'loop' in self.data:
            loop = self.data['loop']
            if loop['algorithm'] in algorithm_cache:
                self.loop_algorithm = algorithm_cache[loop['algorithm']]
            else:
                self.loop_algorithm = algorithm.Algorithm(self.prefix,
                                                          loop['algorithm'],
                                                          dataformat_cache,
                                                          library_cache)
                algorithm_cache[self.loop_algorithm.name] = self.loop_algorithm

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

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

Philip ABBET's avatar
Philip ABBET committed
217
218
219
        # Check that the mapping in coherent
        if len(self.data['inputs']) != len(self.algorithm.input_map):
            self.errors.append("The number of inputs of the algorithm doesn't correspond")
220

221
        if 'outputs' in self.data and (len(self.data['outputs']) != len(self.algorithm.output_map)):
Philip ABBET's avatar
Philip ABBET committed
222
            self.errors.append("The number of outputs of the algorithm doesn't correspond")
223

Philip ABBET's avatar
Philip ABBET committed
224
225
226
        for name in self.data['inputs'].keys():
            if name not in self.algorithm.input_map.keys():
                self.errors.append("The input '%s' doesn't exist in the algorithm" % name)
227

228
        if 'outputs' in self.data:
Philip ABBET's avatar
Philip ABBET committed
229
230
231
            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 algorithm" % name)
232

233
234
235
236
237
        if 'loop' in self.data:
            for name in ['request', 'answer']:
                if name not in self.algorithm.loop_map.keys():
                    self.errors.append("The loop '%s' doesn't exist in the algorithm" % name)

Philip ABBET's avatar
Philip ABBET committed
238
239
        if self.errors:
            return
240

Philip ABBET's avatar
Philip ABBET committed
241
        # Load the databases (if any is required)
242
        self._update_db_cache(self.data['inputs'], database_cache)
243

244
245
        if 'loop' in self.data:
            self._update_db_cache(self.data['loop']['inputs'], database_cache)
246
247


Philip ABBET's avatar
Philip ABBET committed
248
249
    def __enter__(self):
        """Prepares inputs and outputs for the processing task
250

Philip ABBET's avatar
Philip ABBET committed
251
        Raises:
252

Philip ABBET's avatar
Philip ABBET committed
253
          IOError: in case something cannot be properly setup
254

Philip ABBET's avatar
Philip ABBET committed
255
        """
256

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

Philip ABBET's avatar
Philip ABBET committed
259
        return self
260
261


Philip ABBET's avatar
Philip ABBET committed
262
263
264
    def __exit__(self, exc_type, exc_value, traceback):
        """Closes all sinks and disconnects inputs and outputs
        """
265

Philip ABBET's avatar
Philip ABBET committed
266
267
268
269
270
        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()
271

Philip ABBET's avatar
Philip ABBET committed
272
        self.input_list = None
273
        self.data_loaders = []
Philip ABBET's avatar
Philip ABBET committed
274
275
        self.output_list = None
        self.data_sinks = []
276
277


278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
    def _update_db_cache(self, inputs, database_cache):
        """ Update the database cache based on the input list given"""

        for name, details in inputs.items():
            if 'database' in details:

                if details['database'] not in self.databases:

                    if details['database'] in database_cache:
                        db = database_cache[details['database']]
                    else:
                        db = database.Database(self.prefix, details['database'],
                                               dataformat_cache)

                        name = "database/%s" % db.name
                        if name in custom_root_folders:
                            db.data['root_folder'] = custom_root_folders[name]

                        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
304
305
    def _prepare_inputs(self):
        """Prepares all input required by the execution."""
306

Philip ABBET's avatar
Philip ABBET committed
307
        raise NotImplementedError()
308
309


Philip ABBET's avatar
Philip ABBET committed
310
311
    def _prepare_outputs(self):
        """Prepares all output required by the execution."""
312

Philip ABBET's avatar
Philip ABBET committed
313
        raise NotImplementedError()
314
315


Philip ABBET's avatar
Philip ABBET committed
316
317
318
    def process(self, virtual_memory_in_megabytes=0, max_cpu_percent=0,
                timeout_in_minutes=0):
        """Executes the user algorithm code
319

Philip ABBET's avatar
Philip ABBET committed
320
        Parameters:
321

André Anjos's avatar
André Anjos committed
322
323
324
          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.
325

André Anjos's avatar
André Anjos committed
326
327
328
329
330
          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.
331

Philip ABBET's avatar
Philip ABBET committed
332
333
          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
334
335
            killed with ``signal.SIGKILL``. If set to zero, no timeout will be
            applied.
336

Philip ABBET's avatar
Philip ABBET committed
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
        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"""
        return 'result' in self.data


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

        if self.analysis:
            path = os.path.join(self.cache, self.data['result']['path']) + '*'
            if not glob.glob(path): return False

        else:
            for name, details in self.data['outputs'].items():
                path = os.path.join(self.cache, details['path']) + '*'
                if not glob.glob(path): return False

        # 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
        """
385

Philip ABBET's avatar
Philip ABBET committed
386
        return stats.io_statistics(self.data, self.input_list, self.output_list)
387
388


Philip ABBET's avatar
Philip ABBET committed
389
390
    def __str__(self):
        return simplejson.dumps(self.data, indent=4)
391
392


Philip ABBET's avatar
Philip ABBET committed
393
394
    def write(self, path):
        """Writes contents to precise filesystem location"""
395

Philip ABBET's avatar
Philip ABBET committed
396
397
        with open(path, 'wt') as f:
            f.write(str(self))
398
399


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

403
        data = convert_experiment_configuration_to_container(self.data)
404

Philip ABBET's avatar
Philip ABBET committed
405
        with open(os.path.join(directory, 'configuration.json'), 'wb') as f:
406
407
            json_data = simplejson.dumps(data, indent=2)
            f.write(json_data.encode('utf-8'))
408

Philip ABBET's avatar
Philip ABBET committed
409
        tmp_prefix = os.path.join(directory, 'prefix')
410
411
        if not os.path.exists(tmp_prefix):
            os.makedirs(tmp_prefix)
412

Philip ABBET's avatar
Philip ABBET committed
413
        self.algorithm.export(tmp_prefix)
414

415
416
417
        if self.loop_algorithm:
            self.loop_algorithm.export(tmp_prefix)

418

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

Philip ABBET's avatar
Philip ABBET committed
422
        with open(os.path.join(directory, 'configuration.json'), 'wb') as f:
423
424
            json_data = simplejson.dumps(self.data, indent=2)
            f.write(json_data.encode('utf-8'))
425

Philip ABBET's avatar
Philip ABBET committed
426
427
        tmp_prefix = os.path.join(directory, 'prefix')
        if not os.path.exists(tmp_prefix): os.makedirs(tmp_prefix)
428

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