def test_stop_track(self):
        tracemalloc.start()
        tracemalloc.stop()

        with self.assertRaises(RuntimeError):
            self.track()
        self.assertIsNone(self.get_traceback())

    def test_get_traces_intern_traceback(self):
        # dummy wrappers to get more useful and identical frames in the traceback
        def allocate_bytes2(size):
            return allocate_bytes(size)

        def allocate_bytes3(size):
            return allocate_bytes2(size)

        def allocate_bytes4(size):
            return allocate_bytes3(size)

        # Ensure that two identical tracebacks are not duplicated
        tracemalloc.stop()
        tracemalloc.start(4)
        obj_size = 123
        obj1, obj1_traceback = allocate_bytes4(obj_size)
        obj2, obj2_traceback = allocate_bytes4(obj_size)

        traces = tracemalloc._get_traces()

        trace1 = self.find_trace(traces, obj1_traceback)
        trace2 = self.find_trace(traces, obj2_traceback)
        size1, traceback1 = trace1
        size2, traceback2 = trace2
        self.assertEqual(traceback2, traceback1)
        self.assertIs(traceback2, traceback1)

    def test_get_traced_memory(self):
        # Python allocates some internals objects, so the test must tolerate
        # a small difference between the expected size and the real usage
        max_error = 2048

        # allocate one object
        obj_size = 1024 * 1024
        tracemalloc.clear_traces()
        obj, obj_traceback = allocate_bytes(obj_size)
        size, peak_size = tracemalloc.get_traced_memory()
        self.assertGreaterEqual(size, obj_size)
        self.assertGreaterEqual(peak_size, size)

        self.assertLessEqual(size - obj_size, max_error)
        self.assertLessEqual(peak_size - size, max_error)

        # destroy the object
        obj = None
        size2, peak_size2 = tracemalloc.get_traced_memory()
        self.assertLess(size2, size)
        self.assertGreaterEqual(size - size2, obj_size - max_error)
        self.assertGreaterEqual(peak_size2, peak_size)

        # clear_traces() must reset traced memory counters
        tracemalloc.clear_traces()
        self.assertEqual(tracemalloc.get_traced_memory(), (0, 0))

        # allocate another object
        obj, obj_traceback = allocate_bytes(obj_size)
        size, peak_size = tracemalloc.get_traced_memory()
        self.assertGreaterEqual(size, obj_size)

        # stop() also resets traced memory counters
        tracemalloc.stop()
        self.assertEqual(tracemalloc.get_traced_memory(), (0, 0))

    def test_stop_untrack(self):
        tracemalloc.start()
        self.track()

        tracemalloc.stop()
        with self.assertRaises(RuntimeError):
            self.untrack()

def stop():
    """ Stops application memory profiling """
    logging.debug("Stopping memory profiling")
    with _lock:
        if is_running():
            snapshot(_make_snapshot_name)
            tracemalloc.clear_traces()
            tracemalloc.stop()

def _stop_memory_tracing():
    try:
        import tracemalloc
    except ImportError:
        return

    snapshot = tracemalloc.take_snapshot()

    _log_memory_top(snapshot)

    tracemalloc.stop()

    def stop(self):
        if not self.profiling:
            return

        self.profiling = False

        tracemalloc.stop()
        self.timer.stop()
        self.trace_stream.close()
        self.trace_stream = None
        self.timer = None

 def measure_memory_diff(self, func):
     import tracemalloc
     tracemalloc.start()
     try:
         before = tracemalloc.take_snapshot()
         # Keep the result and only delete it after taking a snapshot
         res = func()
         after = tracemalloc.take_snapshot()
         del res
         return after.compare_to(before, 'lineno')
     finally:
         tracemalloc.stop()

 def exec_with_profiler(filename, profiler, backend):
     choose_backend(backend)
     if _backend == 'tracemalloc' and has_tracemalloc:
         tracemalloc.start()
     builtins.__dict__['profile'] = profiler
     # shadow the profile decorator defined above
     ns = dict(_CLEAN_GLOBALS, profile=profiler)
     try:
         with open(filename) as f:
             exec(compile(f.read(), filename, 'exec'), ns, ns)
     finally:
         if has_tracemalloc and tracemalloc.is_tracing():
             tracemalloc.stop()

    def stop_profiler(self):
        self.agent.log('Deactivating memory allocation profiler.')

        with self.profile_lock:
            if self.overhead_monitor:
                self.overhead_monitor.cancel()
                self.overhead_monitor = None

            if tracemalloc.is_tracing():
                snapshot = tracemalloc.take_snapshot()
                self.agent.log('Allocation profiler memory overhead {0} bytes'.format(tracemalloc.get_tracemalloc_memory()))
                tracemalloc.stop()
                self.process_snapshot(snapshot, time.time() - self.start_ts)

