from threading import Condition, RLock, Thread from time import sleep,time debug = False ''' Una semplice classe blocking queue che implementa il metodo put e get in forma bloccante La dimensione della coda è fissata in fase di inizializzazione ''' class BlockingQueue: def __init__(self,size): # lista che contiene gli elementi inseriti nella coda self.elementi = [] # dimensione massima della coda self.size = size # RLock che viene utilizzato per disciplinare le invocazioni simultanee ai metodi put e get self.lock = RLock() # Condizione che viene utilizzata per notificare i thread che attendono che la coda non sia più tutta piena self.conditionTuttoPieno = Condition(self.lock) # Condizione che viene utilizzata per notificare i thread che attendono che la coda non sia più tutta vuota self.conditionTuttoVuoto = Condition(self.lock) # # AGGIUNTE PER SOLUZIONE FEB 2024 # self.logSize = 0 self.logs = [] self.logActive = False def __log(self,action,element): if self.logActive: if len(self.logs) >= self.logSize: self.logs.pop(0) self.logs.append([action,element,time()]) if debug: print(f"{self.logs[-1]}") # # Inserisce un elemento nella coda # Se la coda è piena, il thread che invoca il metodo put viene bloccato # Se la coda non è piena, il thread che invoca il metodo put inserisce l'elemento nella coda e notifica un thread che attende che la coda non sia più tutta vuota # def put(self,t): with self.lock: # # Se non ci sono slot liberi, il thread che invoca il metodo put viene bloccato # while len(self.elementi) == self.size: self.conditionTuttoPieno.wait() # # Questo if serve per evitare notify ridondanti # Non ci possono essere consumatori in attesa a meno che, un attimo prima della append(t) la coda non fosse totalmente vuota # Se non ci sono consumatori in attesa, non c'è bisogno di notificare nessuno # Il codice è corretto anche senza questo if, ma ci saranno notify anche quando non necessari # if len(self.elementi) == 0: self.conditionTuttoVuoto.notify() self.elementi.append(t) ## ## AGGIUNTE PER SOLUZIONE FEB 2024 ## self.__log("put",t) # # Estrae un elemento dalla coda # Se la coda è vuota, il thread che invoca il metodo get viene bloccato # Se la coda contiene almeno un elemento, il thread che invoca il metodo get estrae l'elemento dalla coda e notifica un thread che attende che la coda non sia più tutta piena # def get(self): with self.lock: # # Se non ci sono elementi da estrarre, il thread che invoca il metodo get viene bloccato # while len(self.elementi) == 0: self.conditionTuttoVuoto.wait() # # Questo if serve per evitare notify ridondanti # Non ci possono essere produttori in attesa a meno che, un attimo prima della pop(0) la coda non fosse totalmente piena # Se non ci sono produttori in attesa, non c'è bisogno di notificare nessuno # Il codice è corretto anche senza questo if, ma ci saranno notify anche quando non necessari # if len(self.elementi) == self.size: self.conditionTuttoPieno.notify() ## ## AGGIUNTE PER SOLUZIONE FEB 2024 ## self.__log("get",self.elementi[0]) return self.elementi.pop(0) def startLogging(self,M): with self.lock: self.logSize = M self.logActive = True def stopLogging(self): with self.lock: self.logActive = False self.logs = [] def __find_last(self,op,o,eq=False): # # Non era specificato se cercare per identità o per uguaglianza, quindi ho aggiunto un parametro eq che consente di fare entrambe le cose # # Da non dimenticare che si deve prendere l'elemento più recente! # with self.lock: for [oper,elem,t] in reversed(self.logs): if eq: if op == oper and elem == o: return t else: if op == oper and elem is o: return t return 0 def read_get_log(self,o): return self.__find_last("get",o) def read_put_log(self,o): return self.__find_last("put",o) def read_diff_log(self,o): with self.lock: lastGet = self.read_get_log(o) lastPut = self.read_put_log(o) if lastPut != 0 and lastGet != 0: return lastGet - lastPut else: return -1 # # Questo metodo serve solo per scopi di debug # def __printLog(self): with self.lock: for [op,elem,t] in self.logs: print(f"{op} {elem} {t}") # # Esperimento di utilizzo della classe BlockingQueue # cuochi_e_piatti = { "Cannavacciuolo": ["Pizza", "Pasta", "Tiramisu", "Carbonara"], "Frankie": ["Hamburger", "Patatine", "Frappe"], "Sakura": ["Sushi", "Tempura", "Zuppa di Miso"], } # # Il cuoco svolge il ruolo di produttore # class Cuoco(Thread): def __init__(self,q,nome): super().__init__() self.nastroPiatti = q self.name = nome # # Il ciclo di lavoro del Cuoco prevede che ogni 0.1 secondi inserisca un piatto nella coda # def run(self): numIterazioni = 500 while numIterazioni > 0: numIterazioni -= 1 sleep(0.1) listaPiattiDiQuestoCuoco = cuochi_e_piatti[self.name] piattoProdotto = listaPiattiDiQuestoCuoco[numIterazioni % len(listaPiattiDiQuestoCuoco)] self.nastroPiatti.put(piattoProdotto) #print (f"Cuoco {self.name} ha inserito CIBA: {piattoProdotto}") # # Il thread Cameriere svolge il ruolo di consumatore # class Cameriere(Thread): def __init__(self,q,nome): super().__init__() self.nastroPiatti = q self.name = nome # # Il ciclo di lavoro del Cameriere prevede che ogni 1 secondo si prelevi un piatto dalla coda # def run(self): numIterazioni = 500 while numIterazioni > 0: numIterazioni -= 1 sleep(1) piatto = self.nastroPiatti.get() #print (f"Cameriere {self.name} ha prelevato CIBA: {piatto}") class Logger(Thread): def __init__(self,q): super().__init__() self.nastroPiatti = q def run(self): numIterazioni = 500 self.nastroPiatti.startLogging(1000) while numIterazioni > 0: numIterazioni -= 1 sleep(5) print (f"Ultimo get di Pizza: {self.nastroPiatti.read_get_log('Pizza')}") print (f"Ultimo put di Pizza: {self.nastroPiatti.read_put_log('Pizza')}") print (f"Differenza tra ultimo put e ultimo get di Pizza: {self.nastroPiatti.read_diff_log('Pizza')}") #self.nastroPiatti._BlockingQueue__printLog() self.nastroPiatti.stopLogging() # # Codice main di prova # Questo if consente di evitare che il codice venga eseguito quando il modulo viene importato # Il codice viene eseguito solo quando il modulo viene eseguito come programma principale # if __name__ == "__main__": # crea una coda di dimensione 10 q = BlockingQueue(10) # # Cread un thread logger di prova # logger = Logger(q) logger.start() # # Crea i cuochi in base a come sono definiti in cuochi_e_piatti e li avvia # for c in cuochi_e_piatti: newCuoco = Cuoco(q,c) newCuoco.start() # # Crea 10 camerieri e li avvia # for c in range(0,10): newCameriere = Cameriere(q,f"Cameriere-{c}") newCameriere.start()