Я хочу реализовать графический интерфейс, в котором я запускаю long_running_task с помощью кнопки запуска. Пока задача выполняется, я анимирую статус в графическом интерфейсе. Вот почему я использую многопоточность и создаю поток для анимации и long_running_task в start_task(). Теперь я хочу иметь возможность прервать выполнение long_running_task во время его выполнения. Поэтому моя кнопка запуска переключается на кнопку отмены (update_button()), которая обновляет переменную состояния self.running, щелкнув ее.
Теперь я проверяю эту переменную статуса на каждой итерации в long_running_task, которая представляет собой всего лишь os.walk() для вывода списка всех файлов в очень большом каталоге.
Если в моем классе определена функция long_running_task(), кнопка отмены работает отлично (просто раскомментируйте все закомментированные строки кода, чтобы попробовать это самостоятельно).
Моя проблема: на самом деле моя long_running_task содержит много строк кода. Вот почему я хочу сохранить его в отдельном скрипте Python. Поэтому я ввел и связал функцию back_end() с кнопкой старт. Он вызывает функцию long_running_task из другого скрипта, передавая ей переменную статуса. Это должно работать, потому что если вы передаете переменную, вы ссылаетесь только на ячейку памяти, поэтому переменная также всегда имеет текущее значение в функции. Тем не менее, если я вызову функцию таким образом, моя кнопка отмены больше не остановит long_running_task.
Я работаю над средой Windows.
Мой код:
import os
import tkinter as tk
from tkinter import ttk
import threading
import itertools
import time
from other_python_script import long_running_task
class APP():
def __init__(self, root):
self.root = root
self.root.title("Task Manager")
self.running = False
self.task_thread = None
self.anim_thread = None
self.status_var = tk.StringVar(value = "Ready")
self.directory = "Any directory with great depth"
self.status_label = ttk.Label(self.root, textvariable=self.status_var)
self.status_label.pack(padx=10, pady=10)
self.start_abort_button = ttk.Button(self.root, text = "Start", command=self.start_task)
self.start_abort_button.pack(padx=10, pady=5)
# def long_running_task(self):
# self.status_var.set("Running...")
# for root, dirs, files in os.walk(self.directory):
# if not self.running:
# break
# else:
# self.file_list.append(files)
# print(self.file_list)
# if self.running:
# self.status_var.set("Finished")
# else:
# self.status_var.set("Aborted")
# self.update_button()
def back_end(self):
self.status_var.set("Running...")
file_list = long_running_task(self.running, self.directory)
if self.running:
self.status_var.set("Finished")
else:
self.status_var.set("Aborted")
print("FINISHED with total List: ", file_list)
self.update_button()
def start_task(self):
print(self.task_thread)
if self.task_thread is None or not self.task_thread.is_alive():
self.running = True
self.status_var.set("Running...")
self.task_thread = threading.Thread(target=self.back_end, daemon=True)
#self.task_thread = threading.Thread(target=self.long_running_task, daemon=True)
self.task_thread.start()
self.anim_thread = threading.Thread(target=self.animate_status, daemon=True)
self.anim_thread.start()
self.update_button()
def abort_task(self):
self.running = False
self.file_list = []
self.update_button()
def animate_status(self):
while self.running:
for status in itertools.cycle(["Running.", "Running..", "Running..."]):
if not self.running:
self.status_var.set("Aborted")
return
self.status_var.set(status)
time.sleep(0.5)
def update_button(self):
if self.running:
self.start_abort_button.configure(text = "Abort", command=self.abort_task)
else:
self.start_abort_button.configure(text = "Start", command=self.start_task)
if __name__ == '__main__':
root = tk.Tk()
APP(root)
root.mainloop()
import os
def long_running_task(running, directory):
file_list = []
for root, dirs, files in os.walk(directory):
if not running:
return[]
else:
file_list.append(files)
print(file_list)
return file_list
Попробуйте превратить running
в функцию:
Основной:
def back_end(self):
self.status_var.set("Running...")
file_list = long_running_task((lambda: self.running), self.directory)
if self.running:
self.status_var.set("Finished")
else:
self.status_var.set("Aborted")
print("FINISHED with total List: ", file_list)
self.update_button()
Другой файл:
def long_running_task(running, directory):
file_list = []
for root, dirs, files in os.walk(directory):
if not running():
return[]
else:
file_list.append(files)
print(file_list)
return file_list
Ответ на основе информации здесь: https://www.geeksforgeeks.org/python- Different-ways-to-kill-a-thread/
На всякий случай я запустил ваш код на своем Mac (Python 3.11.1) с обоими вариантами и получил одинаковые результаты и визуальное восприятие.
<__main__.APP object at 0x106b4f210>
None
<Thread(Thread-1 (long_running_task), stopped daemon 123145515970560)>
<Thread(Thread-3 (long_running_task), stopped daemon 123145515970560)>
<__main__.APP object at 0x102523350>
None
<Thread(Thread-1 (back_end), stopped daemon 123145436504064)>
<Thread(Thread-3 (back_end), stopped daemon 123145436504064)>
Мониторинг потоков процесса во время выполнения. В любом случае поток фактически завершается кнопкой отмены.
Ваше утверждение «если вы передаете переменную, вы ссылаетесь только на ячейку памяти» верно, и это источник проблемы. Когда Python выполняется
file_list = long_running_task(self.running, self.directory)
он преобразует self.running
в значение (True
в вашем случае) и передает анонимную ссылку на это значение в функцию. Функция привязывает значение локальной переменной running
. self.running
и running
ссылаются на один и тот же сингл True
— это ячейка памяти, о которой вы говорите. Функция ссылается на значение, но понятия не имеет, откуда оно взялось.
Если self.value
переназначено, теперь оно ссылается на другое значение в другом месте памяти. running
по-прежнему ссылается на оригинал True
, потому что все, что он когда-либо получал, — это ссылка на это место в памяти.
Существует несколько способов решения этой проблемы, но самый простой — заставить функцию ссылаться на пространство имен, содержащее «работающую» переменную. Если функция ищет переменную в пространстве имен, она получит ее текущее значение, а не то значение, которое она получила при первом создании потока. Вы можете сделать это с помощью переменной self
. Он ссылается на пространство имен экземпляра приложения, которое вам нужно в качестве «работающей» переменной.
main.py
import os
import tkinter as tk
from tkinter import ttk
import threading
import itertools
import time
from other_python_script import long_running_task
class APP():
def __init__(self, root):
self.root = root
self.root.title("Task Manager")
self.running = False
self.task_thread = None
self.anim_thread = None
self.status_var = tk.StringVar(value = "Ready")
self.directory = "Any directory with great depth"
self.status_label = ttk.Label(self.root, textvariable=self.status_var)
self.status_label.pack(padx=10, pady=10)
self.start_abort_button = ttk.Button(self.root, text = "Start", command=self.start_task)
self.start_abort_button.pack(padx=10, pady=5)
# def long_running_task(self):
# self.status_var.set("Running...")
# for root, dirs, files in os.walk(self.directory):
# if not self.running:
# break
# else:
# self.file_list.append(files)
# print(self.file_list)
# if self.running:
# self.status_var.set("Finished")
# else:
# self.status_var.set("Aborted")
# self.update_button()
def back_end(self):
self.status_var.set("Running...")
file_list = long_running_task(self, self.directory) # changed
if self.running:
self.status_var.set("Finished")
else:
self.status_var.set("Aborted")
print("FINISHED with total List: ", file_list)
self.update_button()
def start_task(self):
print(self.task_thread)
if self.task_thread is None or not self.task_thread.is_alive():
self.running = True
self.status_var.set("Running...")
self.task_thread = threading.Thread(target=self.back_end, daemon=True)
#self.task_thread = threading.Thread(target=self.long_running_task, daemon=True)
self.task_thread.start()
self.anim_thread = threading.Thread(target=self.animate_status, daemon=True)
self.anim_thread.start()
self.update_button()
def abort_task(self):
self.running = False
self.file_list = []
self.update_button()
def animate_status(self):
while self.running:
for status in itertools.cycle(["Running.", "Running..", "Running..."]):
if not self.running:
self.status_var.set("Aborted")
return
self.status_var.set(status)
time.sleep(0.5)
def update_button(self):
if self.running:
self.start_abort_button.configure(text = "Abort", command=self.abort_task)
else:
self.start_abort_button.configure(text = "Start", command=self.start_task)
if __name__ == '__main__':
root = tk.Tk()
APP(root)
root.mainloop()
другой_python_script.py
import os
def long_running_task(app, directory): # changed
file_list = []
for root, dirs, files in os.walk(directory):
if not app.running: # changed
return[]
else:
file_list.append(files)
print(file_list)
return file_list
Насильственное уничтожение тредов всегда является «Плохой Идеей»™️. Лучше отправить потоку сообщение или установить событие, которое заставит его решить выйти по собственному желанию.