Generating single-frame stacktraces

orphan.py

import time as time_module
import pprint
import collections
import traceback as traceback_module
import itertools
import threading
import sys
import inspect


in_gen_lock = threading.Lock()
out_gen_lock = threading.Lock()


def generator():
    in_gen_lock.acquire()
    while True:
        in_gen_lock.release()
        time_module.sleep(0.01)
        in_gen_lock.acquire()
        yield
        

def get_stacktraces(frames):
    return [''.join(traceback_module.format_stack(frame)) for frame in frames]
    
        
class ActionThread(threading.Thread):
    
    _exit = False
    
    def exit(self):
        self._exit = True
        self.join()
        
    def run(self):
        gen = generator()
        while True:
            out_gen_lock.acquire()
            next(gen)
            out_gen_lock.release()
            time_module.sleep(0.01)
            if self._exit:
                return
        
        
class InspectionThread(threading.Thread):
        
    def run(self):
        stacktrace_counter = collections.Counter()
        for _ in range(100):
            with in_gen_lock:
                frames = sys._current_frames().values()
            with out_gen_lock:
                for stacktrace in get_stacktraces(frames):
                    if stacktrace not in stacktrace_counter:
                        print(stacktrace)
                    stacktrace_counter[stacktrace] += 1
            # time_module.sleep(0)
        pprint.pprint(tuple(stacktrace_counter.values()))
        
        
inspection_thread = InspectionThread()
inspection_thread.start()
action_thread = ActionThread()
action_thread.start()
        

inspection_thread.join()
action_thread.exit()