path: root/rpython/rlib/debug.py

import sys
import time
from collections import Counter

from rpython.rlib.objectmodel import enforceargs
from rpython.rtyper.extregistry import ExtRegistryEntry
from rpython.rlib.objectmodel import we_are_translated, always_inline
from rpython.rlib.rarithmetic import is_valid_int, r_longlong
from rpython.rtyper.extfunc import register_external
from rpython.rtyper.lltypesystem import lltype
from rpython.rtyper.lltypesystem import rffi
from rpython.translator.tool.cbuild import ExternalCompilationInfo

# Expose these here (public interface)
from rpython.rtyper.debug import (
    ll_assert, FatalError, fatalerror, fatalerror_notb, debug_print_traceback,
    ll_assert_not_none)


class DebugLog(list):
    def debug_print(self, *args):
        self.append(('debug_print',) + args)

    def debug_start(self, category, time=None):
        self.append(('debug_start', category, time))

    def debug_stop(self, category, time=None):
        for i in xrange(len(self) - 1, -1, -1):
            if self[i][0] == 'debug_start':
                assert self[i][1] == category, (
                    "nesting error: starts with %r but stops with %r" %
                    (self[i][1], category))
                starttime = self[i][2]
                if starttime is not None or time is not None:
                    self[i:] = [(category, starttime, time, self[i + 1:])]
                else:
                    self[i:] = [(category, self[i + 1:])]
                return
        assert False, ("nesting error: no start corresponding to stop %r" %
                       (category,))

    def reset(self):
        # only for tests: empty the log
        self[:] = []

    def summary(self, flatten=False):
        res = Counter()
        def visit(lst):
            for section, sublist in lst:
                if section == 'debug_print':
                    continue
                res[section] += 1
                if flatten:
                    visit(sublist)
        #
        visit(self)
        return res

    def __repr__(self):
        import pprint
        return pprint.pformat(list(self))

_log = None       # patched from tests to be an object of class DebugLog
                  # or compatible

def debug_print(*args):
    for arg in args:
        print >> sys.stderr, arg,
    print >> sys.stderr
    if _log is not None:
        _log.debug_print(*args)

class Entry(ExtRegistryEntry):
    _about_ = debug_print

    def compute_result_annotation(self, *args_s):
        return None

    def specialize_call(self, hop):
        vlist = hop.inputargs(*hop.args_r)
        hop.exception_cannot_occur()
        t = hop.rtyper.annotator.translator
        if t.config.translation.log:
            hop.genop('debug_print', vlist)


if sys.stderr.isatty():
    _start_colors_1 = "\033[1m\033[31m"
    _start_colors_2 = "\033[31m"
    _stop_colors = "\033[0m"
else:
    _start_colors_1 = ""
    _start_colors_2 = ""
    _stop_colors = ""

@always_inline
@enforceargs(str, bool)
def debug_start(category, timestamp=False):
    """
    Start a PYPYLOG section.

    By default, the return value is undefined.  If timestamp is True, always
    return the current timestamp, even if PYPYLOG is not set.
    """
    return _debug_start(category, timestamp)

@always_inline
@enforceargs(str, bool)
def debug_stop(category, timestamp=False):
    """
    Stop a PYPYLOG section. See debug_start for docs about timestamp
    """
    return _debug_stop(category, timestamp)


def _debug_start(category, timestamp):
    c = int(time.clock() * 100)
    print >> sys.stderr, '%s[%x] {%s%s' % (_start_colors_1, c,
                                           category, _stop_colors)
    if _log is not None:
        _log.debug_start(category)

    if timestamp:
        return r_longlong(c)
    return r_longlong(-42) # random undefined value

def _debug_stop(category, timestamp):
    c = int(time.clock() * 100)
    print >> sys.stderr, '%s[%x] %s}%s' % (_start_colors_2, c,
                                           category, _stop_colors)
    if _log is not None:
        _log.debug_stop(category)

    if timestamp:
        return r_longlong(c)
    return r_longlong(-42) # random undefined value

class Entry(ExtRegistryEntry):
    _about_ = _debug_start, _debug_stop

    def compute_result_annotation(self, s_category, s_timestamp):
        from rpython.rlib.rtimer import s_TIMESTAMP
        return s_TIMESTAMP

    def specialize_call(self, hop):
        from rpython.rtyper.lltypesystem.rstr import string_repr
        from rpython.rlib.rtimer import TIMESTAMP_type
        fn = self.instance
        _, r_timestamp = hop.args_r
        vlist = hop.inputargs(string_repr, r_timestamp)
        hop.exception_cannot_occur()
        t = hop.rtyper.annotator.translator
        if t.config.translation.log:
            opname = fn.__name__[1:] # remove the '_'
            return hop.genop(opname, vlist, resulttype=TIMESTAMP_type)
        else:
            return hop.inputconst(TIMESTAMP_type, r_longlong(0))


