stale-free functionality + thread-safe

.gitattributes

@@ -0,0 +1 @@
+cachier/_version.py export-subst

MANIFEST.in

@@ -0,0 +1,2 @@
+include versioneer.py
+include cachier/_version.py

cachier/__init__.py

@@ -1 +1,4 @@
-from .core import *
+from .core import *
+from ._version import get_versions
+__version__ = get_versions()['version']
+del get_versions

cachier/_version.py

@@ -0,0 +1,484 @@
+
+# This file helps to compute a version number in source trees obtained from
+# git-archive tarball (such as those provided by githubs download-from-tag
+# feature). Distribution tarballs (built by setup.py sdist) and build
+# directories (produced by setup.py build) will contain a much shorter file
+# that just contains the computed version number.
+
+# This file is released into the public domain. Generated by
+# versioneer-0.16 (https://github.com/warner/python-versioneer)
+
+"""Git implementation of _version.py."""
+
+import errno
+import os
+import re
+import subprocess
+import sys
+
+
+def get_keywords():
+    """Get the keywords needed to look up the version information."""
+    # these strings will be replaced by git during git-archive.
+    # setup.py/versioneer.py will grep for the variable names, so they must
+    # each be defined on a line of their own. _version.py will just call
+    # get_keywords().
+    git_refnames = "$Format:%d$"
+    git_full = "$Format:%H$"
+    keywords = {"refnames": git_refnames, "full": git_full}
+    return keywords
+
+
+class VersioneerConfig:
+    """Container for Versioneer configuration parameters."""
+
+
+def get_config():
+    """Create, populate and return the VersioneerConfig() object."""
+    # these strings are filled in when 'setup.py versioneer' creates
+    # _version.py
+    cfg = VersioneerConfig()
+    cfg.VCS = "git"
+    cfg.style = "pep440-pre"
+    cfg.tag_prefix = "v"
+    cfg.parentdir_prefix = "cachier-"
+    cfg.versionfile_source = "cachier/_version.py"
+    cfg.verbose = False
+    return cfg
+
+
+class NotThisMethod(Exception):
+    """Exception raised if a method is not valid for the current scenario."""
+
+
+LONG_VERSION_PY = {}
+HANDLERS = {}
+
+
+def register_vcs_handler(vcs, method):  # decorator
+    """Decorator to mark a method as the handler for a particular VCS."""
+    def decorate(f):
+        """Store f in HANDLERS[vcs][method]."""
+        if vcs not in HANDLERS:
+            HANDLERS[vcs] = {}
+        HANDLERS[vcs][method] = f
+        return f
+    return decorate
+
+
+def run_command(commands, args, cwd=None, verbose=False, hide_stderr=False):
+    """Call the given command(s)."""
+    assert isinstance(commands, list)
+    p = None
+    for c in commands:
+        try:
+            dispcmd = str([c] + args)
+            # remember shell=False, so use git.cmd on windows, not just git
+            p = subprocess.Popen([c] + args, cwd=cwd, stdout=subprocess.PIPE,
+                                 stderr=(subprocess.PIPE if hide_stderr
+                                         else None))
+            break
+        except EnvironmentError:
+            e = sys.exc_info()[1]
+            if e.errno == errno.ENOENT:
+                continue
+            if verbose:
+                print("unable to run %s" % dispcmd)
+                print(e)
+            return None
+    else:
+        if verbose:
+            print("unable to find command, tried %s" % (commands,))
+        return None
+    stdout = p.communicate()[0].strip()
+    if sys.version_info[0] >= 3:
+        stdout = stdout.decode()
+    if p.returncode != 0:
+        if verbose:
+            print("unable to run %s (error)" % dispcmd)
+        return None
+    return stdout
+
+
+def versions_from_parentdir(parentdir_prefix, root, verbose):
+    """Try to determine the version from the parent directory name.
+
+    Source tarballs conventionally unpack into a directory that includes
+    both the project name and a version string.
+    """
+    dirname = os.path.basename(root)
+    if not dirname.startswith(parentdir_prefix):
+        if verbose:
+            print("guessing rootdir is '%s', but '%s' doesn't start with "
+                  "prefix '%s'" % (root, dirname, parentdir_prefix))
+        raise NotThisMethod("rootdir doesn't start with parentdir_prefix")
+    return {"version": dirname[len(parentdir_prefix):],
+            "full-revisionid": None,
+            "dirty": False, "error": None}
+
+
+@register_vcs_handler("git", "get_keywords")
+def git_get_keywords(versionfile_abs):
+    """Extract version information from the given file."""
+    # the code embedded in _version.py can just fetch the value of these
+    # keywords. When used from setup.py, we don't want to import _version.py,
+    # so we do it with a regexp instead. This function is not used from
+    # _version.py.
+    keywords = {}
+    try:
+        f = open(versionfile_abs, "r")
+        for line in f.readlines():
+            if line.strip().startswith("git_refnames ="):
+                mo = re.search(r'=\s*"(.*)"', line)
+                if mo:
+                    keywords["refnames"] = mo.group(1)
+            if line.strip().startswith("git_full ="):
+                mo = re.search(r'=\s*"(.*)"', line)
+                if mo:
+                    keywords["full"] = mo.group(1)
+        f.close()
+    except EnvironmentError:
+        pass
+    return keywords
+
+
+@register_vcs_handler("git", "keywords")
+def git_versions_from_keywords(keywords, tag_prefix, verbose):
+    """Get version information from git keywords."""
+    if not keywords:
+        raise NotThisMethod("no keywords at all, weird")
+    refnames = keywords["refnames"].strip()
+    if refnames.startswith("$Format"):
+        if verbose:
+            print("keywords are unexpanded, not using")
+        raise NotThisMethod("unexpanded keywords, not a git-archive tarball")
+    refs = set([r.strip() for r in refnames.strip("()").split(",")])
+    # starting in git-1.8.3, tags are listed as "tag: foo-1.0" instead of
+    # just "foo-1.0". If we see a "tag: " prefix, prefer those.
+    TAG = "tag: "
+    tags = set([r[len(TAG):] for r in refs if r.startswith(TAG)])
+    if not tags:
+        # Either we're using git < 1.8.3, or there really are no tags. We use
+        # a heuristic: assume all version tags have a digit. The old git %d
+        # expansion behaves like git log --decorate=short and strips out the
+        # refs/heads/ and refs/tags/ prefixes that would let us distinguish
+        # between branches and tags. By ignoring refnames without digits, we
+        # filter out many common branch names like "release" and
+        # "stabilization", as well as "HEAD" and "master".
+        tags = set([r for r in refs if re.search(r'\d', r)])
+        if verbose:
+            print("discarding '%s', no digits" % ",".join(refs-tags))
+    if verbose:
+        print("likely tags: %s" % ",".join(sorted(tags)))
+    for ref in sorted(tags):
+        # sorting will prefer e.g. "2.0" over "2.0rc1"
+        if ref.startswith(tag_prefix):
+            r = ref[len(tag_prefix):]
+            if verbose:
+                print("picking %s" % r)
+            return {"version": r,
+                    "full-revisionid": keywords["full"].strip(),
+                    "dirty": False, "error": None
+                    }
+    # no suitable tags, so version is "0+unknown", but full hex is still there
+    if verbose:
+        print("no suitable tags, using unknown + full revision id")
+    return {"version": "0+unknown",
+            "full-revisionid": keywords["full"].strip(),
+            "dirty": False, "error": "no suitable tags"}
+
+
+@register_vcs_handler("git", "pieces_from_vcs")
+def git_pieces_from_vcs(tag_prefix, root, verbose, run_command=run_command):
+    """Get version from 'git describe' in the root of the source tree.
+
+    This only gets called if the git-archive 'subst' keywords were *not*
+    expanded, and _version.py hasn't already been rewritten with a short
+    version string, meaning we're inside a checked out source tree.
+    """
+    if not os.path.exists(os.path.join(root, ".git")):
+        if verbose:
+            print("no .git in %s" % root)
+        raise NotThisMethod("no .git directory")
+
+    GITS = ["git"]
+    if sys.platform == "win32":
+        GITS = ["git.cmd", "git.exe"]
+    # if there is a tag matching tag_prefix, this yields TAG-NUM-gHEX[-dirty]
+    # if there isn't one, this yields HEX[-dirty] (no NUM)
+    describe_out = run_command(GITS, ["describe", "--tags", "--dirty",
+                                      "--always", "--long",
+                                      "--match", "%s*" % tag_prefix],
+                               cwd=root)
+    # --long was added in git-1.5.5
+    if describe_out is None:
+        raise NotThisMethod("'git describe' failed")
+    describe_out = describe_out.strip()
+    full_out = run_command(GITS, ["rev-parse", "HEAD"], cwd=root)
+    if full_out is None:
+        raise NotThisMethod("'git rev-parse' failed")
+    full_out = full_out.strip()
+
+    pieces = {}
+    pieces["long"] = full_out
+    pieces["short"] = full_out[:7]  # maybe improved later
+    pieces["error"] = None
+
+    # parse describe_out. It will be like TAG-NUM-gHEX[-dirty] or HEX[-dirty]
+    # TAG might have hyphens.
+    git_describe = describe_out
+
+    # look for -dirty suffix
+    dirty = git_describe.endswith("-dirty")
+    pieces["dirty"] = dirty
+    if dirty:
+        git_describe = git_describe[:git_describe.rindex("-dirty")]
+
+    # now we have TAG-NUM-gHEX or HEX
+
+    if "-" in git_describe:
+        # TAG-NUM-gHEX
+        mo = re.search(r'^(.+)-(\d+)-g([0-9a-f]+)$', git_describe)
+        if not mo:
+            # unparseable. Maybe git-describe is misbehaving?
+            pieces["error"] = ("unable to parse git-describe output: '%s'"
+                               % describe_out)
+            return pieces
+
+        # tag
+        full_tag = mo.group(1)
+        if not full_tag.startswith(tag_prefix):
+            if verbose:
+                fmt = "tag '%s' doesn't start with prefix '%s'"
+                print(fmt % (full_tag, tag_prefix))
+            pieces["error"] = ("tag '%s' doesn't start with prefix '%s'"
+                               % (full_tag, tag_prefix))
+            return pieces
+        pieces["closest-tag"] = full_tag[len(tag_prefix):]
+
+        # distance: number of commits since tag
+        pieces["distance"] = int(mo.group(2))
+
+        # commit: short hex revision ID
+        pieces["short"] = mo.group(3)
+
+    else:
+        # HEX: no tags
+        pieces["closest-tag"] = None
+        count_out = run_command(GITS, ["rev-list", "HEAD", "--count"],
+                                cwd=root)
+        pieces["distance"] = int(count_out)  # total number of commits
+
+    return pieces
+
+
+def plus_or_dot(pieces):
+    """Return a + if we don't already have one, else return a ."""
+    if "+" in pieces.get("closest-tag", ""):
+        return "."
+    return "+"
+
+
+def render_pep440(pieces):
+    """Build up version string, with post-release "local version identifier".
+
+    Our goal: TAG[+DISTANCE.gHEX[.dirty]] . Note that if you
+    get a tagged build and then dirty it, you'll get TAG+0.gHEX.dirty
+
+    Exceptions:
+    1: no tags. git_describe was just HEX. 0+untagged.DISTANCE.gHEX[.dirty]
+    """
+    if pieces["closest-tag"]:
+        rendered = pieces["closest-tag"]
+        if pieces["distance"] or pieces["dirty"]:
+            rendered += plus_or_dot(pieces)
+            rendered += "%d.g%s" % (pieces["distance"], pieces["short"])
+            if pieces["dirty"]:
+                rendered += ".dirty"
+    else:
+        # exception #1
+        rendered = "0+untagged.%d.g%s" % (pieces["distance"],
+                                          pieces["short"])
+        if pieces["dirty"]:
+            rendered += ".dirty"
+    return rendered
+
+
+def render_pep440_pre(pieces):
+    """TAG[.post.devDISTANCE] -- No -dirty.
+
+    Exceptions:
+    1: no tags. 0.post.devDISTANCE
+    """
+    if pieces["closest-tag"]:
+        rendered = pieces["closest-tag"]
+        if pieces["distance"]:
+            rendered += ".post.dev%d" % pieces["distance"]
+    else:
+        # exception #1
+        rendered = "0.post.dev%d" % pieces["distance"]
+    return rendered
+
+
+def render_pep440_post(pieces):
+    """TAG[.postDISTANCE[.dev0]+gHEX] .
+
+    The ".dev0" means dirty. Note that .dev0 sorts backwards
+    (a dirty tree will appear "older" than the corresponding clean one),
+    but you shouldn't be releasing software with -dirty anyways.
+
+    Exceptions:
+    1: no tags. 0.postDISTANCE[.dev0]
+    """
+    if pieces["closest-tag"]:
+        rendered = pieces["closest-tag"]
+        if pieces["distance"] or pieces["dirty"]:
+            rendered += ".post%d" % pieces["distance"]
+            if pieces["dirty"]:
+                rendered += ".dev0"
+            rendered += plus_or_dot(pieces)
+            rendered += "g%s" % pieces["short"]
+    else:
+        # exception #1
+        rendered = "0.post%d" % pieces["distance"]
+        if pieces["dirty"]:
+            rendered += ".dev0"
+        rendered += "+g%s" % pieces["short"]
+    return rendered
+
+
+def render_pep440_old(pieces):
+    """TAG[.postDISTANCE[.dev0]] .
+
+    The ".dev0" means dirty.
+
+    Eexceptions:
+    1: no tags. 0.postDISTANCE[.dev0]
+    """
+    if pieces["closest-tag"]:
+        rendered = pieces["closest-tag"]
+        if pieces["distance"] or pieces["dirty"]:
+            rendered += ".post%d" % pieces["distance"]
+            if pieces["dirty"]:
+                rendered += ".dev0"
+    else:
+        # exception #1
+        rendered = "0.post%d" % pieces["distance"]
+        if pieces["dirty"]:
+            rendered += ".dev0"
+    return rendered
+
+
+def render_git_describe(pieces):
+    """TAG[-DISTANCE-gHEX][-dirty].
+
+    Like 'git describe --tags --dirty --always'.
+
+    Exceptions:
+    1: no tags. HEX[-dirty]  (note: no 'g' prefix)
+    """
+    if pieces["closest-tag"]:
+        rendered = pieces["closest-tag"]
+        if pieces["distance"]:
+            rendered += "-%d-g%s" % (pieces["distance"], pieces["short"])
+    else:
+        # exception #1
+        rendered = pieces["short"]
+    if pieces["dirty"]:
+        rendered += "-dirty"
+    return rendered
+
+
+def render_git_describe_long(pieces):
+    """TAG-DISTANCE-gHEX[-dirty].
+
+    Like 'git describe --tags --dirty --always -long'.
+    The distance/hash is unconditional.
+
+    Exceptions:
+    1: no tags. HEX[-dirty]  (note: no 'g' prefix)
+    """
+    if pieces["closest-tag"]:
+        rendered = pieces["closest-tag"]
+        rendered += "-%d-g%s" % (pieces["distance"], pieces["short"])
+    else:
+        # exception #1
+        rendered = pieces["short"]
+    if pieces["dirty"]:
+        rendered += "-dirty"
+    return rendered
+
+
+def render(pieces, style):
+    """Render the given version pieces into the requested style."""
+    if pieces["error"]:
+        return {"version": "unknown",
+                "full-revisionid": pieces.get("long"),
+                "dirty": None,
+                "error": pieces["error"]}
+
+    if not style or style == "default":
+        style = "pep440"  # the default
+
+    if style == "pep440":
+        rendered = render_pep440(pieces)
+    elif style == "pep440-pre":
+        rendered = render_pep440_pre(pieces)
+    elif style == "pep440-post":
+        rendered = render_pep440_post(pieces)
+    elif style == "pep440-old":
+        rendered = render_pep440_old(pieces)
+    elif style == "git-describe":
+        rendered = render_git_describe(pieces)
+    elif style == "git-describe-long":
+        rendered = render_git_describe_long(pieces)
+    else:
+        raise ValueError("unknown style '%s'" % style)
+
+    return {"version": rendered, "full-revisionid": pieces["long"],
+            "dirty": pieces["dirty"], "error": None}
+
+
+def get_versions():
+    """Get version information or return default if unable to do so."""
+    # I am in _version.py, which lives at ROOT/VERSIONFILE_SOURCE. If we have
+    # __file__, we can work backwards from there to the root. Some
+    # py2exe/bbfreeze/non-CPython implementations don't do __file__, in which
+    # case we can only use expanded keywords.
+
+    cfg = get_config()
+    verbose = cfg.verbose
+
+    try:
+        return git_versions_from_keywords(get_keywords(), cfg.tag_prefix,
+                                          verbose)
+    except NotThisMethod:
+        pass
+
+    try:
+        root = os.path.realpath(__file__)
+        # versionfile_source is the relative path from the top of the source
+        # tree (where the .git directory might live) to this file. Invert
+        # this to find the root from __file__.
+        for i in cfg.versionfile_source.split('/'):
+            root = os.path.dirname(root)
+    except NameError:
+        return {"version": "0+unknown", "full-revisionid": None,
+                "dirty": None,
+                "error": "unable to find root of source tree"}
+
+    try:
+        pieces = git_pieces_from_vcs(cfg.tag_prefix, root, verbose)
+        return render(pieces, cfg.style)
+    except NotThisMethod:
+        pass
+
+    try:
+        if cfg.parentdir_prefix:
+            return versions_from_parentdir(cfg.parentdir_prefix, root, verbose)
+    except NotThisMethod:
+        pass
+
+    return {"version": "0+unknown", "full-revisionid": None,
+            "dirty": None,
+            "error": "unable to compute version"}