def exec_with_profiler(filename, profiler, backend, passed_args=[]):
    from runpy import run_module
    builtins.__dict__['profile'] = profiler
    ns = dict(_CLEAN_GLOBALS, profile=profiler)
    _backend = choose_backend(backend)
    sys.argv = [filename] + passed_args
    try:
        if _backend == 'tracemalloc' and has_tracemalloc:
            tracemalloc.start()
        with open(filename) as f:
            exec(compile(f.read(), filename, 'exec'), ns, ns)
    finally:
        if has_tracemalloc and tracemalloc.is_tracing():
            tracemalloc.stop()

    def test_get_traces(self):
        tracemalloc.clear_traces()
        obj_size = 12345
        obj, obj_traceback = allocate_bytes(obj_size)

        traces = tracemalloc._get_traces()
        trace = self.find_trace(traces, obj_traceback)

        self.assertIsInstance(trace, tuple)
        domain, size, traceback = trace
        self.assertEqual(size, obj_size)
        self.assertEqual(traceback, obj_traceback._frames)

        tracemalloc.stop()
        self.assertEqual(tracemalloc._get_traces(), [])

def run_module_with_profiler(module, profiler, backend, passed_args=[]):
    from runpy import run_module
    builtins.__dict__['profile'] = profiler
    ns = dict(_CLEAN_GLOBALS, profile=profiler)
    _backend = choose_backend(backend)
    sys.argv = [module] + passed_args
    if PY2:
        run_module(module, run_name="__main__", init_globals=ns)
    else:
        if _backend == 'tracemalloc' and has_tracemalloc:
            tracemalloc.start()
        try:
            run_module(module, run_name="__main__", init_globals=ns)
        finally:
            if has_tracemalloc and tracemalloc.is_tracing():
                tracemalloc.stop()

 def test_does_not_leak_too_much(self):
     tracemalloc.start()
     gc.collect()
     series = []
     snapshot1 = tracemalloc.take_snapshot()
     for i in range(100):
         try:
             execute_script(self.feature, self)
         except Exception:
             pass
         gc.collect()
         snapshot2 = tracemalloc.take_snapshot()
         stats = snapshot2.compare_to(snapshot1, "lineno")
         snapshot1 = snapshot2
         series.append(sum(stat.size / 1024 for stat in stats))
     tracemalloc.stop()
     series = series[1:]  # ignore first run, which creates regex
     cv = statistics.stdev(series) / statistics.mean(series)
     assert cv < 0.1

    def test_snapshot(self):
        obj, source = allocate_bytes(123)

        # take a snapshot
        snapshot = tracemalloc.take_snapshot()

        # write on disk
        snapshot.dump(support.TESTFN)
        self.addCleanup(support.unlink, support.TESTFN)

        # load from disk
        snapshot2 = tracemalloc.Snapshot.load(support.TESTFN)
        self.assertEqual(snapshot2.traces, snapshot.traces)

        # tracemalloc must be tracing memory allocations to take a snapshot
        tracemalloc.stop()
        with self.assertRaises(RuntimeError) as cm:
            tracemalloc.take_snapshot()
        self.assertEqual(str(cm.exception),
                         "the tracemalloc module must be tracing memory "
                         "allocations to take a snapshot")

    def compute(self):
        args = self.args

        if args.track_memory:
            if MS_WINDOWS:
                from perf._win_memory import get_peak_pagefile_usage
            else:
                from perf._memory import PeakMemoryUsageThread
                mem_thread = PeakMemoryUsageThread()
                mem_thread.start()

        if args.tracemalloc:
            import tracemalloc
            tracemalloc.start()

        WorkerTask.compute(self)

        if args.tracemalloc:
            traced_peak = tracemalloc.get_traced_memory()[1]
            tracemalloc.stop()

            if not traced_peak:
                raise RuntimeError("tracemalloc didn't trace any Python "
                                   "memory allocation")

            # drop timings, replace them with the memory peak
            self._set_memory_value(traced_peak)

        if args.track_memory:
            if MS_WINDOWS:
                mem_peak = get_peak_pagefile_usage()
            else:
                mem_thread.stop()
                mem_peak = mem_thread.peak_usage

            if not mem_peak:
                raise RuntimeError("failed to get the memory peak usage")

            # drop timings, replace them with the memory peak
            self._set_memory_value(mem_peak)

    def pyfaidx_bgzf_faidx(n):
        print('timings for pyfaidx.Faidx with bgzf compression')
        ti = []
        tf = []
        for _ in range(n):
            t = time.time()
            f = pyfaidx.Faidx(fa_file.name + '.gz')
            ti.append(time.time() - t)

            t = time.time()
            read_faidx(f, headers)
            tf.append(time.time() - t)
            os.remove(index)
        # profile memory usage and report timings
        tracemalloc.start()
        f = pyfaidx.Faidx(fa_file.name + '.gz')
        read_faidx(f, headers)
        os.remove(index)
        print(tracemalloc.get_traced_memory())
        print(mean(ti))
        print(mean(tf)/nreads/10*1000*1000)
        tracemalloc.stop()