def have_debug_prints():
    # returns True if the next calls to debug_print show up,
    # and False if they would not have any effect.
    return True

def have_debug_prints_for(category_prefix):
    # returns True if debug prints are enabled for at least some
    # category strings starting with "prefix" (must be a constant).
    assert len(category_prefix) > 0
    return True

class Entry(ExtRegistryEntry):
    _about_ = have_debug_prints, have_debug_prints_for

    def compute_result_annotation(self, s_prefix=None):
        from rpython.annotator import model as annmodel
        t = self.bookkeeper.annotator.translator
        if t.config.translation.log:
            return annmodel.s_Bool
        else:
            return self.bookkeeper.immutablevalue(False)

    def specialize_call(self, hop):
        t = hop.rtyper.annotator.translator
        hop.exception_cannot_occur()
        if t.config.translation.log:
            if hop.args_v:
                [c_prefix] = hop.args_v
                assert len(c_prefix.value) > 0
                args = [hop.inputconst(lltype.Void, c_prefix.value)]
                return hop.genop('have_debug_prints_for', args,
                                 resulttype=lltype.Bool)
            return hop.genop('have_debug_prints', [], resulttype=lltype.Bool)
        else:
            return hop.inputconst(lltype.Bool, False)


def debug_offset():
    """ Return an offset in log file
    """
    return -1

class Entry(ExtRegistryEntry):
    _about_ = debug_offset

    def compute_result_annotation(self):
        from rpython.annotator import model as annmodel
        return annmodel.SomeInteger()

    def specialize_call(self, hop):
        hop.exception_cannot_occur()
        return hop.genop('debug_offset', [], resulttype=lltype.Signed)


def debug_flush():
    """ Flushes the debug file.

    With the reverse-debugger, it also closes the output log.
    """
    pass

class Entry(ExtRegistryEntry):
    _about_ = debug_flush

    def compute_result_annotation(self):
        return None

    def specialize_call(self, hop):
        hop.exception_cannot_occur()
        return hop.genop('debug_flush', [])


def debug_forked(original_offset):
    """ Call after a fork(), passing as argument the result of
        debug_offset() called before the fork.
    """
    pass

class Entry(ExtRegistryEntry):
    _about_ = debug_forked

    def compute_result_annotation(self, s_original_offset):
        return None

    def specialize_call(self, hop):
        vlist = hop.inputargs(lltype.Signed)
        hop.exception_cannot_occur()
        return hop.genop('debug_forked', vlist)


def llinterpcall(RESTYPE, pythonfunction, *args):
    """When running on the llinterp, this causes the llinterp to call to
    the provided Python function with the run-time value of the given args.
    The Python function should return a low-level object of type RESTYPE.
    This should never be called after translation: use this only if
    running_on_llinterp is true.
    """
    raise NotImplementedError

class Entry(ExtRegistryEntry):
    _about_ = llinterpcall

    def compute_result_annotation(self, s_RESTYPE, s_pythonfunction, *args_s):
        from rpython.annotator import model as annmodel
        from rpython.rtyper.llannotation import lltype_to_annotation
        assert s_RESTYPE.is_constant()
        assert s_pythonfunction.is_constant()
        s_result = s_RESTYPE.const
        if isinstance(s_result, lltype.LowLevelType):
            s_result = lltype_to_annotation(s_result)
        assert isinstance(s_result, annmodel.SomeObject)
        return s_result

    def specialize_call(self, hop):
        from rpython.annotator import model as annmodel
        RESTYPE = hop.args_s[0].const
        if not isinstance(RESTYPE, lltype.LowLevelType):
            assert isinstance(RESTYPE, annmodel.SomeObject)
            r_result = hop.rtyper.getrepr(RESTYPE)
            RESTYPE = r_result.lowleveltype
        pythonfunction = hop.args_s[1].const
        c_pythonfunction = hop.inputconst(lltype.Void, pythonfunction)
        args_v = [hop.inputarg(hop.args_r[i], arg=i)
                  for i in range(2, hop.nb_args)]
        hop.exception_is_here()
        return hop.genop('debug_llinterpcall', [c_pythonfunction] + args_v,
                         resulttype=RESTYPE)


def check_annotation(arg, checker):
    """ Function checking if annotation is as expected when translating,
    does nothing when just run. Checker is supposed to be a constant
    callable which checks if annotation is as expected,
    arguments passed are (current annotation, bookkeeper)
    """
    return arg

class Entry(ExtRegistryEntry):
    _about_ = check_annotation

    def compute_result_annotation(self, s_arg, s_checker):
        if not s_checker.is_constant():
            raise ValueError(
                "Second argument of check_annotation must be constant")
        checker = s_checker.const
        checker(s_arg, self.bookkeeper)
        return s_arg

    def specialize_call(self, hop):
        hop.exception_cannot_occur()
        return hop.inputarg(hop.args_r[0], arg=0)