cachier/core.py

@@ -7,22 +7,23 @@
 # http://www.opensource.org/licenses/MIT-license
 # Copyright (c) 2016, Shay Palachy <shaypal5@gmail.com>
 
-# Used a little code from Andrew Barnert's <abarnert at yahoo.com>
-# persistent-lru-cache, which can be found at
-# https://github.com/abarnert/persistent-lru-cache
-
 import os
 from functools import wraps
 import pickle  # for local caching
 import datetime
 import abc  # for the _BaseCore abstract base class
 import concurrent.futures  # for asynchronous file uploads
-from bson.binary import Binary  # to save binary data to mongodb
+import time   # to sleep when waiting on Mongo cache
+import fcntl  # to lock on pickle cache IO
 
+from bson.binary import Binary  # to save binary data to mongodb
+from watchdog.observers import Observer
+from watchdog.events import PatternMatchingEventHandler
 
 CACHIER_DIR = '~/.cachier/'
 EXPANDED_CACHIER_DIR = os.path.expanduser(CACHIER_DIR)
 DEFAULT_MAX_WORKERS = 5
+MONGO_SLEEP_DURATION_IN_SEC = 6
 
 
 # === Cores definitions ===
@@ -30,8 +31,7 @@ DEFAULT_MAX_WORKERS = 5
 class _BaseCore(object):
     __metaclass__ = abc.ABCMeta
 
