fix async
This commit is contained in:
@@ -1,15 +1,14 @@
|
||||
#!.venv/bin/python
|
||||
|
||||
# Import necessary libraries
|
||||
import mysql.connector
|
||||
import logging
|
||||
import importlib
|
||||
import asyncio
|
||||
import os
|
||||
import aiomysql
|
||||
|
||||
# Import custom modules for configuration and database connection
|
||||
from utils.config import loader_load_data as setting
|
||||
from utils.database.connection import connetti_db
|
||||
from utils.database import CSV_RECEIVED
|
||||
|
||||
# Initialize the logger for this module
|
||||
@@ -20,115 +19,91 @@ CSV_PROCESSING_DELAY = 0.1
|
||||
# Tempo di attesa se non ci sono record da elaborare
|
||||
NO_RECORD_SLEEP = 20
|
||||
|
||||
async def worker(worker_id: int, queue: asyncio.Queue, cfg: object) -> None:
|
||||
"""
|
||||
Worker asyncrono che preleva lavori dalla coda e li esegue.
|
||||
async def get_next_csv_atomic(pool, table_name):
|
||||
"""Preleva atomicamente il prossimo CSV da elaborare"""
|
||||
async with pool.acquire() as conn:
|
||||
async with conn.cursor() as cur:
|
||||
# Usa SELECT FOR UPDATE per lock atomico
|
||||
await cur.execute(f"""
|
||||
SELECT id, unit_type, tool_type, unit_name, tool_name
|
||||
FROM {table_name}
|
||||
WHERE locked = 0 AND status = %s
|
||||
ORDER BY id
|
||||
LIMIT 1
|
||||
FOR UPDATE SKIP LOCKED
|
||||
""", (CSV_RECEIVED,))
|
||||
|
||||
Args:
|
||||
worker_id (int): ID univoco del worker.
|
||||
queue (asyncio.Queue): Coda da cui prendere i lavori.
|
||||
cfg (object): Configurazione caricata.
|
||||
"""
|
||||
result = await cur.fetchone()
|
||||
if result:
|
||||
await cur.execute(f"""
|
||||
UPDATE {table_name}
|
||||
SET locked = 1
|
||||
WHERE id = %s
|
||||
""", (result[0],))
|
||||
await conn.commit()
|
||||
|
||||
return result
|
||||
|
||||
async def worker(worker_id: int, cfg: object, pool) -> None:
|
||||
debug_mode = (logging.getLogger().getEffectiveLevel() == logging.DEBUG)
|
||||
logger.info(f"Worker {worker_id} - Avviato")
|
||||
|
||||
while True:
|
||||
try:
|
||||
# Preleva un "lavoro" dalla coda (in questo caso non ci sono parametri)
|
||||
await queue.get()
|
||||
|
||||
logger.info(f"Worker {worker_id} - Inizio elaborazione")
|
||||
|
||||
record, success = await load_csv(cfg, worker_id)
|
||||
record = await get_next_csv_atomic(pool, cfg.dbrectable)
|
||||
|
||||
if not record:
|
||||
logger.debug(f"Worker {worker_id} - Nessun record trovato")
|
||||
await asyncio.sleep(NO_RECORD_SLEEP)
|
||||
if not success:
|
||||
logger.error(f"Worker {worker_id} - Errore durante l'elaborazione")
|
||||
if record:
|
||||
success = await load_csv(record, cfg, worker_id, pool)
|
||||
if not success:
|
||||
logger.error(f"Worker {worker_id} - Errore durante l'elaborazione")
|
||||
await asyncio.sleep(CSV_PROCESSING_DELAY)
|
||||
else:
|
||||
logger.debug(f"Worker {worker_id} - Elaborazione completata correttamente")
|
||||
await asyncio.sleep(CSV_PROCESSING_DELAY)
|
||||
await asyncio.sleep(NO_RECORD_SLEEP)
|
||||
|
||||
# Segnala che il lavoro è completato
|
||||
queue.task_done()
|
||||
except Exception as e:
|
||||
logger.error(f"Worker {worker_id} - Errore durante l'esecuzione: {e}", exc_info=debug_mode)
|
||||
queue.task_done()
|
||||
await asyncio.sleep(1)
|
||||
|
||||
|
||||
async def load_csv(cfg: object, worker_id: int) -> tuple:
|
||||
"""
|
||||
Cerca e carica un file CSV da elaborare dal database.
|
||||
|
||||
Args:
|
||||
cfg (object): Oggetto configurazione contenente dati per DB e altro.
|
||||
|
||||
Returns:
|
||||
bool: True se è stato trovato ed elaborato un record, False altrimenti.
|
||||
"""
|
||||
async def load_csv(record: tuple, cfg: object, worker_id: int, pool) -> bool:
|
||||
|
||||
debug_mode = (logging.getLogger().getEffectiveLevel() == logging.DEBUG)
|
||||
logger.debug(f"Worker {worker_id} - Inizio ricerca nuovo CSV da elaborare")
|
||||
|
||||
try:
|
||||
with connetti_db(cfg) as conn:
|
||||
cur = conn.cursor()
|
||||
logger.debug(f"Worker {worker_id} - Connessione al database stabilita")
|
||||
id, unit_type, tool_type, unit_name, tool_name = record
|
||||
logger.info(f"Worker {worker_id} - Trovato CSV da elaborare: ID={id}, Tipo={unit_type}_{tool_type}, Nome={unit_name}_{tool_name}")
|
||||
|
||||
query = f"""
|
||||
SELECT id, unit_type, tool_type, unit_name, tool_name
|
||||
FROM {cfg.dbname}.{cfg.dbrectable}
|
||||
WHERE locked = 0 AND status = {CSV_RECEIVED}
|
||||
LIMIT 1
|
||||
"""
|
||||
logger.debug(f"Worker {worker_id} - Esecuzione query: {query}")
|
||||
cur.execute(query)
|
||||
result = cur.fetchone()
|
||||
# Costruisce il nome del modulo da caricare dinamicamente
|
||||
module_names = [f'utils.parsers.by_name.{unit_name.lower()}_{tool_name.lower()}',
|
||||
f'utils.parsers.by_name.{unit_name.lower()}_{tool_type.lower()}',
|
||||
f'utils.parsers.by_name.{unit_name.lower()}_all',
|
||||
f'utils.parsers.by_type.{unit_type.lower()}_{tool_type.lower()}']
|
||||
modulo = None
|
||||
for module_name in module_names:
|
||||
try:
|
||||
logger.debug(f"Worker {worker_id} - Caricamento dinamico del modulo: {module_name}")
|
||||
modulo = importlib.import_module(module_name)
|
||||
logger.debug(f"Worker {worker_id} - Funzione 'main_loader' caricata dal modulo {module_name}")
|
||||
break
|
||||
except (ImportError, AttributeError) as e:
|
||||
logger.info(f"Worker {worker_id} - Modulo {module_name} non presente o non valido. {e}", exc_info=debug_mode)
|
||||
|
||||
if result:
|
||||
id, unit_type, tool_type, unit_name, tool_name = result
|
||||
logger.info(f"Worker {worker_id} - Trovato CSV da elaborare: ID={id}, Tipo={unit_type}_{tool_type}, Nome={unit_name}_{tool_name}")
|
||||
if not modulo:
|
||||
logger.error(f"Worker {worker_id} - Nessun modulo trovato {module_names}")
|
||||
return False
|
||||
|
||||
lock_query = f"UPDATE {cfg.dbname}.{cfg.dbrectable} SET locked = 1 WHERE id = {id}"
|
||||
logger.debug(f"Worker {worker_id} - Esecuzione lock del record: {lock_query}")
|
||||
cur.execute(lock_query)
|
||||
conn.commit()
|
||||
# Ottiene la funzione 'main_loader' dal modulo
|
||||
|
||||
# Costruisce il nome del modulo da caricare dinamicamente
|
||||
module_names = [f'utils.parsers.by_name.{unit_name.lower()}_{tool_name.lower()}',
|
||||
f'utils.parsers.by_name.{unit_name.lower()}_{tool_type.lower()}',
|
||||
f'utils.parsers.by_name.{unit_name.lower()}_all',
|
||||
f'utils.parsers.by_type.{unit_type.lower()}_{tool_type.lower()}']
|
||||
modulo = None
|
||||
for module_name in module_names:
|
||||
try:
|
||||
logger.debug(f"Worker {worker_id} - Caricamento dinamico del modulo: {module_name}")
|
||||
modulo = importlib.import_module(module_name)
|
||||
logger.debug(f"Worker {worker_id} - Funzione 'main_loader' caricata dal modulo {module_name}")
|
||||
except (ImportError, AttributeError) as e:
|
||||
logger.info(f"Worker {worker_id} - Modulo {module_name} non presente o non valido. {e}", exc_info=debug_mode)
|
||||
funzione = getattr(modulo, "main_loader")
|
||||
|
||||
if not modulo:
|
||||
logger.error(f"Worker {worker_id} - Nessun modulo trovato {module_names}")
|
||||
return True, False
|
||||
# Esegui la funzione
|
||||
|
||||
# Ottiene la funzione 'main_loader' dal modulo
|
||||
logger.info(f"Worker {worker_id} - Elaborazione con modulo {modulo} per ID={id}")
|
||||
await funzione(cfg, id, pool)
|
||||
logger.info(f"Worker {worker_id} - Elaborazione completata per ID={id}")
|
||||
return True
|
||||
|
||||
funzione = getattr(modulo, "main_loader")
|
||||
|
||||
# Esegui la funzione
|
||||
await funzione(cfg, id)
|
||||
logger.info(f"Worker {worker_id} - Elaborazione completata per ID={id}")
|
||||
return True, True
|
||||
else:
|
||||
logger.debug(f"Worker {worker_id} - Nessun record disponibile per l'elaborazione")
|
||||
return False, False
|
||||
|
||||
except mysql.connector.Error as e:
|
||||
logger.error(f"Worker {worker_id} - Errore database: {e}", exc_info=debug_mode)
|
||||
return False, False
|
||||
|
||||
|
||||
async def main():
|
||||
@@ -150,45 +125,36 @@ async def main():
|
||||
)
|
||||
logger.info("Logging configurato correttamente")
|
||||
|
||||
# Crea una coda di lavoro illimitata
|
||||
queue = asyncio.Queue(maxsize=cfg.max_threads * 2 or 20)
|
||||
logger.debug("Coda di lavoro creata")
|
||||
|
||||
# Numero massimo di worker concorrenti
|
||||
num_workers = cfg.max_threads
|
||||
logger.info(f"Avvio di {num_workers} worker concorrenti")
|
||||
logger.info(f"Avvio di {cfg.max_threads} worker concorrenti")
|
||||
|
||||
pool = await aiomysql.create_pool(
|
||||
host=cfg.dbhost,
|
||||
user=cfg.dbuser,
|
||||
password=cfg.dbpass,
|
||||
db=cfg.dbname,
|
||||
minsize=1,
|
||||
maxsize=cfg.max_threads*4
|
||||
)
|
||||
|
||||
# Avvia i worker
|
||||
workers = [
|
||||
asyncio.create_task(worker(i, queue, cfg)) for i in range(num_workers)
|
||||
asyncio.create_task(worker(i, cfg, pool))
|
||||
for i in range(cfg.max_threads)
|
||||
]
|
||||
|
||||
logger.info("Sistema avviato correttamente. In attesa di nuovi task...")
|
||||
|
||||
# Ciclo infinito per aggiungere lavori alla coda
|
||||
while True:
|
||||
logger.debug("Aggiunta di un nuovo lavoro alla coda")
|
||||
await queue.put(None)
|
||||
|
||||
# Breve attesa prima di aggiungere un altro lavoro
|
||||
await asyncio.sleep(0.5)
|
||||
try:
|
||||
await asyncio.gather(*workers, return_exceptions=debug_mode)
|
||||
finally:
|
||||
pool.close()
|
||||
await pool.wait_closed()
|
||||
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Info: Shutdown richiesto... chiusura in corso")
|
||||
|
||||
# Attendi che tutti i lavori pendenti siano completati
|
||||
logger.info("Attesa completamento dei task in coda...")
|
||||
await queue.join()
|
||||
|
||||
# Ferma i worker
|
||||
logger.info("Chiusura dei worker in corso...")
|
||||
for task in workers:
|
||||
task.cancel()
|
||||
|
||||
await asyncio.gather(*workers, return_exceptions=debug_mode)
|
||||
|
||||
logger.info("Info: Tutti i task terminati. Uscita.")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Errore principale: {e}", exc_info=debug_mode)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user