-
Notifications
You must be signed in to change notification settings - Fork 0
/
gestore_pipeline.py
425 lines (401 loc) · 21.9 KB
/
gestore_pipeline.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
# This Python file uses the following encoding: utf-8
"""
This work is licensed under the Creative Commons Attribution 4.0 International
License. To view a copy of this license, visit
http://creativecommons.org/licenses/by/4.0/ or send a letter to Creative
Commons, PO Box 1866, Mountain View, CA 94042, USA.
"""
import logging
from multiprocessing import Queue,Lock
from time import time,sleep
#Framework
from oggetto import oggetto
from gestore_segnali import gestore_segnali
# Operazioni
## Mettere qui le operazioni della pipeline
from monitor import monitor
from comunicazione_tcp import comunicazione_tcp
ATTESA_CICLO_PRINCIPALE = 0.001
class gestore_pipeline(oggetto):
"""Gestore Pipeline
Gestisce la coda delle operazioni che il programma deve eseguire.
Fa da arbitro nelle comunicazioni tra le operazioni e tra le operazioni e
il Gestore Pipeline (sé stesso) ed orchestra le operazioni.
Si assicura che le operazioni vengano eseguite nell'ordine stabilito
"""
def __init__(self,
file_configurazione,
coda_ipc_entrata,
lock_ipc_entrata,
coda_ipc_uscita,
lock_ipc_uscita):
super().__init__(coda_ipc_entrata,
lock_ipc_entrata,
coda_ipc_uscita,
lock_ipc_uscita)
logging.info(type(self).__name__ + " inizializzazione")
##### Inizializzazione comune a tutti gli oggetti del framework ########
##################### Lettura delle impostazioni #######################
configurazione = []
lista_configurazione = []
impostazioni = []
# Leggi le impostazioni dal file configurazione e mettile in una lista
with open(file_configurazione) as f:
configurazione = f.readlines()
lista_configurazione[:] = [x.strip() for x in configurazione]
# La lista delle impostazioni è una lista di liste, così da permettere
# indici non unici
for impostazione in lista_configurazione:
nome,valore = impostazione.split(" ")
impostazioni.append([nome,valore])
################# Fine lettura delle impostazioni ######################
#### Fine inizializzazione comune a tutti gli oggetti del framework ####
################### Inizializza le impostazioni ########################
# TODO: controlla le impostazioni già scritte e inizializza le
# impostazioni mancanti
self.lista_segnali = []
# Dizionario con le operazioni da eseguire nell'ordine di esecuzione
self.operazioni = {} # "nome": operazione
# Dizionario con le code per le comunicazioni tra il Gestore Pipeline e
# le operazioni. Le code sono unidirezionali: qui ci sono i messaggi che
# devono essere mandati alle operazioni.
# Usato *solo* dai Gestori Segnali per le comunicazioni oggetto-oggetto
# (Gestore Segnali - Gestore Segnali)
self.ipc_entrata_operazioni = {} # "nome operazione": coda
# Dizionario con i lock per le code IPC in entrata
self.lock_ipc_entrata_operazioni = {} # "nome operazione": coda
# Dizionario con le code per le comunicazioni tra il Gestore Pipeline e
# le operazioni. Le code sono unidirezionali: qui ci sono i messaggi
# ricevuti dalle operazioni.
# Usato *solo* dai Gestori Segnali per le comunicazioni oggetto-oggetto
# (Gestore Segnali - Gestore Segnali)
self.ipc_uscita_operazioni = {}
# Dizionario con i lock per le code IPC in uscita
self.lock_ipc_uscita_operazioni = {}
# Dizionario con le code per i segnali ricevuti dalle operazioni. Il
# Gestore Pipeline legge i segnali delle operazioni da qui
self.coda_segnali_entrata_operazioni = {}
# dizionario con i lock per le code in uscita
self.lock_segnali_uscita_operazioni = {}
# Dizionario con le code per i segnali da inviare alle operazioni. Il
# Gestore Pipeline scrive i segnali per le operazioni da qui
self.coda_segnali_uscita_operazioni = {}
# Dizionario con i lock per le code in entrata
self.lock_segnali_entrata_operazioni = {}
# Dizionario con i gestori segnali associati alle singole operazioni.
# Per ogni operazione della pipeline, il Gestore Pipeline si crea un
# Gestore Segnali per la comunicazione con quella operazione
self.gestore_segnali_operazioni = {}
# Segnale in entrata dall'esterno dell'applicazione (dalla coda IPC)
# Preleva le impostazioni del Gestore Pipeline. Le impostazioni sono:
# -) Operazione: il nome dell'operazione da aggiungere alla pipeline
# -) Segnale: un segnale che il Gestore Pipeline può inviare
for impostazione in impostazioni:
nome,valore = impostazione
# Aggiungi il segnale alla lista dei segnali
if nome == "segnale":
self.lista_segnali.append(valore)
# Aggiungi l'operazione alla pipeline
if nome == "operazione":
# Inizializza le code e i lock *associati* all'operazione nel
# Gestore Pipeline
self.ipc_entrata_operazioni[valore] = Queue()
self.lock_ipc_entrata_operazioni[valore] = Lock()
self.ipc_uscita_operazioni[valore] = Queue()
self.lock_ipc_uscita_operazioni[valore] = Lock()
self.coda_segnali_entrata_operazioni[valore] = Queue()
self.lock_segnali_entrata_operazioni[valore] = Lock()
self.coda_segnali_uscita_operazioni[valore] = Queue()
self.lock_segnali_uscita_operazioni[valore] = Lock()
# Inizializza il Gestore Segnali *associato* all'operazione
self.gestore_segnali_operazioni[valore] = gestore_segnali(
type(self).__name__,
self.ipc_entrata_operazioni[valore],
self.lock_ipc_entrata_operazioni[valore],
self.ipc_uscita_operazioni[valore],
self.lock_ipc_uscita_operazioni[valore],
self.coda_segnali_entrata_operazioni[valore],
self.lock_segnali_entrata_operazioni[valore],
self.coda_segnali_uscita_operazioni[valore],
self.lock_segnali_uscita_operazioni[valore],
controlla_destinatario=False,
inoltra=True)
# Avvia il Gestore Segnali *associato* all'operazione
self.gestore_segnali_operazioni[valore].start()
sleep(0.01)
with self.lock_segnali_uscita_operazioni[valore]:
if not self.coda_segnali_uscita_operazioni[valore].full():
self.coda_segnali_uscita_operazioni[valore].put_nowait(["avvia","gestore_segnali"])
# Inizializza l'operazione nella coda delle operazioni
self.operazioni[valore] = globals()[valore](
str(valore + ".conf"),
self.ipc_uscita_operazioni[valore],
self.lock_ipc_uscita_operazioni[valore],
self.ipc_entrata_operazioni[valore],
self.lock_ipc_entrata_operazioni[valore])
logging.info(self.operazioni[valore])
sleep(0.1)
# Avvia tutte le operazioni
for nome,operazione in self.operazioni.items():
print(type(operazione))
logging.info(type(self).__name__ + " sta avviando " + nome + " " + str(operazione))
operazione.start()
################ Fine inizializza le impostazioni ######################
logging.info(type(self).__name__ + " inizializzato")
# def run(self):
# """Punto d'entrata del processo/thread"""
# logging.info(type(self).__name__ + " creato")
# # Entra nello stato richiesto
# while True:
# logging.info(type(self).__name__ + " entrando in " + self.stato)
# s = getattr(self,self.stato)()
# if isinstance(s,int):
# if s != 0:
# sleep(ATTESA_CICLO_PRINCIPALE)
# break
# return int(s)
def idle(self):
logging.info(type(self).__name__ + " idle")
pacchetto_segnale_entrata = []
segnale = ""
mittente = ""
destinatario = ""
timestamp = 0
# Segnala all'esterno che sei in idle
with self.lock_segnali_uscita:
if not self.coda_segnali_uscita.full():
self.coda_segnali_uscita.put_nowait(["idle",""])
# Attendi il segnale di avvio
while True:
pacchetto_segnale_entrata[:] = []
segnale = ""
mittente = ""
destinatario = ""
timestamp = 0
with self.lock_segnali_entrata:
if not self.coda_segnali_entrata.empty():
pacchetto_segnale_entrata[:] = \
self.coda_segnali_entrata.get_nowait()
if len(pacchetto_segnale_entrata) == 4:
segnale,mittente,destinatario,timestamp = \
pacchetto_segnale_entrata
logging.info(type(self).__name__)
logging.info(pacchetto_segnale_entrata)
logging.info(segnale)
elif len(pacchetto_segnale_entrata) == 3:
segnale,mittente,timestamp = pacchetto_segnale_entrata
logging.info(type(self).__name__)
logging.info(pacchetto_segnale_entrata)
logging.info(segnale)
elif len(pacchetto_segnale_entrata) == 0:
pass
else:
with self.lock_segnali_uscita:
self.coda_segnali_uscita.put_nowait([" segnale mal formato",
""])
sleep(ATTESA_CICLO_PRINCIPALE)
logging.info(type(self).__name__ + " Segnale mal formato")
pacchetto_segnale_entrata[:] = []
continue
pacchetto_segnale_entrata[:] = []
if segnale == "":
sleep(ATTESA_CICLO_PRINCIPALE)
continue
# Se hai ricevuto il segnale di stop
elif segnale == "stop":
with self.lock_segnali_uscita:
self.coda_segnali_uscita.put_nowait(["terminato",""])
# Invia il segnale di stop anche al tuo Gestore Segnali
self.coda_segnali_uscita.put_nowait(["stop",
"gestore_segnali"])
# Termina segnalando l'uscita per segnale di stop
return int(-1)
else:
# Se il segnale è tra i metodi riconosciuti dal Gestore Pipeline
if segnale in dir(self):
#Esegui il segnale
s = getattr(self,segnale)()
return int(s)
else:
with self.lock_segnali_uscita:
self.coda_segnali_uscita.put_nowait( \
["Segnale non valido",
""])
sleep(ATTESA_CICLO_PRINCIPALE)
sleep(ATTESA_CICLO_PRINCIPALE)
############## Fine ricezione messaggi dall'esterno ################
def avvia(self):
logging.info(type(self).__name__ + " avviato")
pacchetto_segnale_entrata = []
segnale = ""
mittente = ""
destinatario = ""
timestamp = 0
richiesta_stop = False
# Segnala all'esterno che sei avviato
with self.lock_segnali_uscita:
if not self.coda_segnali_uscita.full():
self.coda_segnali_uscita.put_nowait(["avviato",""])
for nome,operazione in self.operazioni.items():
# Manda il segnale di avvio all'operazione.
with self.lock_ipc_uscita_operazioni[nome]:
self.ipc_uscita_operazioni[nome].put_nowait("avvia:" + \
str(time()) + ":" + \
type(self).__name__ \
+ ":" + nome)
while True:
pacchetto_segnale_entrata[:] = []
segnale = ""
mittente = ""
destinatario = ""
timestamp = 0
if richiesta_stop:
for operazione in self.operazioni:
with self.lock_segnali_uscita:
self.coda_segnali_uscita.put_nowait(["terminando: " + \
str(operazione),
""])
with self.lock_segnali_uscita_operazioni[operazione]:
self.coda_segnali_uscita_operazioni[operazione].put_nowait(["stop",operazione])
with self.lock_segnali_uscita_operazioni[operazione]:
self.coda_segnali_uscita_operazioni[operazione].put_nowait(["stop","gestore_segnali"])
#self.operazioni[operazione].join()
with self.lock_segnali_uscita:
self.coda_segnali_uscita.put_nowait([str(operazione) + \
" terminata",""])
with self.lock_segnali_uscita:
self.coda_segnali_uscita.put_nowait(["stop",
"gestore_segnali"])
return int(-1)
with self.lock_segnali_entrata:
if not self.coda_segnali_entrata.empty():
pacchetto_segnale_entrata[:] = \
self.coda_segnali_entrata.get_nowait()
logging.debug("IPC")
logging.debug(pacchetto_segnale_entrata)
if len(pacchetto_segnale_entrata) == 4:
segnale,mittente,destinatario,timestamp = \
pacchetto_segnale_entrata
pacchetto_segnale_entrata[:] = []
elif len(pacchetto_segnale_entrata) == 3:
segnale,mittente,timestamp = pacchetto_segnale_entrata
pacchetto_segnale_entrata[:] = []
elif len(pacchetto_segnale_entrata) == 0:
pass
else:
with self.lock_segnali_uscita:
self.coda_segnali_uscita.put_nowait(["segnale mal formato",
""])
logging.info(type(self).__name__ + " Segnale mal formato")
pacchetto_segnale_entrata[:] = []
sleep(ATTESA_CICLO_PRINCIPALE)
continue
# Se hai ricevuto il segnale di stop
if segnale == "stop":
# Invia il segnale di stop anche al tuo Gestore Segnali
with self.lock_segnali_uscita:
self.coda_segnali_uscita.put_nowait( \
["terminando: " + \
type(self).__name__,
""])
richiesta_stop = True
elif segnale != "":
if destinatario == "":
for (ogg,lock_uscita), \
(ogg,coda_segnali_uscita) \
in \
zip(self.lock_segnali_uscita_operazioni.items(), \
self.coda_segnali_uscita_operazioni.items()):
with self.lock_segnali_uscita_operazioni[str(ogg)]:
self.coda_segnali_uscita_operazioni[str(ogg)].put_nowait([segnale,destinatario,mittente])
############## Fine ricezione messaggi dall'esterno ################
########## Comunicazione con le operazioni della pipeline ##########
# Per ogni operazione
for (ogg,lock_entrata), \
(ogg,coda_segnali_entrata), \
(ogg,lock_uscita), \
(ogg,coda_segnali_uscita) \
in \
zip(self.lock_segnali_entrata_operazioni.items(), \
self.coda_segnali_entrata_operazioni.items(), \
self.lock_segnali_uscita_operazioni.items(), \
self.coda_segnali_uscita_operazioni.items()):
#TODO: CONTROLLA DA QUI
pacchetto_segnale_entrata[:] = []
segnale = ""
mittente = ""
destinatario = ""
timestamp = 0
logging.debug(id(coda_segnali_entrata))
# Leggi l'eventuale segnale dall'operazione
with lock_entrata:
if not coda_segnali_entrata.empty():
pacchetto_segnale_entrata[:] = \
coda_segnali_entrata.get_nowait()
logging.debug(ogg)
logging.debug(pacchetto_segnale_entrata)
if len(pacchetto_segnale_entrata) == 4:
segnale,mittente,destinatario,timestamp = \
pacchetto_segnale_entrata
pacchetto_segnale_entrata[:] = []
elif len(pacchetto_segnale_entrata) == 3:
segnale,mittente,timestamp = pacchetto_segnale_entrata
pacchetto_segnale_entrata[:] = []
elif len(pacchetto_segnale_entrata) == 0:
sleep(ATTESA_CICLO_PRINCIPALE)
continue
else:
with self.lock_segnali_uscita:
self.coda_segnali_uscita.put_nowait( \
["segnale mal formato",""])
with lock_uscita:
coda_segnali_uscita.put_nowait(["segnale mal formato",
""])
logging.info(type(self).__name__ + " Segnale mal formato")
pacchetto_segnale_entrata[:] = []
sleep(ATTESA_CICLO_PRINCIPALE)
continue
logging.debug("Gestore Pipeline " + \
segnale + " " + \
mittente + " " + \
destinatario + " " + \
str(timestamp))
sleep(ATTESA_CICLO_PRINCIPALE)
if str(segnale) != "":
# Se il destinatario è il Gestore Pipeline
if str(destinatario) == type(self).__name__:
if segnale == "stop":
richiesta_stop = True
sleep(ATTESA_CICLO_PRINCIPALE)
break
elif segnale == "lista_operazioni":
ops = ""
prima_operazione = 1
for op in self.operazioni:
if prima_operazione == 1:
ops = str(op)
prima_operazione = 0
else:
ops = ops + "," + str(op)
with self.lock_segnali_uscita_operazioni[ogg]:
self.coda_segnali_uscita_operazioni[ogg].put_nowait([ops,destinatario,mittente])
# Se il destinatario è una delle altre operazioni
elif str(destinatario) in self.operazioni:
# Inoltra il segnale a quella specifica operazione
with self.lock_segnali_uscita_operazioni[str(destinatario)]:
self.coda_segnali_uscita_operazioni[str(destinatario)].put_nowait([segnale,destinatario,mittente])
# Se il destinatario è "broadcast"
elif str(destinatario) == "":
# Inoltra il segnale a tutte le altre operazioni
for operazione in self.operazioni:
if operazione == ogg:
sleep(ATTESA_CICLO_PRINCIPALE)
continue
else:
with self.lock_segnali_uscita_operazioni[str(operazione)]:
self.coda_segnali_uscita_operazioni[str(operazione)].put_nowait([segnale,destinatario,mittente])
sleep(ATTESA_CICLO_PRINCIPALE)
if segnale == "stop":
richiesta_stop = True
############## Fine comunicazione con le operazioni ################
sleep(ATTESA_CICLO_PRINCIPALE)