Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/joblib/memory.py: 24%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""
2A context object for caching a function's return value each time it
3is called with the same input arguments.
5"""
7# Author: Gael Varoquaux <gael dot varoquaux at normalesup dot org>
8# Copyright (c) 2009 Gael Varoquaux
9# License: BSD Style, 3 clauses.
11import asyncio
12import datetime
13import functools
14import inspect
15import logging
16import os
17import pathlib
18import pydoc
19import re
20import textwrap
21import time
22import tokenize
23import traceback
24import warnings
25import weakref
27from . import hashing
28from ._store_backends import (
29 CacheWarning, # noqa
30 FileSystemStoreBackend,
31 StoreBackendBase,
32)
33from .func_inspect import (
34 filter_args,
35 format_call,
36 format_signature,
37 get_func_code,
38 get_func_name,
39)
40from .logger import Logger, format_time, pformat
42FIRST_LINE_TEXT = "# first line:"
44# TODO: The following object should have a data store object as a sub
45# object, and the interface to persist and query should be separated in
46# the data store.
47#
48# This would enable creating 'Memory' objects with a different logic for
49# pickling that would simply span a MemorizedFunc with the same
50# store (or do we want to copy it to avoid cross-talks?), for instance to
51# implement HDF5 pickling.
53# TODO: Same remark for the logger, and probably use the Python logging
54# mechanism.
57def extract_first_line(func_code):
58 """Extract the first line information from the function code
59 text if available.
60 """
61 if func_code.startswith(FIRST_LINE_TEXT):
62 func_code = func_code.split("\n")
63 first_line = int(func_code[0][len(FIRST_LINE_TEXT) :])
64 func_code = "\n".join(func_code[1:])
65 else:
66 first_line = -1
67 return func_code, first_line
70class JobLibCollisionWarning(UserWarning):
71 """Warn that there might be a collision between names of functions."""
74_STORE_BACKENDS = {"local": FileSystemStoreBackend}
77def register_store_backend(backend_name, backend):
78 """Extend available store backends.
80 The Memory, MemorizeResult and MemorizeFunc objects are designed to be
81 agnostic to the type of store used behind. By default, the local file
82 system is used but this function gives the possibility to extend joblib's
83 memory pattern with other types of storage such as cloud storage (S3, GCS,
84 OpenStack, HadoopFS, etc) or blob DBs.
86 Parameters
87 ----------
88 backend_name: str
89 The name identifying the store backend being registered. For example,
90 'local' is used with FileSystemStoreBackend.
91 backend: StoreBackendBase subclass
92 The name of a class that implements the StoreBackendBase interface.
94 """
95 if not isinstance(backend_name, str):
96 raise ValueError(
97 "Store backend name should be a string, '{0}' given.".format(backend_name)
98 )
99 if backend is None or not issubclass(backend, StoreBackendBase):
100 raise ValueError(
101 "Store backend should inherit StoreBackendBase, '{0}' given.".format(
102 backend
103 )
104 )
106 _STORE_BACKENDS[backend_name] = backend
109def _store_backend_factory(backend, location, verbose=0, backend_options=None):
110 """Return the correct store object for the given location."""
111 if backend_options is None:
112 backend_options = {}
114 if isinstance(location, pathlib.Path):
115 location = str(location)
117 if isinstance(location, StoreBackendBase):
118 return location
119 elif isinstance(location, str):
120 obj = None
121 location = os.path.expanduser(location)
122 # The location is not a local file system, we look in the
123 # registered backends if there's one matching the given backend
124 # name.
125 for backend_key, backend_obj in _STORE_BACKENDS.items():
126 if backend == backend_key:
127 obj = backend_obj()
129 # By default, we assume the FileSystemStoreBackend can be used if no
130 # matching backend could be found.
131 if obj is None:
132 raise TypeError(
133 "Unknown location {0} or backend {1}".format(location, backend)
134 )
136 # The store backend is configured with the extra named parameters,
137 # some of them are specific to the underlying store backend.
138 obj.configure(location, verbose=verbose, backend_options=backend_options)
139 return obj
140 elif location is not None:
141 warnings.warn(
142 "Instantiating a backend using a {} as a location is not "
143 "supported by joblib. Returning None instead.".format(
144 location.__class__.__name__
145 ),
146 UserWarning,
147 )
149 return None
152def _build_func_identifier(func):
153 """Build a roughly unique identifier for the cached function."""
154 modules, funcname = get_func_name(func)
155 # We reuse historical fs-like way of building a function identifier
156 return os.path.join(*modules, funcname)
159# An in-memory store to avoid looking at the disk-based function
160# source code to check if a function definition has changed
161_FUNCTION_HASHES = weakref.WeakKeyDictionary()
164###############################################################################
165# class `MemorizedResult`
166###############################################################################
167class MemorizedResult(Logger):
168 """Object representing a cached value.
170 Attributes
171 ----------
172 location: str
173 The location of joblib cache. Depends on the store backend used.
175 func: function or str
176 function whose output is cached. The string case is intended only for
177 instantiation based on the output of repr() on another instance.
178 (namely eval(repr(memorized_instance)) works).
180 argument_hash: str
181 hash of the function arguments.
183 backend: str
184 Type of store backend for reading/writing cache files.
185 Default is 'local'.
187 mmap_mode: {None, 'r+', 'r', 'w+', 'c'}
188 The memmapping mode used when loading from cache numpy arrays. See
189 numpy.load for the meaning of the different values.
191 verbose: int
192 verbosity level (0 means no message).
194 timestamp, metadata: string
195 for internal use only.
196 """
198 def __init__(
199 self,
200 location,
201 call_id,
202 backend="local",
203 mmap_mode=None,
204 verbose=0,
205 timestamp=None,
206 metadata=None,
207 ):
208 Logger.__init__(self)
209 self._call_id = call_id
210 self.store_backend = _store_backend_factory(
211 backend,
212 location,
213 verbose=verbose,
214 backend_options=dict(mmap_mode=mmap_mode),
215 )
216 self.mmap_mode = mmap_mode
218 if metadata is not None:
219 self.metadata = metadata
220 else:
221 self.metadata = self.store_backend.get_metadata(self._call_id)
223 self.duration = self.metadata.get("duration", None)
224 self.verbose = verbose
225 self.timestamp = timestamp
227 @property
228 def func(self):
229 return self.func_id
231 @property
232 def func_id(self):
233 return self._call_id[0]
235 @property
236 def args_id(self):
237 return self._call_id[1]
239 def get(self):
240 """Read value from cache and return it."""
241 try:
242 return self.store_backend.load_item(
243 self._call_id,
244 timestamp=self.timestamp,
245 metadata=self.metadata,
246 verbose=self.verbose,
247 )
248 except ValueError as exc:
249 new_exc = KeyError(
250 "Error while trying to load a MemorizedResult's value. "
251 "It seems that this folder is corrupted : {}".format(
252 os.path.join(self.store_backend.location, *self._call_id)
253 )
254 )
255 raise new_exc from exc
257 def clear(self):
258 """Clear value from cache"""
259 self.store_backend.clear_item(self._call_id)
261 def __repr__(self):
262 return '{}(location="{}", func="{}", args_id="{}")'.format(
263 self.__class__.__name__, self.store_backend.location, *self._call_id
264 )
266 def __getstate__(self):
267 state = self.__dict__.copy()
268 state["timestamp"] = None
269 return state
272class NotMemorizedResult(object):
273 """Class representing an arbitrary value.
275 This class is a replacement for MemorizedResult when there is no cache.
276 """
278 __slots__ = ("value", "valid")
280 def __init__(self, value):
281 self.value = value
282 self.valid = True
284 def get(self):
285 if self.valid:
286 return self.value
287 else:
288 raise KeyError("No value stored.")
290 def clear(self):
291 self.valid = False
292 self.value = None
294 def __repr__(self):
295 if self.valid:
296 return "{class_name}({value})".format(
297 class_name=self.__class__.__name__, value=pformat(self.value)
298 )
299 else:
300 return self.__class__.__name__ + " with no value"
302 # __getstate__ and __setstate__ are required because of __slots__
303 def __getstate__(self):
304 return {"valid": self.valid, "value": self.value}
306 def __setstate__(self, state):
307 self.valid = state["valid"]
308 self.value = state["value"]
311###############################################################################
312# class `NotMemorizedFunc`
313###############################################################################
314class NotMemorizedFunc(object):
315 """No-op object decorating a function.
317 This class replaces MemorizedFunc when there is no cache. It provides an
318 identical API but does not write anything on disk.
320 Attributes
321 ----------
322 func: callable
323 Original undecorated function.
324 """
326 # Should be a light as possible (for speed)
327 def __init__(self, func):
328 self.func = func
330 def __call__(self, *args, **kwargs):
331 return self.func(*args, **kwargs)
333 def call_and_shelve(self, *args, **kwargs):
334 return NotMemorizedResult(self.func(*args, **kwargs))
336 def __repr__(self):
337 return "{0}(func={1})".format(self.__class__.__name__, self.func)
339 def clear(self, warn=True):
340 # Argument "warn" is for compatibility with MemorizedFunc.clear
341 pass
343 def call(self, *args, **kwargs):
344 return self.func(*args, **kwargs), {}
346 def check_call_in_cache(self, *args, **kwargs):
347 return False
350###############################################################################
351# class `AsyncNotMemorizedFunc`
352###############################################################################
353class AsyncNotMemorizedFunc(NotMemorizedFunc):
354 async def call_and_shelve(self, *args, **kwargs):
355 return NotMemorizedResult(await self.func(*args, **kwargs))
358###############################################################################
359# class `MemorizedFunc`
360###############################################################################
361class MemorizedFunc(Logger):
362 """Callable object decorating a function for caching its return value
363 each time it is called.
365 Methods are provided to inspect the cache or clean it.
367 Attributes
368 ----------
369 func: callable
370 The original, undecorated, function.
372 location: string
373 The location of joblib cache. Depends on the store backend used.
375 backend: str
376 Type of store backend for reading/writing cache files.
377 Default is 'local', in which case the location is the path to a
378 disk storage.
380 ignore: list or None
381 List of variable names to ignore when choosing whether to
382 recompute.
384 mmap_mode: {None, 'r+', 'r', 'w+', 'c'}
385 The memmapping mode used when loading from cache
386 numpy arrays. See numpy.load for the meaning of the different
387 values.
389 compress: boolean, integer or 2-tuple, optional
390 Whether to zip the stored data on disk. If an integer is
391 given, it should be between 1 and 9, and sets the amount
392 of compression. Note that compressed arrays cannot be
393 read by memmapping. If a 2-tuple is given, the first
394 element must be a string corresponding to a compression
395 method (e.g 'zlib', 'gzip', 'bz2', 'lzma' or 'xz') and
396 the second element must be an integer from 0 to 9,
397 corresponding to the compression level.
399 verbose: int, optional
400 The verbosity flag, controls messages that are issued as
401 the function is evaluated.
403 cache_validation_callback: callable, optional
404 Callable to check if a result in cache is valid or is to be recomputed.
405 When the function is called with arguments for which a cache exists,
406 the callback is called with the cache entry's metadata as its sole
407 argument. If it returns True, the cached result is returned, else the
408 cache for these arguments is cleared and the result is recomputed.
409 """
411 # ------------------------------------------------------------------------
412 # Public interface
413 # ------------------------------------------------------------------------
415 def __init__(
416 self,
417 func,
418 location,
419 backend="local",
420 ignore=None,
421 mmap_mode=None,
422 compress=False,
423 verbose=1,
424 timestamp=None,
425 cache_validation_callback=None,
426 ):
427 Logger.__init__(self)
428 self.mmap_mode = mmap_mode
429 self.compress = compress
430 self.func = func
431 self.cache_validation_callback = cache_validation_callback
432 self.func_id = _build_func_identifier(func)
433 self.ignore = ignore if ignore is not None else []
434 self._verbose = verbose
436 # retrieve store object from backend type and location.
437 self.store_backend = _store_backend_factory(
438 backend,
439 location,
440 verbose=verbose,
441 backend_options=dict(compress=compress, mmap_mode=mmap_mode),
442 )
443 if self.store_backend is not None:
444 # Create func directory on demand.
445 self.store_backend.store_cached_func_code([self.func_id])
447 self.timestamp = timestamp if timestamp is not None else time.time()
448 try:
449 functools.update_wrapper(self, func)
450 except Exception:
451 pass # Objects like ufunc don't like that
452 if inspect.isfunction(func):
453 doc = pydoc.TextDoc().document(func)
454 # Remove blank line
455 doc = doc.replace("\n", "\n\n", 1)
456 # Strip backspace-overprints for compatibility with autodoc
457 doc = re.sub("\x08.", "", doc)
458 else:
459 # Pydoc does a poor job on other objects
460 doc = func.__doc__
461 self.__doc__ = "Memoized version of %s" % doc
463 self._func_code_info = None
464 self._func_code_id = None
466 def _is_in_cache_and_valid(self, call_id):
467 """Check if the function call is cached and valid for given arguments.
469 - Compare the function code with the one from the cached function,
470 asserting if it has changed.
471 - Check if the function call is present in the cache.
472 - Call `cache_validation_callback` for user define cache validation.
474 Returns True if the function call is in cache and can be used, and
475 returns False otherwise.
476 """
477 # Check if the code of the function has changed
478 if not self._check_previous_func_code(stacklevel=4):
479 return False
481 # Check if this specific call is in the cache
482 if not self.store_backend.contains_item(call_id):
483 return False
485 # Call the user defined cache validation callback
486 metadata = self.store_backend.get_metadata(call_id)
487 if (
488 self.cache_validation_callback is not None
489 and not self.cache_validation_callback(metadata)
490 ):
491 self.store_backend.clear_item(call_id)
492 return False
494 return True
496 def _cached_call(self, args, kwargs, shelving):
497 """Call wrapped function and cache result, or read cache if available.
499 This function returns the wrapped function output or a reference to
500 the cached result.
502 Arguments:
503 ----------
505 args, kwargs: list and dict
506 input arguments for wrapped function
508 shelving: bool
509 True when called via the call_and_shelve function.
512 Returns
513 -------
514 output: Output of the wrapped function if shelving is false, or a
515 MemorizedResult reference to the value if shelving is true.
516 metadata: dict containing the metadata associated with the call.
517 """
518 args_id = self._get_args_id(*args, **kwargs)
519 call_id = (self.func_id, args_id)
520 _, func_name = get_func_name(self.func)
521 func_info = self.store_backend.get_cached_func_info([self.func_id])
522 location = func_info["location"]
524 if self._verbose >= 20:
525 logging.basicConfig(level=logging.INFO)
526 _, signature = format_signature(self.func, *args, **kwargs)
527 self.info(
528 textwrap.dedent(
529 f"""
530 Querying {func_name} with signature
531 {signature}.
533 (argument hash {args_id})
535 The store location is {location}.
536 """
537 )
538 )
540 # Compare the function code with the previous to see if the
541 # function code has changed and check if the results are present in
542 # the cache.
543 if self._is_in_cache_and_valid(call_id):
544 if shelving:
545 return self._get_memorized_result(call_id), {}
547 try:
548 start_time = time.time()
549 output = self._load_item(call_id)
550 if self._verbose > 4:
551 self._print_duration(
552 time.time() - start_time, context="cache loaded "
553 )
554 return output, {}
555 except Exception:
556 # XXX: Should use an exception logger
557 _, signature = format_signature(self.func, *args, **kwargs)
558 self.warn(
559 "Exception while loading results for {}\n {}".format(
560 signature, traceback.format_exc()
561 )
562 )
564 if self._verbose > 10:
565 self.warn(
566 f"Computing func {func_name}, argument hash {args_id} "
567 f"in location {location}"
568 )
570 # Returns the output but not the metadata
571 return self._call(call_id, args, kwargs, shelving)
573 @property
574 def func_code_info(self):
575 # 3-tuple property containing: the function source code, source file,
576 # and first line of the code inside the source file
577 if hasattr(self.func, "__code__"):
578 if self._func_code_id is None:
579 self._func_code_id = id(self.func.__code__)
580 elif id(self.func.__code__) != self._func_code_id:
581 # Be robust to dynamic reassignments of self.func.__code__
582 self._func_code_info = None
584 if self._func_code_info is None:
585 # Cache the source code of self.func . Provided that get_func_code
586 # (which should be called once on self) gets called in the process
587 # in which self.func was defined, this caching mechanism prevents
588 # undesired cache clearing when the cached function is called in
589 # an environment where the introspection utilities get_func_code
590 # relies on do not work (typically, in joblib child processes).
591 # See #1035 for more info
592 # TODO (pierreglaser): do the same with get_func_name?
593 self._func_code_info = get_func_code(self.func)
594 return self._func_code_info
596 def call_and_shelve(self, *args, **kwargs):
597 """Call wrapped function, cache result and return a reference.
599 This method returns a reference to the cached result instead of the
600 result itself. The reference object is small and picklable, allowing
601 to send or store it easily. Call .get() on reference object to get
602 result.
604 Returns
605 -------
606 cached_result: MemorizedResult or NotMemorizedResult
607 reference to the value returned by the wrapped function. The
608 class "NotMemorizedResult" is used when there is no cache
609 activated (e.g. location=None in Memory).
610 """
611 # Return the wrapped output, without the metadata
612 return self._cached_call(args, kwargs, shelving=True)[0]
614 def __call__(self, *args, **kwargs):
615 # Return the output, without the metadata
616 return self._cached_call(args, kwargs, shelving=False)[0]
618 def __getstate__(self):
619 # Make sure self.func's source is introspected prior to being pickled -
620 # code introspection utilities typically do not work inside child
621 # processes
622 _ = self.func_code_info
624 # We don't store the timestamp when pickling, to avoid the hash
625 # depending from it.
626 state = self.__dict__.copy()
627 state["timestamp"] = None
629 # Invalidate the code id as id(obj) will be different in the child
630 state["_func_code_id"] = None
632 return state
634 def check_call_in_cache(self, *args, **kwargs):
635 """Check if the function call is cached and valid for given arguments.
637 Does not call the function or do any work besides function inspection
638 and argument hashing.
640 - Compare the function code with the one from the cached function,
641 asserting if it has changed.
642 - Check if the function call is present in the cache.
643 - Call `cache_validation_callback` for user define cache validation.
645 Returns
646 -------
647 is_call_in_cache: bool
648 Whether or not the function call is in cache and can be used.
649 """
650 call_id = (self.func_id, self._get_args_id(*args, **kwargs))
651 return self._is_in_cache_and_valid(call_id)
653 # ------------------------------------------------------------------------
654 # Private interface
655 # ------------------------------------------------------------------------
657 def _get_args_id(self, *args, **kwargs):
658 """Return the input parameter hash of a result."""
659 return hashing.hash(
660 filter_args(self.func, self.ignore, args, kwargs),
661 coerce_mmap=self.mmap_mode is not None,
662 )
664 def _hash_func(self):
665 """Hash a function to key the online cache"""
666 func_code_h = hash(getattr(self.func, "__code__", None))
667 return id(self.func), hash(self.func), func_code_h
669 def _write_func_code(self, func_code, first_line):
670 """Write the function code and the filename to a file."""
671 # We store the first line because the filename and the function
672 # name is not always enough to identify a function: people
673 # sometimes have several functions named the same way in a
674 # file. This is bad practice, but joblib should be robust to bad
675 # practice.
676 func_code = "%s %i\n%s" % (FIRST_LINE_TEXT, first_line, func_code)
677 self.store_backend.store_cached_func_code([self.func_id], func_code)
679 # Also store in the in-memory store of function hashes
680 is_named_callable = (
681 hasattr(self.func, "__name__") and self.func.__name__ != "<lambda>"
682 )
683 if is_named_callable:
684 # Don't do this for lambda functions or strange callable
685 # objects, as it ends up being too fragile
686 func_hash = self._hash_func()
687 try:
688 _FUNCTION_HASHES[self.func] = func_hash
689 except TypeError:
690 # Some callable are not hashable
691 pass
693 def _check_previous_func_code(self, stacklevel=2):
694 """
695 stacklevel is the depth a which this function is called, to
696 issue useful warnings to the user.
697 """
698 # First check if our function is in the in-memory store.
699 # Using the in-memory store not only makes things faster, but it
700 # also renders us robust to variations of the files when the
701 # in-memory version of the code does not vary
702 try:
703 if self.func in _FUNCTION_HASHES:
704 # We use as an identifier the id of the function and its
705 # hash. This is more likely to falsely change than have hash
706 # collisions, thus we are on the safe side.
707 func_hash = self._hash_func()
708 if func_hash == _FUNCTION_HASHES[self.func]:
709 return True
710 except TypeError:
711 # Some callables are not hashable
712 pass
714 # Here, we go through some effort to be robust to dynamically
715 # changing code and collision. We cannot inspect.getsource
716 # because it is not reliable when using IPython's magic "%run".
717 func_code, source_file, first_line = self.func_code_info
718 try:
719 old_func_code, old_first_line = extract_first_line(
720 self.store_backend.get_cached_func_code([self.func_id])
721 )
722 except (IOError, OSError): # some backend can also raise OSError
723 self._write_func_code(func_code, first_line)
724 return False
725 if old_func_code == func_code:
726 return True
728 # We have differing code, is this because we are referring to
729 # different functions, or because the function we are referring to has
730 # changed?
732 _, func_name = get_func_name(
733 self.func, resolv_alias=False, win_characters=False
734 )
735 if old_first_line == first_line == -1 or func_name == "<lambda>":
736 if not first_line == -1:
737 func_description = "{0} ({1}:{2})".format(
738 func_name, source_file, first_line
739 )
740 else:
741 func_description = func_name
742 warnings.warn(
743 JobLibCollisionWarning(
744 "Cannot detect name collisions for function '{0}'".format(
745 func_description
746 )
747 ),
748 stacklevel=stacklevel,
749 )
751 # Fetch the code at the old location and compare it. If it is the
752 # same than the code store, we have a collision: the code in the
753 # file has not changed, but the name we have is pointing to a new
754 # code block.
755 if not old_first_line == first_line and source_file is not None:
756 if os.path.exists(source_file):
757 _, func_name = get_func_name(self.func, resolv_alias=False)
758 num_lines = len(func_code.split("\n"))
759 with tokenize.open(source_file) as f:
760 on_disk_func_code = f.readlines()[
761 old_first_line - 1 : old_first_line - 1 + num_lines - 1
762 ]
763 on_disk_func_code = "".join(on_disk_func_code)
764 possible_collision = (
765 on_disk_func_code.rstrip() == old_func_code.rstrip()
766 )
767 else:
768 possible_collision = source_file.startswith("<doctest ")
769 if possible_collision:
770 warnings.warn(
771 JobLibCollisionWarning(
772 "Possible name collisions between functions "
773 "'%s' (%s:%i) and '%s' (%s:%i)"
774 % (
775 func_name,
776 source_file,
777 old_first_line,
778 func_name,
779 source_file,
780 first_line,
781 )
782 ),
783 stacklevel=stacklevel,
784 )
786 # The function has changed, wipe the cache directory.
787 # XXX: Should be using warnings, and giving stacklevel
788 if self._verbose > 10:
789 _, func_name = get_func_name(self.func, resolv_alias=False)
790 self.warn(
791 "Function {0} (identified by {1}) has changed.".format(
792 func_name, self.func_id
793 )
794 )
795 self.clear(warn=True)
796 return False
798 def clear(self, warn=True):
799 """Empty the function's cache."""
800 func_id = self.func_id
801 if self._verbose > 0 and warn:
802 self.warn("Clearing function cache identified by %s" % func_id)
803 self.store_backend.clear_path(
804 [
805 func_id,
806 ]
807 )
809 func_code, _, first_line = self.func_code_info
810 self._write_func_code(func_code, first_line)
812 def call(self, *args, **kwargs):
813 """Force the execution of the function with the given arguments.
815 The output values will be persisted, i.e., the cache will be updated
816 with any new values.
818 Parameters
819 ----------
820 *args: arguments
821 The arguments.
822 **kwargs: keyword arguments
823 Keyword arguments.
825 Returns
826 -------
827 output : object
828 The output of the function call.
829 metadata : dict
830 The metadata associated with the call.
831 """
832 call_id = (self.func_id, self._get_args_id(*args, **kwargs))
834 # Return the output and the metadata
835 return self._call(call_id, args, kwargs)
837 def _call(self, call_id, args, kwargs, shelving=False):
838 # Return the output and the metadata
839 self._before_call(args, kwargs)
840 start_time = time.time()
841 output = self.func(*args, **kwargs)
842 return self._after_call(call_id, args, kwargs, shelving, output, start_time)
844 def _before_call(self, args, kwargs):
845 if self._verbose > 0:
846 print(format_call(self.func, args, kwargs))
848 def _after_call(self, call_id, args, kwargs, shelving, output, start_time):
849 self.store_backend.dump_item(call_id, output, verbose=self._verbose)
850 duration = time.time() - start_time
851 if self._verbose > 0:
852 self._print_duration(duration)
853 metadata = self._persist_input(duration, call_id, args, kwargs)
854 if shelving:
855 return self._get_memorized_result(call_id, metadata), metadata
857 if self.mmap_mode is not None:
858 # Memmap the output at the first call to be consistent with
859 # later calls
860 output = self._load_item(call_id, metadata)
861 return output, metadata
863 def _persist_input(self, duration, call_id, args, kwargs, this_duration_limit=0.5):
864 """Save a small summary of the call using json format in the
865 output directory.
867 output_dir: string
868 directory where to write metadata.
870 duration: float
871 time taken by hashing input arguments, calling the wrapped
872 function and persisting its output.
874 args, kwargs: list and dict
875 input arguments for wrapped function
877 this_duration_limit: float
878 Max execution time for this function before issuing a warning.
879 """
880 start_time = time.time()
881 argument_dict = filter_args(self.func, self.ignore, args, kwargs)
883 input_repr = dict((k, repr(v)) for k, v in argument_dict.items())
884 # This can fail due to race-conditions with multiple
885 # concurrent joblibs removing the file or the directory
886 metadata = {
887 "duration": duration,
888 "input_args": input_repr,
889 "time": start_time,
890 }
892 self.store_backend.store_metadata(call_id, metadata)
894 this_duration = time.time() - start_time
895 if this_duration > this_duration_limit:
896 # This persistence should be fast. It will not be if repr() takes
897 # time and its output is large, because json.dump will have to
898 # write a large file. This should not be an issue with numpy arrays
899 # for which repr() always output a short representation, but can
900 # be with complex dictionaries. Fixing the problem should be a
901 # matter of replacing repr() above by something smarter.
902 warnings.warn(
903 "Persisting input arguments took %.2fs to run."
904 "If this happens often in your code, it can cause "
905 "performance problems "
906 "(results will be correct in all cases). "
907 "The reason for this is probably some large input "
908 "arguments for a wrapped function." % this_duration,
909 stacklevel=5,
910 )
911 return metadata
913 def _get_memorized_result(self, call_id, metadata=None):
914 return MemorizedResult(
915 self.store_backend,
916 call_id,
917 metadata=metadata,
918 timestamp=self.timestamp,
919 verbose=self._verbose - 1,
920 )
922 def _load_item(self, call_id, metadata=None):
923 return self.store_backend.load_item(
924 call_id, metadata=metadata, timestamp=self.timestamp, verbose=self._verbose
925 )
927 def _print_duration(self, duration, context=""):
928 _, name = get_func_name(self.func)
929 msg = f"{name} {context}- {format_time(duration)}"
930 print(max(0, (80 - len(msg))) * "_" + msg)
932 # ------------------------------------------------------------------------
933 # Private `object` interface
934 # ------------------------------------------------------------------------
936 def __repr__(self):
937 return "{class_name}(func={func}, location={location})".format(
938 class_name=self.__class__.__name__,
939 func=self.func,
940 location=self.store_backend.location,
941 )
944###############################################################################
945# class `AsyncMemorizedFunc`
946###############################################################################
947class AsyncMemorizedFunc(MemorizedFunc):
948 async def __call__(self, *args, **kwargs):
949 out = self._cached_call(args, kwargs, shelving=False)
950 out = await out if asyncio.iscoroutine(out) else out
951 return out[0] # Don't return metadata
953 async def call_and_shelve(self, *args, **kwargs):
954 out = self._cached_call(args, kwargs, shelving=True)
955 out = await out if asyncio.iscoroutine(out) else out
956 return out[0] # Don't return metadata
958 async def call(self, *args, **kwargs):
959 out = super().call(*args, **kwargs)
960 return await out if asyncio.iscoroutine(out) else out
962 async def _call(self, call_id, args, kwargs, shelving=False):
963 self._before_call(args, kwargs)
964 start_time = time.time()
965 output = await self.func(*args, **kwargs)
966 return self._after_call(call_id, args, kwargs, shelving, output, start_time)
969###############################################################################
970# class `Memory`
971###############################################################################
972class Memory(Logger):
973 """A context object for caching a function's return value each time it
974 is called with the same input arguments.
976 All values are cached on the filesystem, in a deep directory
977 structure.
979 Read more in the :ref:`User Guide <memory>`.
981 Parameters
982 ----------
983 location: str, pathlib.Path or None
984 The path of the base directory to use as a data store
985 or None. If None is given, no caching is done and
986 the Memory object is completely transparent. This option
987 replaces cachedir since version 0.12.
989 backend: str, optional, default='local'
990 Type of store backend for reading/writing cache files.
991 The 'local' backend is using regular filesystem operations to
992 manipulate data (open, mv, etc) in the backend.
994 mmap_mode: {None, 'r+', 'r', 'w+', 'c'}, optional
995 The memmapping mode used when loading from cache
996 numpy arrays. See numpy.load for the meaning of the
997 arguments.
999 compress: boolean, integer or 2-tuple, optional
1000 Whether to zip the stored data on disk. If an integer is
1001 given, it should be between 1 and 9, and sets the amount
1002 of compression. Note that compressed arrays cannot be
1003 read by memmapping. If a 2-tuple is given, the first
1004 element must be a string corresponding to a compression
1005 method (e.g 'zlib', 'gzip', 'bz2', 'lzma' or 'xz') and
1006 the second element must be an integer from 0 to 9,
1007 corresponding to the compression level.
1009 verbose: int, optional
1010 Verbosity flag, controls the debug messages that are issued
1011 as functions are evaluated.
1013 backend_options: dict, optional
1014 Contains a dictionary of named parameters used to configure
1015 the store backend.
1016 """
1018 # ------------------------------------------------------------------------
1019 # Public interface
1020 # ------------------------------------------------------------------------
1022 def __init__(
1023 self,
1024 location=None,
1025 backend="local",
1026 mmap_mode=None,
1027 compress=False,
1028 verbose=1,
1029 backend_options=None,
1030 ):
1031 Logger.__init__(self)
1032 self._verbose = verbose
1033 self.mmap_mode = mmap_mode
1034 self.timestamp = time.time()
1035 self.backend = backend
1036 self.compress = compress
1037 if backend_options is None:
1038 backend_options = {}
1039 self.backend_options = backend_options
1041 if compress and mmap_mode is not None:
1042 warnings.warn("Compressed results cannot be memmapped", stacklevel=2)
1044 self.location = location
1045 if isinstance(location, str):
1046 location = os.path.join(location, "joblib")
1048 self.store_backend = _store_backend_factory(
1049 backend,
1050 location,
1051 verbose=self._verbose,
1052 backend_options=dict(
1053 compress=compress, mmap_mode=mmap_mode, **backend_options
1054 ),
1055 )
1057 def cache(
1058 self,
1059 func=None,
1060 ignore=None,
1061 verbose=None,
1062 mmap_mode=False,
1063 cache_validation_callback=None,
1064 ):
1065 """Decorates the given function func to only compute its return
1066 value for input arguments not cached on disk.
1068 Parameters
1069 ----------
1070 func: callable, optional
1071 The function to be decorated
1072 ignore: list of strings
1073 A list of arguments name to ignore in the hashing
1074 verbose: integer, optional
1075 The verbosity mode of the function. By default that
1076 of the memory object is used.
1077 mmap_mode: {None, 'r+', 'r', 'w+', 'c'}, optional
1078 The memmapping mode used when loading from cache
1079 numpy arrays. See numpy.load for the meaning of the
1080 arguments. By default that of the memory object is used.
1081 cache_validation_callback: callable, optional
1082 Callable to validate whether or not the cache is valid. When
1083 the cached function is called with arguments for which a cache
1084 exists, this callable is called with the metadata of the cached
1085 result as its sole argument. If it returns True, then the
1086 cached result is returned, else the cache for these arguments
1087 is cleared and recomputed.
1089 Returns
1090 -------
1091 decorated_func: MemorizedFunc object
1092 The returned object is a MemorizedFunc object, that is
1093 callable (behaves like a function), but offers extra
1094 methods for cache lookup and management. See the
1095 documentation for :class:`joblib.memory.MemorizedFunc`.
1096 """
1097 if cache_validation_callback is not None and not callable(
1098 cache_validation_callback
1099 ):
1100 raise ValueError(
1101 "cache_validation_callback needs to be callable. "
1102 f"Got {cache_validation_callback}."
1103 )
1104 if func is None:
1105 # Partial application, to be able to specify extra keyword
1106 # arguments in decorators
1107 return functools.partial(
1108 self.cache,
1109 ignore=ignore,
1110 mmap_mode=mmap_mode,
1111 verbose=verbose,
1112 cache_validation_callback=cache_validation_callback,
1113 )
1114 if self.store_backend is None:
1115 cls = (
1116 AsyncNotMemorizedFunc
1117 if inspect.iscoroutinefunction(func)
1118 else NotMemorizedFunc
1119 )
1120 return cls(func)
1121 if verbose is None:
1122 verbose = self._verbose
1123 if mmap_mode is False:
1124 mmap_mode = self.mmap_mode
1125 if isinstance(func, MemorizedFunc):
1126 func = func.func
1127 cls = AsyncMemorizedFunc if inspect.iscoroutinefunction(func) else MemorizedFunc
1128 return cls(
1129 func,
1130 location=self.store_backend,
1131 backend=self.backend,
1132 ignore=ignore,
1133 mmap_mode=mmap_mode,
1134 compress=self.compress,
1135 verbose=verbose,
1136 timestamp=self.timestamp,
1137 cache_validation_callback=cache_validation_callback,
1138 )
1140 def clear(self, warn=True):
1141 """Erase the complete cache directory."""
1142 if warn:
1143 self.warn("Flushing completely the cache")
1144 if self.store_backend is not None:
1145 self.store_backend.clear()
1147 # As the cache is completely clear, make sure the _FUNCTION_HASHES
1148 # cache is also reset. Else, for a function that is present in this
1149 # table, results cached after this clear will be have cache miss
1150 # as the function code is not re-written.
1151 _FUNCTION_HASHES.clear()
1153 def reduce_size(self, bytes_limit=None, items_limit=None, age_limit=None):
1154 """Remove cache elements to make the cache fit its limits.
1156 The limitation can impose that the cache size fits in ``bytes_limit``,
1157 that the number of cache items is no more than ``items_limit``, and
1158 that all files in cache are not older than ``age_limit``.
1160 Parameters
1161 ----------
1162 bytes_limit: int | str, optional
1163 Limit in bytes of the size of the cache. By default, the size of
1164 the cache is unlimited. When reducing the size of the cache,
1165 ``joblib`` keeps the most recently accessed items first. If a
1166 str is passed, it is converted to a number of bytes using units
1167 { K | M | G} for kilo, mega, giga.
1169 items_limit: int, optional
1170 Number of items to limit the cache to. By default, the number of
1171 items in the cache is unlimited. When reducing the size of the
1172 cache, ``joblib`` keeps the most recently accessed items first.
1174 age_limit: datetime.timedelta, optional
1175 Maximum age of items to limit the cache to. When reducing the size
1176 of the cache, any items last accessed more than the given length of
1177 time ago are deleted. Example: to remove files older than 5 days,
1178 use datetime.timedelta(days=5). Negative timedelta are not
1179 accepted.
1180 """
1181 if self.store_backend is None:
1182 # No cached results, this function does nothing.
1183 return
1185 if bytes_limit is None and items_limit is None and age_limit is None:
1186 # No limitation to impose, returning
1187 return
1189 # Defers the actual limits enforcing to the store backend.
1190 self.store_backend.enforce_store_limits(bytes_limit, items_limit, age_limit)
1192 def eval(self, func, *args, **kwargs):
1193 """Eval function func with arguments `*args` and `**kwargs`,
1194 in the context of the memory.
1196 This method works similarly to the builtin `apply`, except
1197 that the function is called only if the cache is not
1198 up to date.
1200 """
1201 if self.store_backend is None:
1202 return func(*args, **kwargs)
1203 return self.cache(func)(*args, **kwargs)
1205 # ------------------------------------------------------------------------
1206 # Private `object` interface
1207 # ------------------------------------------------------------------------
1209 def __repr__(self):
1210 return "{class_name}(location={location})".format(
1211 class_name=self.__class__.__name__,
1212 location=(
1213 None if self.store_backend is None else self.store_backend.location
1214 ),
1215 )
1217 def __getstate__(self):
1218 """We don't store the timestamp when pickling, to avoid the hash
1219 depending from it.
1220 """
1221 state = self.__dict__.copy()
1222 state["timestamp"] = None
1223 return state
1226###############################################################################
1227# cache_validation_callback helpers
1228###############################################################################
1231def expires_after(
1232 days=0, seconds=0, microseconds=0, milliseconds=0, minutes=0, hours=0, weeks=0
1233):
1234 """Helper cache_validation_callback to force recompute after a duration.
1236 Parameters
1237 ----------
1238 days, seconds, microseconds, milliseconds, minutes, hours, weeks: numbers
1239 argument passed to a timedelta.
1240 """
1241 delta = datetime.timedelta(
1242 days=days,
1243 seconds=seconds,
1244 microseconds=microseconds,
1245 milliseconds=milliseconds,
1246 minutes=minutes,
1247 hours=hours,
1248 weeks=weeks,
1249 )
1251 def cache_validation_callback(metadata):
1252 computation_age = time.time() - metadata["time"]
1253 return computation_age < delta.total_seconds()
1255 return cache_validation_callback