-    def __init__(self, mongetter, stale_after, next_time):
-        self.mongetter = mongetter
+    def __init__(self, stale_after, next_time):
         self.stale_after = stale_after
         self.next_time = next_time
         self.func = None
@@ -41,6 +41,11 @@ class _BaseCore(object):
         any method is called"""
         self.func = func
 
+    @abc.abstractmethod
+    def get_entry_by_key(self, key):
+        """Returns the result mapped to the given key in this core's cache,
+        if such a mapping exists."""
+
     @abc.abstractmethod
     def get_entry(self, args, kwds):
         """Returns the result mapped to the given arguments in this core's
@@ -54,11 +59,17 @@ class _BaseCore(object):
     def mark_entry_being_calculated(self, key):
         """Marks the entry mapped by the given key as being calculated."""
 
+    @abc.abstractmethod
+    def wait_on_entry_calc(self, key):
+        """Waits on the entry mapped by key being calculated and returns the
+        result."""
+
 
 class _MongoCore(_BaseCore):
 
     def __init__(self, mongetter, stale_after, next_time):
-        super().__init__(mongetter, stale_after, next_time)
+        super().__init__(stale_after, next_time)
+        self.mongetter = mongetter
         self.mongo_collection = None
 
     @staticmethod
@@ -70,83 +81,168 @@ class _MongoCore(_BaseCore):
             self.mongo_collection = self.mongetter()
         return self.mongo_collection
 
