jumeaux-numeriques/twin_threading.py

121 lines
3.4 KiB
Python

import bge # Bibliothèque Blender Game Engine (UPBGE)
import threading # Multithreading
import trace
import sys
import time
###############################################################################
# twin_threading.py
# @title: Gestion des tâches (threads)
# @project: Blender-EduTech
# @lang: fr
# @authors: Philippe Roy <philippe.roy@ac-grenoble.fr>
# @copyright: Copyright (C) 2020-2022 Philippe Roy
# @license: GNU GPL
###############################################################################
scene = bge.logic.getCurrentScene()
# Threads
threads_cmd=[]
debug_thread = scene.objects['System']['debug_thread']
###############################################################################
# Méthode kill pour les tâches (threads)
###############################################################################
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
###############################################################################
# Start et stop des tâches (threads)
###############################################################################
def thread_start(threads, type_txt, fct):
threads.append(thread_with_trace(target = fct))
threads[len(threads)-1].start()
if (debug_thread):
print ("Thread",type_txt, "#", len(threads)-1, "open.")
def thread_stop(threads, type_txt):
i=0
zombie_flag=False
for t in threads:
if not t.is_alive():
if (debug_thread):
print ("Thread",type_txt, "#",i,"closed.")
else:
if (debug_thread):
print ("Thread",type_txt, "#",i,"still open ...")
t.kill()
t.join()
if not t.is_alive():
if (debug_thread):
print ("Thread",type_txt, "#",i,"killed.")
else:
if (debug_thread):
print ("Thread",type_txt, "#",i,"zombie...")
zombie_flag=True
i +=1
if zombie_flag==False:
if (debug_thread):
print ("All threads",type_txt, "are closed.")
scene.objects['System']['thread_cmd']=False
return True
else:
if (debug_thread):
print ("There are zombies threads",type_txt, ".")
return False
def thread_cmd_start(fct):
thread_start(threads_cmd, "commands", fct)
def thread_cmd_stop():
thread_stop(threads_cmd, "commands")
def end():
# Jumeau numérique
# if scene.objects['System']['twins']:
# jumeau_close()
# Thread
if (debug_thread):
print ("Thread commands is arrived.")
time.sleep(0.125)
scene.objects['System']['thread_cmd']=False
time.sleep(0.125)
def fin():
end()
def quit():
end()