def make_sure_not_resized(arg):
    """ Function checking whether annotation of SomeList is never resized,
    useful for debugging. Does nothing when run directly
    """
    return arg

class Entry(ExtRegistryEntry):
    _about_ = make_sure_not_resized

    def compute_result_annotation(self, s_arg):
        from rpython.annotator.model import SomeList, s_None
        if s_None.contains(s_arg):
            return s_arg    # only None: just return
        assert isinstance(s_arg, SomeList)
        # the logic behind it is that we try not to propagate
        # make_sure_not_resized, when list comprehension is not on
        config = self.bookkeeper.annotator.translator.config
        if config.translation.list_comprehension_operations:
            s_arg.listdef.never_resize()
        else:
            from rpython.annotator.annrpython import log
            log.WARNING(
                "make_sure_not_resized called, but has no effect since "
                "list_comprehension is off")
        return s_arg

    def specialize_call(self, hop):
        hop.exception_cannot_occur()
        return hop.inputarg(hop.args_r[0], arg=0)


def mark_dict_non_null(d):
    """ Mark dictionary as having non-null keys and values. A warning would
    be emitted (not an error!) in case annotation disagrees.

    This doesn't work for r_dicts. For them, pass
    r_dict(..., force_non_null=True) to the constructor.
    """
    assert isinstance(d, dict)
    return d


class DictMarkEntry(ExtRegistryEntry):
    _about_ = mark_dict_non_null

    def compute_result_annotation(self, s_dict):
        from rpython.annotator.model import SomeDict

        assert isinstance(s_dict, SomeDict)
        s_dict.dictdef.force_non_null = True
        return s_dict

    def specialize_call(self, hop):
        hop.exception_cannot_occur()
        return hop.inputarg(hop.args_r[0], arg=0)

class IntegerCanBeNegative(Exception):
    pass

class UnexpectedRUInt(Exception):
    pass

class ExpectedRegularInt(Exception):
    pass

class NegativeArgumentNotAllowed(Exception):
    pass

def check_nonneg(x):
    """Give a translation-time error if 'x' is not known to be non-negative.
    To help debugging, this also gives a translation-time error if 'x' is
    actually typed as an r_uint (in which case the call to check_nonneg()
    is a bit strange and probably unexpected).
    """
    try:
        assert type(x)(-1) < 0     # otherwise, 'x' is a r_uint or similar
    except NegativeArgumentNotAllowed:
        pass
    else:
        assert x >= 0
    return x

class Entry(ExtRegistryEntry):
    _about_ = check_nonneg

    def compute_result_annotation(self, s_arg):
        from rpython.annotator.model import SomeInteger
        if isinstance(s_arg, SomeInteger) and s_arg.unsigned:
            raise UnexpectedRUInt("check_nonneg() arg is a %s" % (
                s_arg.knowntype,))
        s_nonneg = SomeInteger(nonneg=True)
        if not s_nonneg.contains(s_arg):
            raise IntegerCanBeNegative
        return s_arg

    def specialize_call(self, hop):
        hop.exception_cannot_occur()
        return hop.inputarg(hop.args_r[0], arg=0)

def check_regular_int(x):
    """Give a translation-time error if 'x' is not a plain int
    (e.g. if it's a r_longlong or an r_uint).
    """
    assert is_valid_int(x)
    return x

class Entry(ExtRegistryEntry):
    _about_ = check_regular_int

    def compute_result_annotation(self, s_arg):
        from rpython.annotator.model import SomeInteger
        if not SomeInteger().contains(s_arg):
            raise ExpectedRegularInt(s_arg)
        return s_arg

    def specialize_call(self, hop):
        hop.exception_cannot_occur()
        return hop.inputarg(hop.args_r[0], arg=0)

def check_list_of_chars(l):
    if not we_are_translated():
        assert isinstance(l, list)
        for x in l:
            assert isinstance(x, (unicode, str)) and len(x) == 1
    return l

class NotAListOfChars(Exception):
    pass

class Entry(ExtRegistryEntry):
    _about_ = check_list_of_chars

    def compute_result_annotation(self, s_arg):
        from rpython.annotator.model import SomeList, s_None
        from rpython.annotator.model import SomeChar, SomeUnicodeCodePoint
        from rpython.annotator.model import SomeImpossibleValue
        if s_None.contains(s_arg):
            return s_arg    # only None: just return
        assert isinstance(s_arg, SomeList)
        if not isinstance(
                s_arg.listdef.listitem.s_value,
                (SomeChar, SomeUnicodeCodePoint, SomeImpossibleValue)):
            raise NotAListOfChars
        return s_arg

    def specialize_call(self, hop):
        hop.exception_cannot_occur()
        return hop.inputarg(hop.args_r[0], arg=0)