-    def get_entry(self, args, kwds):
-        key = pickle.dumps(args + tuple(sorted(kwds.items())))
-        print('key type={}, key={}'.format(
-            type(key), key))
+    def get_entry_by_key(self, key):
         res = self._get_mongo_collection().find_one({
             'func': _MongoCore._get_func_str(self.func),
             'key': key
         })
         if res:
-            entry = {
-                'value': pickle.loads(res['value']),
-                'time': res['time'],
-                'stale': res['stale'],
-                'being_calculated': res['being_calculated']
-            }
+            try:
+                entry = {
+                    'value': pickle.loads(res['value']),
+                    'time': res.get('time', None),
+                    'stale': res.get('stale', False),
+                    'being_calculated': res.get('being_calculated', False)
+                }
+            except KeyError:
+                entry = {
+                    'value': None,
+                    'time': res.get('time', None),
+                    'stale': res.get('stale', False),
+                    'being_calculated': res.get('being_calculated', False)
+                }
             return key, entry
         return key, None
 
+    def get_entry(self, args, kwds):
+        key = pickle.dumps(args + tuple(sorted(kwds.items())))
+        print('key type={}, key={}'.format(
+            type(key), key))
+        return self.get_entry_by_key(key)
+
     def set_entry(self, key, func_res):
         thebytes = pickle.dumps(func_res)
