Synchronization decorator; provide thread-safe locking on a function
    http://code.activestate.com/recipes/465057/
194367
def synchronized_limit(lock):
    
    def wrap(f):
        def synchronize(*args, **kw):
            if lock[1] < 10:
                lock[1] += 1
                lock[0].acquire()
                try:
                    return f(*args, **kw)
                finally:
                    lock[1] -= 1
                    lock[0].release()
            else:
                raise Exception('Too busy')
        return synchronize
    return wrap
624144
def sync(lock):
    
    def sync(f):
        @wraps(f)
        def new_function(*args, **kwargs):
            lock.acquire()
            try:
                return f(*args, **kwargs)
            finally:
                lock.release()
        return new_function
    return sync
597877
def cached_method(func):
    
    @functools.wraps(func)
    def wrapper(self, *args):
        if not hasattr(self, "_cache"):
            self._cache = {}
        key = _argstring((func.__name__,) + args)
        if key not in self._cache:
            self._cache[key] = func(self, *args)
        return self._cache[key]
    return wrapper
255559
def with_lock(lock, func, *args, **kwargs):
    
    d = lock.acquire()

    def release_lock(result):
        deferred = lock.release()
        return deferred.addCallback(lambda x: result)

    def lock_acquired(lock):
        return defer.maybeDeferred(func, *args, **kwargs).addBoth(release_lock)

    d.addCallback(lock_acquired)
    return d
1263190
def memoize_methodcalls(func, pickle=False, dumps=pickle.dumps):
    '''
    '''
    cache = func._memoize_cache = {}
    @functools.wraps(func)
    def memoizer(self, *args, **kwargs):
        if pickle:
            key = dumps((args, kwargs))
        else:
            key = args
        if key not in cache:
            cache[key] = func(self, *args, **kwargs)
        return cache[key]
    return memoizer
433249
def decorate_callable(self, func):
        
        def wrapper(*args, **kwargs):
            try:
                with self:
                    result = func(*args, **kwargs)
                return result
            except DBMutexError as e:
                if self.suppress_acquisition_exceptions:
                    LOG.error(e)
                else:
                    raise e
        functools.update_wrapper(wrapper, func)
        return wrapper
893733
def synchronized(func):  # pylint: disable=invalid-name
  """"""
  @functools.wraps(func)
  def synchronized_method(self, *args, **kwargs):
    """Wrapper to return."""
    if not hasattr(self, '_lock'):
      if func.__name__ in type(self).__dict__:
        hint = ''
      else:
        hint = (' Might be missing call to super in %s.__init__?' %
                type(self).__name__)
      raise RuntimeError('Can\'t synchronize method `%s` of %s without '
                         'attribute `_lock`.%s' %
                         (func.__name__, type(self).__name__, hint))
    with self._lock:  # pylint: disable=protected-access
      return func(self, *args, **kwargs)
  return synchronized_method
1187215
def _synced(method, self, args, kwargs):
    """"""
    with self._lock:
        return method(*args, **kwargs)
1307451
def return_self_updater(func):
    '''
    
    '''
    @functools.wraps(func)
    def decorator(k,v):
        func(k,v)
        return v
    return decorator
942445
async def async_enqueue_sync(self, func, *func_args):
        '''
        
        '''
        worker = self.pick_sticky(0)  # just pick first always
        args = (func,) + func_args
        await worker.enqueue(enums.Task.FUNC, args)
770129
def run_on_new_thread(func, daemon=True):
    
    f = concurrent.futures.Future()
    def wrapper():
        if not f.set_running_or_notify_cancel():
            return
        try:
            f.set_result(func())
        except Exception as e:
            f.set_exception(e)
    t = threading.Thread(target=wrapper)
    t.daemon = daemon
    t.start()
    return f
204715
def sync(func):
    """"""
    sync_timeout = 3600  # Match standard synchronous timeout.

    def wraps(*args, **kwargs):
        task = func(*args, **kwargs)
        task.wait_for_result(timeout=sync_timeout)
        result = json.loads(task.result)
        return result

    return wraps
1361054
def with_lock(lock):
  """"""
  def decorator(func):
    @functools.wraps(func)
    def _with_lock(*args, **kwargs):
      with lock:
        return func(*args, **kwargs)
    return _with_lock
  return decorator
323406
def thread_safe(method):
    

    @functools.wraps(method)
    def _locker(self, *args, **kwargs):
        assert hasattr(self, 'lock'), \
            'thread_safe decorator applied to method {0}.{1}: missing required field {0}.lock'\
            .format(self.__class__.__name__, method.__name__)

        try:
            self.lock.acquire()
            return method(self, *args, **kwargs)
        finally:
            try:
                self.lock.release()
            except:
                sys.stderr.write('Exception on releasing lock at method {0}'.format(method.__name__))
                traceback.print_exc(file=sys.stderr)

    return _locker
284908
def lock_host(func, *args, **kwargs):
    
    def inner(self, *args, **kwargs):
        self.host_lock.acquire()
        try:
            res = func(self, *args, **kwargs)
            self.host_lock.release()
        except Exception as e:
            self.host_lock.release()
            raise e
        return res
    return inner
19607
def threadpooled(
    func: typing.Callable[..., typing.Union["typing.Awaitable[typing.Any]", typing.Any]],
    *,
    loop_getter: typing.Union[typing.Callable[..., asyncio.AbstractEventLoop], asyncio.AbstractEventLoop],
    loop_getter_need_context: bool = False,
) -> typing.Callable[..., "asyncio.Task[typing.Any]"]:
    """"""
1351141
def singleton(func):
    

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        if wrapper.__call_state__ == CALLED:
            return
        ret = func(*args, **kwargs)
        wrapper.__call_state__ = CALLED
        return ret

    def reset():
        wrapper.__call_state__ = NOT_CALLED

    wrapper.reset = reset
    reset()

    # save original func to be able to patch and restore multiple times from
    # unit tests
    wrapper.__original_func = func
    return wrapper