from queue import Queue, Empty
from threading import Thread
from multiprocessing import Process
import time


class Streamer:
    def __init__ (self, _text_queue):
        self.text_queue = _text_queue
        self.stop_signal = "stop"
    
    def put(self, value):
        self.text_queue.put(value, timeout=0.5)
    
    def __iter__(self):
        return self
    
    def __next__(self):
        try:
            value = self.text_queue.get(timeout=0.5)
        except Empty as empty:
            value = self.stop_signal
        if value == self.stop_signal:
            print("stop here!")
            raise StopIteration()
        return value

import ctypes
import inspect

def async_raise(tid, exctype):
   """raises the exception, performs cleanup if needed"""
   tid = ctypes.c_long(tid)
   if not inspect.isclass(exctype):
      exctype = type(exctype)
   res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
   if res == 0:
      raise ValueError("invalid thread id")
   elif res != 1:
      # """if it returns a number greater than one, you're in trouble,
      # and you should call it again with exc=NULL to revert the effect"""
      ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
      raise SystemError("PyThreadState_SetAsyncExc failed")

def stop_thread(thread):
   async_raise(thread.ident, SystemExit)

import sys
import trace
import threading
import time
class thread_with_trace(threading.Thread):
    def __init__(self, *args, **keywords):
        threading.Thread.__init__(self, *args, **keywords)
        self.killed = False

    def start(self):
        self.__run_backup = self.run
        self.run = self.__run
        threading.Thread.start(self)

    def __run(self):
        sys.settrace(self.globaltrace)
        self.__run_backup()
        self.run = self.__run_backup

    def globaltrace(self, frame, event, arg):
        if event == 'call':
            return self.localtrace
        else:
            return None

    def localtrace(self, frame, event, arg):
        if self.killed:
            if event == 'line':
                raise SystemExit()
        return self.localtrace

    def kill(self):
        self.killed = True


def test_func():
    def _inference():
        count = 0
        while True:
            if count < 10:
                streamer.put("shit")
                time.sleep(20) # 例如卡在so一直卡着
                print("put shit & getting gem")
                time.sleep(0.2 * count) #后面推理超时
                count += 1
            else:
                print("breaking!")
                break
    t_queue = Queue()
    streamer = Streamer(t_queue)
    
    # 模式1 Thread daemon
    # inference_thread = Thread(target=_inference, daemon=True)
    # inference_thread.start()
    # for idx, i in enumerate(streamer):
    #     print(f"get {i}{idx}")
    
    # 模式2 Thread+stop
    # https://www.cnblogs.com/conscience-remain/p/16930488.html
    # stop_thread(inference_thread) # 54行会导致停止失败

    # 模式3 ThreadPoolExecutor的cancel
    # import concurrent.futures
    # with concurrent.futures.ThreadPoolExecutor(max_workers=1) as tpe:
    #     future = tpe.submit(_inference)
    #     for idx, i in enumerate(streamer):
    #         print(i, idx)
    #     # getting jammed or finished
    #     if future.running():
    #         print("canceling here!")
    #         future.cancel() # 取消线程？如果正在运行的，并不会生效
    #     try:
    #         future.result(timeout=1)
    #     except concurrent.futures.TimeoutError as ex:
    #         print(f"ex: {ex}")

    # 模式4 trace 主线程都退出了，自线程还gam
    thread = thread_with_trace(target=_inference, daemon=True)
    thread.start()
    for idx, i in enumerate(streamer):
        print(i, idx)
    thread.kill()

def worker():
    worker = Thread(target=test_func)
    worker.start()
    worker.join()

if __name__ == "__main__":
    worker()
    # time.sleep(5) # 主线程如果不退出，daemon也不会退出
    print("out")