-        self._get_mongo_collection().insert_one({
-            'func': _MongoCore._get_func_str(self.func),
-            'key': key,
-            'value': Binary(thebytes),
-            'time': datetime.datetime.now(),
-            'stale': False,
-            'being_calculated': False
-        })
-
-    def mark_entry_being_calculated(self, key):
-        self._get_mongo_collection().update(
+        self._get_mongo_collection().update_one(
             {
                 'func': _MongoCore._get_func_str(self.func),
                 'key': key
             },
             {
-                'being_calculated': False
-            }
+                '$set': {
+                    'func': _MongoCore._get_func_str(self.func),
+                    'key': key,
+                    'value': Binary(thebytes),
+                    'time': datetime.datetime.now(),
+                    'stale': False,
+                    'being_calculated': False
+                }
+            },
+            upsert=True
         )
 
+    def mark_entry_being_calculated(self, key):
+        self._get_mongo_collection().update_one(
+            {
+                'func': _MongoCore._get_func_str(self.func),
+                'key': key
+            },
+            {
+                '$set': {'being_calculated': True}
+            },
+            upsert=True
+        )
+
+    def wait_on_entry_calc(self, key):
+        while True:
+            time.sleep(MONGO_SLEEP_DURATION_IN_SEC)
+            key, entry = self.get_entry_by_key(key)
+            if entry and not entry['being_calculated']:
+                return entry['value']
+        # key, entry = self.get_entry_by_key(key)
+        # if entry:
+        #     return entry['value']
+        # return None
+
 
 class _PickleCore(_BaseCore):
 
