Tuesday, October 4

Case Studies in Python Memory Analysis using High-Watermark Testing

#!/usr/bin/env python3

from collections import namedtuple
from functools import wraps
from inspect import signature
from itertools import islice, tee
from math import isclose
from shlex import quote
from subprocess import check_call, DEVNULL, CalledProcessError
from textwrap import dedent

from numpy import arange

nwise = lambda g, *, n=2: zip(*(islice(g, i, None) for i, g in enumerate(tee(g, n))))

def binary_search(values):
    lidx, ridx = 0, len(values) - 1
    while True:
        if (ridx - lidx) <= 1:
        if (yield values[(midx := (ridx - lidx) // 2 + lidx)]):
            lidx = midx
            ridx = midx

class BisectedRun(namedtuple('RunBase', 'signature result exception')):
    sig = property(lambda s: s.signature)
    res = property(lambda s: s.result)
    exc = property(lambda s: s.exception)

    @lambda cls: cls()
    class args:
        def __get__(self, instance, owner):
            return namedtuple('Arguments', instance.sig.arguments.keys())(**instance.sig.arguments)

    def support(cls, arguments, exceptions):
        def dec(f):
            sig = signature(f)
            @lambda bisected: setattr(f, 'bisected', bisected)
            def bisected(*arg_args, **arg_kwargs):
                ci = binary_search(arguments(*arg_args, **arg_kwargs))
                def inner(*f_fixed_args, **f_fixed_kwargs):
                    (f_args, f_kwargs), exc = next(ci), None
                    while True:
                        f_bound_args = sig.bind(
                            *f_fixed_args, *f_args,
                            **f_fixed_kwargs, **f_kwargs
                            res = f(*f_bound_args.args, **f_bound_args.kwargs)
                        except (*exceptions,) as e:
                            exc = e
                            exc = None
                        yield cls(f_bound_args, exc, res)
                            f_args, f_kwargs = ci.send(exc is not None)
                        except StopIteration:
                return inner
            return f
        return dec

    arguments=lambda min_vmem, max_vmem:
        [((), {'vmem': m}) for m in arange(min_vmem, max_vmem, 1024, dtype=int)],
def run_python_ulimit(code, vmem):
    return check_call([
        'sh', '-c', f'''(
            ulimit -v {vmem}
            python -c {quote(code)}
    ], stderr=DEVNULL, stdout=DEVNULL)

if __name__ == '__main__':
    #### Case Study: NumPy Import
    if True:
        print(' {} '.format('What is the minimum memory needed to `import numpy`?').center(120, '\N{box drawings light horizontal}'))

        code = dedent('''
            import numpy

        for prev_run, curr_run in nwise(run_python_ulimit.bisected(0, 2**20)(code)):
        bad_mem, ok_mem = sorted([prev_run.args.vmem, curr_run.args.vmem])

            run_python_ulimit(code, ok_mem)
        except Exception as e:
            raise ValueError(f'unanticipated failure with {ok_mem = }') from e
            run_python_ulimit(code, bad_mem)
        except Exception as e: pass
        else: raise ValueError(f'anticipate failure with {bad_mem = }')

        print(f'Need `ulimit -v ≥{ok_mem}` to `import numpy`.')

    #### Case Study: Attribute Dictionaries
    if True:
        print(' {} '.format('What is the preferred implementation of `attrdict`?').center(120, '\N{box drawings light horizontal}'))

        attrdict_protocols = dedent('''
            # attrdict via dispatch to `dict.__*item__` protocols
            class attrdict(dict):
                __getattr__ = dict.__getitem__
                __setattr__ = dict.__setitem__
                __delattr__ = dict.__delitem__
        attrdict_circular = dedent('''
            # attrdict via circular reference on `self.__dict__`
            class attrdict(dict):
                def __init__(self, *args, **kwargs):
                    super().__init__(*args, **kwargs)
                    self.__dict__ = self
        simple_test = dedent('''
            for _ in range(2):

Leave a Reply