def attach_gdb():
    import pdb; pdb.set_trace()

if not sys.platform.startswith('win'):
    if sys.platform.startswith('linux'):
        # Only necessary on Linux
        eci = ExternalCompilationInfo(includes=['string.h', 'assert.h',
                                                'sys/prctl.h'],
                                        post_include_bits=["""
/* If we have an old Linux kernel (or compile with old system headers),
   the following two macros are not defined.  But we would still like
   a pypy translated on such a system to run on a more modern system. */
#ifndef PR_SET_PTRACER
#  define PR_SET_PTRACER 0x59616d61
#endif
#ifndef PR_SET_PTRACER_ANY
#  define PR_SET_PTRACER_ANY ((unsigned long)-1)
#endif
static void pypy__allow_attach(void) {
    prctl(PR_SET_PTRACER, PR_SET_PTRACER_ANY);
}
"""])
        allow_attach = rffi.llexternal(
            "pypy__allow_attach", [], lltype.Void,
            compilation_info=eci, _nowrapper=True)
    else:
        # Do nothing, there's no prctl
        def allow_attach():
            pass

    def impl_attach_gdb():
        import os
        allow_attach()
        pid = os.getpid()
        gdbpid = os.fork()
        if gdbpid == 0:
            shell = os.environ.get("SHELL") or "/bin/sh"
            sepidx = shell.rfind(os.sep) + 1
            if sepidx > 0:
                argv0 = shell[sepidx:]
            else:
                argv0 = shell
            try:
                os.execv(shell, [argv0, "-c", "gdb -p %d" % pid])
            except OSError as e:
                os.write(2, "Could not start GDB: %s" % (
                    os.strerror(e.errno)))
                os._exit(1)
        else:
            time.sleep(1)  # give the GDB time to attach

else:
    def make_vs_attach_eci():
        # The COM interface to the Debugger has to be compiled as a .cpp file by
        # Visual C. So we generate the source and then add a commandline switch
        # to treat this source file as C++
        import os
        eci = ExternalCompilationInfo(post_include_bits=["""
#ifdef __cplusplus
extern "C" {
#endif
RPY_EXPORTED void AttachToVS();
#ifdef __cplusplus
}
#endif
                                      """],
                                      separate_module_sources=["""
#import "libid:80cc9f66-e7d8-4ddd-85b6-d9e6cd0e93e2" version("8.0") lcid("0") raw_interfaces_only named_guids
extern "C" RPY_EXPORTED void AttachToVS() {
    CoInitialize(0);
    HRESULT hr;
    CLSID Clsid;

    CLSIDFromProgID(L"VisualStudio.DTE", &Clsid);
    IUnknown *Unknown;
    if (FAILED(GetActiveObject(Clsid, 0, &Unknown))) {
        puts("Could not attach to Visual Studio (is it not running?");
        return;
    }

    EnvDTE::_DTE *Interface;
    hr = Unknown->QueryInterface(&Interface);
    if (FAILED(GetActiveObject(Clsid, 0, &Unknown))) {
        puts("Could not open COM interface to Visual Studio (no permissions?)");
        return;
    }

    EnvDTE::Debugger *Debugger;
    puts("Waiting for Visual Studio Debugger to become idle");
    while (FAILED(Interface->get_Debugger(&Debugger)));

    EnvDTE::Processes *Processes;
    while (FAILED(Debugger->get_LocalProcesses(&Processes)));

    long Count = 0;
    if (FAILED(Processes->get_Count(&Count))) {
        puts("Cannot query Process count");
    }

    for (int i = 0; i <= Count; i++) {
        EnvDTE::Process *Process;
        if (FAILED(Processes->Item(variant_t(i), &Process))) {
            continue;
        }

        long ProcessID;
        while (FAILED(Process->get_ProcessID(&ProcessID)));

        if (ProcessID == GetProcessId(GetCurrentProcess())) {
            printf("Found process ID %d\\n", ProcessID);
            Process->Attach();
            Debugger->Break(false);
            CoUninitialize();
            return;
        }
    }
}
                                      """]
        )
        eci = eci.convert_sources_to_files()
        d = eci._copy_attributes()
        cfile = d['separate_module_files'][0]
        cppfile = cfile.replace(".c", "_vsdebug.cpp")
        os.rename(cfile, cppfile)
        d['separate_module_files'] = [cppfile]
        return ExternalCompilationInfo(**d)

    ll_attach = rffi.llexternal("AttachToVS", [], lltype.Void,
                                compilation_info=make_vs_attach_eci())
    def impl_attach_gdb():
        #ll_attach()
        print "AttachToVS is disabled at the moment (compilation failure)"

register_external(attach_gdb, [], result=None,
                  export_name="impl_attach_gdb", llimpl=impl_attach_gdb)