-    def __init__(self, mongetter, stale_after, next_time):
-        super().__init__(mongetter, stale_after, next_time)
+    class CacheChangeHandler(PatternMatchingEventHandler):
+        """Handles cache-file modification events."""
+
+        def __init__(self, filename, core, key):
+            super(_PickleCore.CacheChangeHandler, self).__init__(
+                patterns=["*" + filename],
+                ignore_patterns=None,
+                ignore_directories=True,
+                case_sensitive=False
+            )
+            self.core = core
+            self.key = key
+            self.observer = None
+            self.value = None
+
+        def inject_observer(self, observer):
+            """Inject the observer running this handler."""
+            self.observer = observer
+
+        def _check_calculation(self):
+            print('checking calc')
+            entry = self.core.get_entry_by_key(self.key, True)[1]
+            print(entry)
+            if not entry['being_calculated']:
+                self.value = entry['value']
+                self.observer.stop()
+
+        def on_created(self, event):
+            self._check_calculation()
+
+        def on_modified(self, event):
+            self._check_calculation()
+
+    def __init__(self, stale_after, next_time, reload):
+        super().__init__(stale_after, next_time)
         self.cache = None
+        self.reload = reload
+
+    def _get_cache_file_name(self):
+        return '.{}.{}'.format(
+            self.func.__module__, self.func.__name__)  # pylint: disable=W0212
 
     def _get_cache_path(self):
         # print(EXPANDED_CACHIER_DIR)
         if not os.path.exists(EXPANDED_CACHIER_DIR):
             os.makedirs(EXPANDED_CACHIER_DIR)
-        fname = '.{}.{}'.format(
-            self.func.__module__, self.func.__name__)  # pylint: disable=W0212
         fpath = os.path.abspath(os.path.join(
-            os.path.realpath(EXPANDED_CACHIER_DIR), fname))
+            os.path.realpath(EXPANDED_CACHIER_DIR),
+            self._get_cache_file_name()
+        ))
         # print(fpath)
         return fpath
 
+    def _reload_cache(self):
+        fpath = self._get_cache_path()
+        try:
+            with open(fpath, 'rb') as cache_file:
+                fcntl.flock(cache_file, fcntl.LOCK_SH)
+                self.cache = pickle.load(cache_file)
+                fcntl.flock(cache_file, fcntl.LOCK_UN)
+        except FileNotFoundError:
+            self.cache = {}
+
     def _get_cache(self):
         if not self.cache:
-            fpath = self._get_cache_path()
-            try:
-                self.cache = pickle.load(open(fpath, 'rb'))
-            except FileNotFoundError:
-                self.cache = {}
+            self._reload_cache()
         return self.cache
 
     def _save_cache(self, cache):
         self.cache = cache
         fpath = self._get_cache_path()
