180 lines
6.4 KiB
Python
180 lines
6.4 KiB
Python
"""
|
|
This is the main printing thread. A worker thread consums Tasks from
|
|
a PrintQueue, while trying to find available printers.
|
|
"""
|
|
|
|
import threading
|
|
import time
|
|
import traceback
|
|
from printers import Printers
|
|
|
|
|
|
class PrintWorker(threading.Thread):
|
|
"""
|
|
A thread used to consume Tasks added to a Print Queue.
|
|
|
|
On initialisation, the worker will try to find Printers,
|
|
and on each print, if the choosen Printer becomes unavailable, it
|
|
will choose an available Printer from a list of Printers.
|
|
|
|
If a print fails, it's not retried, but will be added to a list of completed
|
|
tasks. TODO : Change that
|
|
"""
|
|
|
|
def __init__(self, app, print_queue, printers_pool):
|
|
super().__init__(daemon=True)
|
|
self.app = app
|
|
self.print_queue = print_queue
|
|
self.printer = None
|
|
self._lock = threading.Lock()
|
|
self.running = True
|
|
self.state = "idle" # idle, printing, dead, drinking-a-beer
|
|
|
|
# All workers share the same pool of printers.
|
|
self.printers = printers_pool
|
|
self.printers_obj = self.printers.printers
|
|
self._printer_index = 0
|
|
|
|
self.app.logger.debug("Worker initialized.")
|
|
self.app.logger.debug("Printers connected:")
|
|
for printer in self.printers_obj:
|
|
self.app.logger.debug("Printer type: %s", printer.printer_type)
|
|
self.app.logger.debug("Printer state: %s", printer.ready)
|
|
|
|
self.app.logger.debug("Ho great, I'm alive... I'm ready to work another day...")
|
|
|
|
def _acquire_available_printer(self):
|
|
result = self.printers.acquire_available_printer(self._printer_index)
|
|
if result is None:
|
|
return None
|
|
|
|
index, printer = result
|
|
self._printer_index = (index + 1) % len(self.printers_obj)
|
|
return printer
|
|
|
|
def run(self):
|
|
"""Background thread that processes queue items"""
|
|
self.app.logger.debug("Worker %s started working.", threading.get_ident())
|
|
self.app.logger.debug("Current threads : %s", threading.active_count())
|
|
self.app.logger.debug("Threads actives : %s ", threading.enumerate())
|
|
|
|
while True:
|
|
|
|
# If the printer is dead or asleep, it can't work.
|
|
if not self.running:
|
|
time.sleep(0.2)
|
|
continue
|
|
|
|
# If we have no available printer, we look at the list printers
|
|
# we know about, and try to find one that is available.
|
|
# When we find a printer, we acquire it
|
|
# When we are finished with a printer, we release it to the world.
|
|
|
|
# If we do not have a printer, we try to find one
|
|
|
|
while self.running and self.printer is None:
|
|
self.app.logger.debug(
|
|
"Trying to find a available printer for the worker."
|
|
)
|
|
self.printer = self._acquire_available_printer()
|
|
self.app.logger.debug(
|
|
"The worker got a %s printer and it's %s",
|
|
self.printer.printer_type,
|
|
"Ready" if self.printer.ready else "Not ready",
|
|
)
|
|
|
|
if self.printer is None:
|
|
time.sleep(1)
|
|
|
|
self.printer.use_printer()
|
|
|
|
# If the printer is not ready, we wait for it to become ready.
|
|
while not self.printer.ready:
|
|
self.app.debug("Waiting for our printer to become ready...")
|
|
time.sleep(1)
|
|
|
|
if self.state != "idle":
|
|
self.app.logger("We are not idle, waiting...")
|
|
time.sleep(1)
|
|
continue
|
|
|
|
self.state = "printing"
|
|
|
|
with self._lock:
|
|
try:
|
|
task = self.print_queue.dequeue()
|
|
except Exception as e:
|
|
self.app.logger.error("Could not get a new task ! %s ", str(e))
|
|
self.state = "idle"
|
|
self.printer.free_printer()
|
|
raise RuntimeError(
|
|
"We could not get a new task because " + str(e)
|
|
) from e
|
|
|
|
if task:
|
|
try:
|
|
self.app.logger.info("Got a new task")
|
|
self.app.logger.debug(
|
|
"Got task %s %s", task.task_type, task.task_id
|
|
)
|
|
task.status = "processing"
|
|
|
|
print_data = task.get_print_data()
|
|
|
|
try:
|
|
self.printer.print_task(task.task_type, print_data)
|
|
except RuntimeError as e:
|
|
self.state = "idle"
|
|
self.app.logger.error("Could not print : %s", str(e))
|
|
self.app.logger.error(traceback.format_exc())
|
|
self.printer.free_printer()
|
|
raise e
|
|
|
|
task.status = "completed"
|
|
self.print_queue.mark_completed(task.task_id, "completed")
|
|
self.app.logger.debug(
|
|
"Finished printing task %s ", task.task_id
|
|
)
|
|
self.state = "idle"
|
|
|
|
except RuntimeError as e:
|
|
task.status = "failed"
|
|
self.state = "idle"
|
|
self.print_queue.mark_completed(task.task_id, "failed")
|
|
self.app.logger.error(
|
|
"Could not print task %s because %s ", task.task_id, str(e)
|
|
)
|
|
|
|
else:
|
|
# When they are no new tasks to handle, we put the thread to sleep.
|
|
self.state = "idle"
|
|
time.sleep(0.1)
|
|
|
|
def stop_worker(self):
|
|
"""
|
|
Give the worker a break
|
|
"""
|
|
self.app.logger.debug("Giving the worker a break")
|
|
self.state = "drinking-a-beer"
|
|
self.printer.free_printer()
|
|
self.running = False
|
|
|
|
def start_worker(self):
|
|
"""
|
|
Get the worker back to it
|
|
"""
|
|
self.app.logger.debug("Time to work !")
|
|
self.state = "idle"
|
|
self.running = True
|
|
|
|
def current_state(self):
|
|
"""
|
|
Return the worker state
|
|
"""
|
|
return {
|
|
"is_running": self.running,
|
|
"queue_size": len(self.print_queue),
|
|
"state": self.state,
|
|
"printers": len(self.printers_obj),
|
|
}
|