def bench(func, trace=True, nframe=1):
    if trace:
        tracemalloc.stop()
        tracemalloc.start(nframe)
    gc.collect()
    best = None
    for run in range(BENCH_RUNS):
        start = time.monotonic()
        func()
        dt = time.monotonic() - start
        if best is not None:
            best = min(best, dt)
        else:
            best = dt
    if trace:
        mem = tracemalloc.get_tracemalloc_memory()
        ntrace = len(tracemalloc.take_snapshot().traces)
        tracemalloc.stop()
    else:
        mem = ntrace = None
    gc.collect()
    return best * 1e3, mem, ntrace

    def pyfaidx_fasta(n):
        print('timings for pyfaidx.Fasta')
        ti = []
        tf = []
        for _ in range(n):
            t = time.time()
            f = pyfaidx.Fasta(fa_file.name)
            ti.append(time.time() - t)

            t = time.time()
            read_dict(f, headers)
            tf.append(time.time() - t)
            os.remove(index)
        # profile memory usage and report timings
        tracemalloc.start()
        f = pyfaidx.Fasta(fa_file.name)
        read_dict(f, headers)
        os.remove(index)
        print(tracemalloc.get_traced_memory())
        print(mean(ti))
        print(mean(tf)/nreads/10*1000*1000)
        tracemalloc.stop()

def show(filename=None):
    global before_objects
    gc.collect()
    after_objects = gc.get_objects()
    frame = sys._getframe()
    globals_ = globals()
    num_leaks = 0
    before_id_set = set(map(id, before_objects))
    if filename is not None:
        f = open(filename, 'w')
    else:
        f = sys.stderr

    for obj in after_objects:
        if id(obj) not in before_id_set:
            if obj in (before_objects, frame):
                continue
            num_leaks += 1
            print("Leak: id=0x{:x} type={} {}".format(id(obj), type(obj), _get_detail(obj)), file=f)
            tb = tracemalloc.get_object_traceback(obj)
            if tb is None:
                print("Traceback: None", file=f)
            else:
                print("Traceback:", file=f)
                print("\n".join(tb.format(MAX_FRAMES)), file=f)
            print(file=f)
            print("Referrers:", file=f)
            for ref in gc.get_referrers(obj):
                if ref in (after_objects, before_objects, frame, globals_):
                    continue
                print("     id=0x{:x} type={} {}".format(id(ref), type(ref), _get_detail(ref)), file=f)
                print("     traceback: {}".format(tracemalloc.get_object_traceback(ref)), file=f)
            print(file=f)
            print(file=f)
    print("Total leaks: {}".format(num_leaks), file=f)
    if filename is not None:
        f.close()
    tracemalloc.stop()
    gc.enable()