-        pickle.dump(cache, open(fpath, 'wb'))
+        with open(fpath, 'wb') as cache_file:
+            fcntl.flock(cache_file, fcntl.LOCK_EX)
+            pickle.dump(cache, cache_file)
+            fcntl.flock(cache_file, fcntl.LOCK_UN)
+        self._reload_cache()
+
+    def get_entry_by_key(self, key, reload=False):  # pylint: disable=W0221
+        print('{}, {}'.format(self.reload, reload))
+        if self.reload or reload:
+            self._reload_cache()
+        return key, self._get_cache().get(key, None)
 
     def get_entry(self, args, kwds):
         key = args + tuple(sorted(kwds.items()))
         print('key type={}, key={}'.format(type(key), key))
-        cache = self._get_cache()
-        return key, cache.get(key, None)
+        return self.get_entry_by_key(key)
 
     def set_entry(self, key, func_res):
         cache = self._get_cache()
@@ -160,9 +256,38 @@ class _PickleCore(_BaseCore):
 
     def mark_entry_being_calculated(self, key):
         cache = self._get_cache()
-        cache[key]['being_calculated'] = True
+        try:
+            cache[key]['being_calculated'] = True
+        except KeyError:
+            cache[key] = {
+                'value': None,
+                'time': datetime.datetime.now(),
+                'stale': False,
+                'being_calculated': True
+            }
         self._save_cache(cache)
 
+    def wait_on_entry_calc(self, key):
+        entry = self._get_cache()[key]
+        if not entry['being_calculated']:
+            return entry['value']
+        event_handler = _PickleCore.CacheChangeHandler(
+            filename=self._get_cache_file_name(),
+            core=self,
+            key=key
+        )
+        observer = Observer()
+        event_handler.inject_observer(observer)
+        observer.schedule(
+            event_handler,
+            path=EXPANDED_CACHIER_DIR,
+            recursive=True
+        )
+        observer.start()
+        observer.join()
+        print("Returned value: {}".format(event_handler.value))
+        return event_handler.value
+
 
 # === Main functionality ===
 
@@ -197,12 +322,14 @@ def _function_thread(core, key, func, args, kwds):
         core.set_entry(key, func_res)
     except BaseException as exc:  # pylint: disable=W0703
         print(
-            'Function call failed with following exception:\n{}'.format(exc),
+            'Function call failed with the following exception:\n{}'.format(
+                exc),
             flush=True
         )
 
 
-def cachier(mongetter=None, stale_after=None, next_time=True):
+def cachier(stale_after=None, next_time=True, pickle_reload=True,
+            mongetter=None):
     """A persistent, stale-free memoization decorator.
 
     When using a MongoDB-backed caching, the positional and keyword arguments
@@ -214,10 +341,6 @@ def cachier(mongetter=None, stale_after=None, next_time=True):
 
     Arguments
     ---------
-    mongetter (optional) : callable
-        A callable that takes no arguments and returns a pymongo.Collection
-        object with writing permissions. If unset a local pickle cache is used
-        instead.
     stale_after (optional) : datetime.timedelta
         The time delta afterwhich a cached result is considered stale. Calls
         made after the result goes stale will trigger a recalculation of the
@@ -227,6 +350,14 @@ def cachier(mongetter=None, stale_after=None, next_time=True):
         If set to True, a stale result will be returned when finding one, not
         waiting for the calculation of the fresh result to return. Defaults to
         True.
+    pickle_reload (optional) : bool
+        If set to True, in-memory cache will be reloaded on each cache read,
+        enabling different threads to share cache. Should be set to False for
+        faster reads in single-read programs. Defaults to True.
+    mongetter (optional) : callable
+        A callable that takes no arguments and returns a pymongo.Collection
+        object with writing permissions. If unset a local pickle cache is used
+        instead.
     """
     print('Inside the wrapper maker')
     print('mongetter={}'.format(mongetter))
@@ -237,35 +368,52 @@ def cachier(mongetter=None, stale_after=None, next_time=True):
         core = _MongoCore(mongetter, stale_after, next_time)
     else:
         core = _PickleCore(  # pylint: disable=R0204
-            mongetter, stale_after, next_time)
+            stale_after, next_time, pickle_reload)
 
     def _cachier_decorator(func):
         core.set_func(func)
 
         @wraps(func)
-        def func_wrapper(*args, **kwds):  # pylint: disable=C0111
+        def func_wrapper(*args, **kwds):  # pylint: disable=C0111,R0911
             print('Inside general wrapper for {}.'.format(func.__name__))
             key, entry = core.get_entry(args, kwds)
-            if entry:
-                print('Cached result found.')
-                if stale_after:
-                    now = datetime.datetime.now()
-                    if now - entry['time'] > stale_after:
-                        if next_time:
+            if entry:  # pylint: disable=R0101
+                print('Entry found.')
+                if entry.get('value', None):
+                    print('Cached result found.')
+                    if stale_after:
+                        now = datetime.datetime.now()
+                        if now - entry['time'] > stale_after:
+                            print('But it is stale... :(')
                             if entry['being_calculated']:
+                                print('Already calculated. Waiting on change.')
+                                return core.wait_on_entry_calc(key)
+                            if next_time:
+                                if entry['being_calculated']:
+                                    return entry['value']  # return stale val
+                                # trigger async calculation and return stale
+                                core.mark_entry_being_calculated(key)
+                                _get_executor().submit(
+                                    _function_thread, core, key, func, args,
+                                    kwds)
                                 return entry['value']
-                            # trigger async calculation and return stale
+                            print('Calling decorated function and waiting')
                             core.mark_entry_being_calculated(key)
+                            func_res = func(*args, **kwds)
                             _get_executor().submit(
-                                _function_thread, core, key, func, args, kwds)
-                            return entry['value']
-                        print('Calling decorated function and waiting')
-                        func_res = func(*args, **kwds)
-                        core.set_entry(key, func_res)
-                        return func_res
-                return entry['value']
+                                core.set_entry, key, func_res)
+                            # core.set_entry(key, func_res)
+                            return func_res
+                    print('And it is fresh!')
+                    return entry['value']
+                if entry['being_calculated']:
+                    print('No value but already being calculated. Waiting.')
+                    return core.wait_on_entry_calc(key)
+            # core.mark_entry_being_calculated(key)
+            print('No entry found. Calling like a boss.')
+            _get_executor().submit(core.mark_entry_being_calculated, key)
             func_res = func(*args, **kwds)
-            core.set_entry(key, func_res)
+            _get_executor().submit(core.set_entry, key, func_res)
             return func_res
         return func_wrapper
 

setup.py

@@ -22,6 +22,6 @@ setup(
     packages=find_packages(),
     install_requires=[
     ],
-    keywords=['cache', 'persistence', 'mongo'],
+    keywords=['cache', 'persistence', 'mongo', 'memoization', 'decorator'],
     classifiers=[],
 )

tests/test_cachier.py

@@ -7,6 +7,8 @@
 # http://www.opensource.org/licenses/MIT-license
 # Copyright (c) 2016, Shay Palachy <shaypal5@gmail.com>
 
+import time
+import datetime
 from cachier import cachier
 from datapy.mongo import get_collection
 
@@ -15,13 +17,48 @@ def _mongo_getter():
     return get_collection('cachier', server_name='production', mode='writing')
 
 
+# Pickle core tests
+
 @cachier()
 def test_int_pickling(int_1, int_2):
     """Add the two given ints."""
     return int_1 + int_2
 
 
+@cachier(next_time=False)
+def takes_30_seconds(arg_1, arg_2):
+    """Some function."""
+    time.sleep(30)
+    return 'arg_1:{}, arg_2:{}'.format(arg_1, arg_2)
+
+
+DELTA = datetime.timedelta(seconds=10)
+
+
+@cachier(stale_after=DELTA, next_time=False)
+def stale_after_seconds(arg_1, arg_2):
+    """Some function."""
+    return {'arg_1': arg_1, 'arg_2': arg_2}
+
+
+# Mongo core tests
+
 @cachier(mongetter=_mongo_getter)
 def test_mongo_caching(arg_1, arg_2):
     """Some function."""
     return 'arg_1:{}, arg_2:{}'.format(arg_1, arg_2)
+
+
+@cachier(mongetter=_mongo_getter, next_time=False)
+def takes_30_seconds_mongo(arg_1, arg_2):
+    """Some function."""
+    time.sleep(30)
+    return 'arg_1:{}, arg_2:{}'.format(arg_1, arg_2)
+
+MONGO_DELTA = datetime.timedelta(seconds=30)
+
+
+@cachier(mongetter=_mongo_getter, stale_after=MONGO_DELTA, next_time=False)
+def stale_after_mongo(arg_1, arg_2):
+    """Some function."""
+    return {'arg_1': arg_1, 'arg_2': arg_2}