#/*########################################################################## # Copyright (C) 2001-2013 European Synchrotron Radiation Facility # # PyHST2 # European Synchrotron Radiation Facility, Grenoble,France # # PyHST2 is developed at # the ESRF by the Scientific Software staff. # Principal author for PyHST2: Alessandro Mirone. # # This program is free software; you can redistribute it and/or modify it # under the terms of the GNU General Public License as published by the Free # Software Foundation; either version 2 of the License, or (at your option) # any later version. # # PyHST2 is distributed in the hope that it will be useful, but WITHOUT ANY # WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. See the GNU General Public License for more # details. # # You should have received a copy of the GNU General Public License along with # PyHST2; if not, write to the Free Software Foundation, Inc., 59 Temple Place, # Suite 330, Boston, MA 02111-1307, USA. # # PyHST2 follows the dual licensing model of Trolltech's Qt and Riverbank's PyQt # and cannot be used as a free plugin for a non-free program. # # Please contact the ESRF industrial unit (industry@esrf.fr) if this license # is a problem for you. #############################################################################*/ from __future__ import absolute_import from __future__ import division from __future__ import print_function import threading global P, C import numpy import mpi4py.MPI as MPI import time myrank = MPI.COMM_WORLD.Get_rank() GLobalLock = threading.Lock() WriteLock = threading.Lock() VERBOSE_RESOURCES_DEBUG=0 DOSHARED=1 class Control: def __init__(self, Parameters, cspaceA, nofCpus , mygpus, npbunches , Aspace): global P,C global NPBUNCHES self.Aspace = Aspace P=Parameters C=cspaceA NPBUNCHES=npbunches ntokens=C.get_n_arraysraw() self.RawDataResources = ArrayResources( ntokens ) self.TreatedDataResources = ArrayResources( C.get_n_arraystreated () ) self.TransposedDataResources = ArrayResources( C.get_n_arraystransposed () ) self.MPI_Resources = MPI_Resources( 1 ) self.PassCheckResources = PassCheckResources(self.MPI_Resources ) if myrank==0: with WriteLock: if P.VERBOSITY>0 : print( "len(P.first_slices) " , len(P.first_slices)) self.PacketsManager = PacketsManager( len(P.first_slices), nslots = 100 ) self.SiliciumResources = SiliciumResources( nofCpus *P.ncpus_expansion , mygpus ) if Parameters.STEAM_INVERTER: STEAM_DIRECTION = -1 else: STEAM_DIRECTION = +1 if Parameters.DOUBLEFFCORRECTION_ONTHEFLY: DO_FF2 = 1 else: DO_FF2 = 0 if P.VERBOSITY>0 : print( " STEAM_DIRECTION ", STEAM_DIRECTION ) start = ProcessingStart ( processingLevel=0, resources={"SiliciumResources":self.SiliciumResources }, resources_free = ["SiliciumResources"], pmanager = self.PacketsManager ) if STEAM_DIRECTION==1: ## ----------------------------------------------------------------------------- if Parameters.DO_OUTPUT_PAGANIN or Parameters.DOUBLERUN4DFF : preprocessing_frees = ["SiliciumResources","RawDataResources","TreatedDataResources" ] else: preprocessing_frees = ["SiliciumResources","RawDataResources" ] if Parameters.RAWDATA_MEMORY_REUSE<2: # if it is 1 : cas peu encombrant , qui rentre raisonablement bien en memoire reading1 = ProcessingReading( processingLevel=1, resources={"SiliciumResources":self.SiliciumResources , "RawDataResources":self.RawDataResources }, resources_free = ["SiliciumResources"], pmanager = self.PacketsManager, STEAM_DIRECTION = STEAM_DIRECTION ) start.NEXTPOOL= [reading1] preprocessing = Processing_PreProcessing( processingLevel=2, resources={"SiliciumResources":self.SiliciumResources , "RawDataResources":self.RawDataResources , "TreatedDataResources":self.TreatedDataResources , }, resources_free = preprocessing_frees, pmanager = self.PacketsManager, STEAM_DIRECTION = STEAM_DIRECTION ) reading1.NEXTPOOL= [preprocessing] else: # pour les cas encombrants if DO_FF2: tobefreed = ["SiliciumResources"] else: tobefreed = preprocessing_frees preprocessing0 = Processing_InterleavedReadingPreProcessing( processingLevel=2, resources={"SiliciumResources":self.SiliciumResources , "RawDataResources":self.RawDataResources , "TreatedDataResources":self.TreatedDataResources , }, resources_free = tobefreed, pmanager = self.PacketsManager, STEAM_DIRECTION = STEAM_DIRECTION, FF2=DO_FF2*1 ) start.NEXTPOOL= [preprocessing0] if DO_FF2 : # The threads are done computing their local mean. Now, one thread will reduce the results, # And the process 0 will accumulate all the means preprocessing = Processing_InterleavedReadingPreProcessing( processingLevel=4, resources={"SiliciumResources":self.SiliciumResources , "RawDataResources":self.RawDataResources , "TreatedDataResources":self.TreatedDataResources , }, resources_free = preprocessing_frees, pmanager = self.PacketsManager, STEAM_DIRECTION = STEAM_DIRECTION, FF2=DO_FF2*2 ) preprocessing0.NEXTPOOL= [preprocessing] else: preprocessing = preprocessing0 ## ----------------------------------------------------------------------------- if Parameters.DO_OUTPUT_PAGANIN and STEAM_DIRECTION==-1: raise Exception( "INPUT inconsistency : you have selected DO_OUTPUT_PAGANIN and STEAM_DIRECTION=-1. ") LEVEL=6 if (not Parameters.DO_OUTPUT_PAGANIN ) and (not Parameters.DOUBLERUN4DFF) : TLEVEL = LEVEL transposing = Processing_Transposing( processingLevel=LEVEL-0*DOSHARED*(STEAM_DIRECTION==1), resources={"MPI_Resources":self.MPI_Resources , "TreatedDataResources":self.TreatedDataResources , "TransposedDataResources":self.TransposedDataResources , }, resources_free = { 1:["TreatedDataResources"]+["MPI_Resources"]*DOSHARED, -1:["MPI_Resources","TransposedDataResources", "TreatedDataResources" ] }[STEAM_DIRECTION], pmanager = self.PacketsManager, STEAM_DIRECTION = STEAM_DIRECTION, Ctr = self ) if STEAM_DIRECTION==1: preprocessing.NEXTPOOL= [transposing] if(DOSHARED==0): dispenser = Processing_Dispenser( processingLevel=LEVEL, resources={"MPI_Resources":self.MPI_Resources , "TransposedDataResources":self.TransposedDataResources , }, resources_free = [], pmanager = self.PacketsManager, STEAM_DIRECTION = STEAM_DIRECTION ) if STEAM_DIRECTION==1: transposing.NEXTPOOL= [dispenser] else: dispenser.NEXTPOOL= [transposing] if STEAM_DIRECTION == -1: LEVEL = 0 reconstructor = Processing_Reconstructing( processingLevel=LEVEL, resources={ 1:{"SiliciumResources":self.SiliciumResources , "MPI_Resources":self.MPI_Resources , "TransposedDataResources":self.TransposedDataResources }, -1:{"SiliciumResources":self.SiliciumResources , "TransposedDataResources":self.TransposedDataResources } }[STEAM_DIRECTION], resources_free = { 1:["SiliciumResources","MPI_Resources", "TransposedDataResources"], -1:["SiliciumResources"]}[STEAM_DIRECTION], pmanager = self.PacketsManager, STEAM_DIRECTION = STEAM_DIRECTION ) if STEAM_DIRECTION==1: dispenser.NEXTPOOL= [reconstructor] else: reconstructor.NEXTPOOL= [dispenser] else: if STEAM_DIRECTION == -1: LEVEL = 0 reconstructor = Processing_ReconstructingSHARED( processingLevel=LEVEL, resources={ 1:{"SiliciumResources":self.SiliciumResources , "TransposedDataResources":self.TransposedDataResources }, -1:{"SiliciumResources":self.SiliciumResources , "TransposedDataResources":self.TransposedDataResources } }[STEAM_DIRECTION], resources_free = { 1:["SiliciumResources", "TransposedDataResources"], -1:["SiliciumResources"]}[STEAM_DIRECTION], pmanager = self.PacketsManager, STEAM_DIRECTION = STEAM_DIRECTION ) if STEAM_DIRECTION==1: transposing.NEXTPOOL= [reconstructor] else: reconstructor.NEXTPOOL= [transposing] dumsink = ProcessingBaseClass( processingLevel=6, resources={ }, resources_free = [], pmanager = self.PacketsManager ) dumsink.collapse=1 dumsink.NEXTPOOL = [] if STEAM_DIRECTION==1: self.PacketsManager.setPacketStart(start) if (not Parameters.DO_OUTPUT_PAGANIN ) and (not Parameters.DOUBLERUN4DFF) : reconstructor.NEXTPOOL=[dumsink] else: preprocessing.NEXTPOOL=[dumsink] else: self.PacketsManager.setPacketStart(start) start.NEXTPOOL = [ reconstructor ] transposing.NEXTPOOL=[dumsink] def execute(self, postprocess=None): while(1): with GLobalLock: possibility = self.PacketsManager.CheckAssignPackets_getppn2operate( ) # e lo mette a start se e nuovo # qui si riservano le # risorse , le si affettano al packet # si incrementano le fasi di processo # ma il packet lo si puo mettere nel gestore if possibility is None: break if possibility==-1: time.sleep(0.001) continue packetSerialNumber, processingObject = possibility if myrank==0 : with WriteLock: if P.VERBOSITY>0 : print( "PROCESSING PACKET N: " , packetSerialNumber , " WITH PROCESSING OBJECT :", processingObject) self.PacketsManager.process_step_packet_in_thread( packetSerialNumber, processingObject) if postprocess : postprocess(C) class SiliciumResources: def __init__(self, nofCpus , mygpus): self.nofCpus=nofCpus self.mygpus=mygpus self.nofusedcpus=0 self.usedgpus={} self.resourcesForPacket={} def freeResources(self, processingObject=None, packetSerialNumber=None , notused=0): nusedcpus, usedgpus = self.resourcesForPacket[packetSerialNumber] del self.resourcesForPacket[packetSerialNumber] self.nofusedcpus = self.nofusedcpus - nusedcpus for gpu in usedgpus: del self.usedgpus[gpu] def print_resources(self): if myrank==0: with WriteLock: if P.VERBOSITY>0 : print( " in " , self, " usati " , self.nofusedcpus,len(self.usedgpus), " totale " , self.nofCpus, len(self.mygpus) ) def getResources(self, processingObject=None, packetSerialNumber=None , resourcename=None): # print( " oggetto richiedente " , processingObject, processingObject.requiredCpusGpus) Ncpus, ngpus = processingObject.requiredCpusGpus # print( " richiesta " , ncpus, ngpus) try: iter(Ncpus) except TypeError:Ncpus=[Ncpus] for ncpus in Ncpus: ncpus=int(ncpus) if ncpus<=(self.nofCpus-self.nofusedcpus) and ngpus<=(len(self.mygpus)-len(self.usedgpus)) : count=0 availablegpus=[] for tok in self.mygpus: if tok not in self.usedgpus: availablegpus.append(tok) count=count+1 if count==ngpus: break # take the resources and keeps track self.nofusedcpus =self.nofusedcpus+ncpus for gpu in availablegpus: self.usedgpus[gpu] = packetSerialNumber # if myrank==0: # with WriteLock: # print( "AVAILABLE SILICIUM RESOURCES : " , ( ncpus, availablegpus), " FOR PACKET :" , packetSerialNumber ) self.resourcesForPacket[packetSerialNumber]=( ncpus, availablegpus) return 1 return 0 class ArrayResources: def __init__(self, ntokens ) : self.Ntokens = ntokens self.PacketID=[None]*self.Ntokens self.ArrayStatus = [ ARRAYFREE ] *self.Ntokens self.tokenperpacketid={} self.reference_count_on ={} self.reference_count_off={} self.packet_has_gone = {} # QUI OCCORRE FARE UNA GESTIONE PUNTUALE DEGLI ACCESSI ALLA RISORSA # CIOE reference_count_on/off PER OGNI pid = packetSerialNumber / NPBUNCHES # deve essere un array di lunghezza NPBUNCHES def verbosedebug_resource_search1( self, pid, pid_i ): if VERBOSE_RESOURCES_DEBUG: if myrank==0: with WriteLock: print( "///////////////////////////////////////////") print( " SEARCHING RESOURCES FOR PID : " , pid , " PID_I :" ,pid_i ) print( " self.tokenperpacketid " , self.tokenperpacketid) print( " self.reference_count_on " , self.reference_count_on) print( "\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\") def verbosedebug_resource_search2( self, pid, pid_i , packetSerialNumber ): if VERBOSE_RESOURCES_DEBUG: if myrank==0: with WriteLock: print( "///////////////////////////////////////////") print( "ALREADY AVAILABLE ARRAY RESOURCES : " , self.tokenperpacketid[ pid] , " FOR UNDER PACKET :" , packetSerialNumber, " OVER NPBUNCHES", NPBUNCHES, " OBTAINABLES FROM ", self ) print( " self.ArrayStatus " , self.ArrayStatus) print( self.reference_count_on[pid ]) print( "\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\") def verbosedebug_resource_search3( self, pid, pid_i , pos, packetSerialNumber ): if VERBOSE_RESOURCES_DEBUG: if myrank==0: with WriteLock: print( "///////////////////////////////////////////") print( "AVAILABLE ARRAY RESOURCES : " , pos , " FOR PACKET :" , packetSerialNumber, " OVER NPBUNCHES", NPBUNCHES, " OBTAINABLES FROM ", self , " IN SLOT " , pos) print( " self.ArrayStatus " , self.ArrayStatus) print( self.reference_count_on[ pid ]) print( "\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\") def verbosedebug_resource_search4( self, pid, pid_i ): if VERBOSE_RESOURCES_DEBUG: if myrank==0: with WriteLock: print( " //////////////////////// " ) print( " THERE ARE NO FREE ARRAYS AT THE MOMENT " ) print( self.ArrayStatus) print( self.reference_count_on) print( self.reference_count_off) print( "\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\" ) def verbosedebug_resource_free( self, pid, pid_i, packetSerialNumber ): if VERBOSE_RESOURCES_DEBUG: if myrank==0: with WriteLock: print( "PACKET : " , packetSerialNumber," PARTIALLY FREES " , self, self.reference_count_off[pid], self.reference_count_on[pid]) ############################################### def getResources(self, processingObject=None, packetSerialNumber=None , resourcename=None): global NPBUNCHES nrequiredarrays=1 pid = packetSerialNumber // NPBUNCHES pid_i = packetSerialNumber % NPBUNCHES self.verbosedebug_resource_search1( pid, pid_i ) if (pid in self.tokenperpacketid) and numpy.sum(self.reference_count_on[pid]) > 0 : #(self.reference_count_on[pid ][pid_i]-1): self.reference_count_on[pid ][pid_i]=1 self.verbosedebug_resource_search2( pid, pid_i , packetSerialNumber ) return 1 else: for pidsearch, astat, pos in zip( self.PacketID,self.ArrayStatus, range(self.Ntokens )): if astat==ARRAYFREE: if pidsearch is not None: raise Exception( " self.PacketID has not been freed " ) self.PacketID[pos]= pid self.ArrayStatus[pos] = ARRAYINUSE self.tokenperpacketid[ pid ]=pos if not (pid in self.reference_count_on): self.reference_count_on[pid ]=numpy.zeros(NPBUNCHES,"i") self.reference_count_on[pid ][pid_i]=1 self.verbosedebug_resource_search3( pid, pid_i , pos, packetSerialNumber ) return 1 else: self.verbosedebug_resource_search4( pid, pid_i ) return 0 ############################################# def freeResources(self, processingObject=None, packetSerialNumber=None, notused=0 ): global NPBUNCHES pid = packetSerialNumber // NPBUNCHES pid_i = packetSerialNumber % NPBUNCHES if not (pid in self.reference_count_off): self.reference_count_off[pid ]=numpy.zeros(NPBUNCHES, "i") if notused: self.reference_count_on[pid][pid_i] =0 if numpy.sum(self.reference_count_on[pid])==0: self.totalfree(pid,packetSerialNumber, notused) return else: self.reference_count_off[pid][pid_i] =1 self.verbosedebug_resource_free( pid, pid_i , packetSerialNumber ) if numpy.sum(self.reference_count_on[pid]) ==NPBUNCHES and numpy.sum(self.reference_count_off[pid])==NPBUNCHES : self.totalfree(pid,packetSerialNumber) def totalfree(self,pid,packetSerialNumber,notused=0): del self.reference_count_on [pid] del self.reference_count_off[pid] # po=processingObject pos = self.tokenperpacketid[pid] if myrank==0 and notused==0: with WriteLock: if P.VERBOSITY>0 : print( "PACKET N: " , packetSerialNumber," FREES " , self, " FROM SLOT " , pos) del self.tokenperpacketid[pid] self.ArrayStatus[pos] =ARRAYFREE self.PacketID[pos]= None self.packet_has_gone[pid]=1 class ARRAYFREE: pass class ARRAYINUSE: pass class ARRAYREADY: pass class MPI_Resources(ArrayResources): pass class PassCheckResources: def __init__(self, resourcetocheck ) : self.resourcetocheck = resourcetocheck def getResources(self, processingObject=None, packetSerialNumber=None , resourcename=None): global NPBUNCHES pid = packetSerialNumber // NPBUNCHES pid_i = packetSerialNumber % NPBUNCHES if pid in self.resourcetocheck.packet_has_gone: return 1 else: return 0 def freeResources(self, processingObject=None, packetSerialNumber=None, notused=0 ): pass class PacketsManager: def __init__(self, npackets=None, nslots=100): if nslots<(npackets+1)*NPBUNCHES: nslots=(npackets+1)*NPBUNCHES self.Nslots=nslots self.Npackets = npackets*NPBUNCHES self.PacketStatus_s = [None]*self.Nslots self.ProcessingObjects = [None]*self.Nslots self.packetsSerialNumber = [None]*self.Nslots self.packetcount=0 self.posbyserial={} def packethasdone(self, sn ): pos=self.posbyserial[sn] self.PacketStatus_s[pos] = PACKETDONE def CheckAssignPackets_getppn2operate(self, olddump=[""], oldN=[0] ): if(0): # for debugging inspection of states lvs=[] for o in self.ProcessingObjects: if o is None: lvs.append(None) else: lvs.append(o.processingLevel) tosave={} tosave["packetsSerialNumber"] = self.packetsSerialNumber tosave["PacketStatus_s"] = self.PacketStatus_s tosave["posbyserial"] = self.posbyserial tosave["ProcessingLevels"] = lvs tosave["packetcount"] = self.packetcount tosave["NPBUNCHES"] = NPBUNCHES tosave["Npackets"] = self.Npackets s=pickle.dumps(tosave) if s!=olddump[0]: nome = "status_"+str(myrank)+"_"+str(oldN[0]) olddump[0]=s oldN[0]=oldN[0]+1 pickle.dump(tosave,open(nome,"w")) nome = "status_"+str(myrank) pickle.dump(tosave,open(nome,"w")) for pos in range( self.Nslots ): if self.PacketStatus_s[pos] == PACKETDONE: po = self.ProcessingObjects[pos] sn = self.packetsSerialNumber[pos] if po.iscollapsed( sn ): del self.posbyserial[sn] self.packetsSerialNumber[pos] = None self.ProcessingObjects [pos] = None self.PacketStatus_s [pos] = None if self.packetcount< self.Npackets: pos = self.PacketStatus_s.index(None) if pos!=-1: # per start le cose sono semplici self.PacketStatus_s [pos]=PACKETDONE self.ProcessingObjects[pos]=self.start self.packetsSerialNumber[pos]=self.packetcount self.posbyserial[self.packetcount]=pos if myrank==0: with WriteLock: if P.VERBOSITY>0 : print( "STARTING OF PACKET N:" , self.packetcount) self.packetcount+=1 # cerca fra quelli che che sono DONE # fra questi priorita al serialnumber piu basso # Se questi non supera di livello il serialseguente # Se lo supera passare al serialN subito piu alto dones=[] poss =[] for tok, serial in zip( self.PacketStatus_s , self.packetsSerialNumber) : if tok==PACKETDONE: dones.append(serial) if tok is not None: poss.append(serial) # verifica di consistenza messa a giorno di self.posbyserial if( len(poss)!= len(self.posbyserial.keys() ) ): raise Exception( " Inconsistency " ) if( self.packetcount==(self.Npackets) and len(poss)==0 ): # c'e' la possibilita' che si finisca return None dones.sort() for sn in dones: pos=self.posbyserial[sn] old_pO = self.ProcessingObjects[pos] pool = old_pO.NEXTPOOL old_npl = old_pO.processingLevel for next_pO in pool: npl = next_pO.processingLevel passabile=1 if(npl>0): # sui positivi non deve arrivare prima # di sn minori for serial,pO,pOstatus in zip(self.packetsSerialNumber, self.ProcessingObjects,self.PacketStatus_s): if ( serial is not None and pO is not None ): if serial//NPBUNCHES < sn//NPBUNCHES: if npl>= abs(pO.processingLevel): passabile=0 break if serial//NPBUNCHES == sn//NPBUNCHES: if old_npl+10 : print( "ENDED PROCESSING PACKET N: " , self.packetSerialNumber , " WITH PROCESSING OBJECT :", self.processingObject) class PACKETONPROCESS: pass class PACKETDONE: pass class ProcessingBaseClass: num_of_algorithms=1 collapse=0 def __init__(self, processingLevel=0, choosedAlgorithm=0 , resources={}, resources_free=[], pmanager=None, STEAM_DIRECTION=1): self.STEAM_DIRECTION=STEAM_DIRECTION self.pmanager=pmanager assert(type(resources)==type({})) assert(type(resources_free)==type([])) self.choosedAlgorithm=choosedAlgorithm self.processingLevel = processingLevel self.resources = resources self.resources_free = resources_free if self.choosedAlgorithm>=self.num_of_algorithms: raise " self.choosedAlgorithm>=self.num_of_algorithms " self.NEXTPOOL=None self.requiredCpusGpus = numpy.array( [ 0,0 ]) def get_resources(self, sn ): # chiama pr tutte le resource nel dizionario self.resources # il metodo get resources passando anche self # perche ci sono le requiredCpusGpus da interrogare # o altre cose a seconda dei casi # Passare anche sn per che l' oggetto chiamato # tenga traccia, in modo da poter in seguito cancellare ## pr " IN GET_RESOURCES PER ", sn, self allocated=[] for resourcename in self.resources.keys(): resource = self.resources[resourcename] if resource.getResources(processingObject =self, packetSerialNumber =sn, resourcename=resourcename): allocated.append( resource) else: # print( " PACCHETTO ", sn, " FALLISCE SU RISORSA ", resourcename, " per oggetto ", self) for tok in allocated: tok.freeResources(self,sn,notused=1) return 0 return 1 def free_resources(self, sn ): for resourcename in self.resources_free: if myrank==0: with WriteLock: if P.VERBOSITY>0 : print( "PACKET N: " , sn, " FREES " , resourcename, " da oggetto " , self) resource = self.resources[resourcename] resource.freeResources( processingObject = self , packetSerialNumber = sn ) self.pmanager.packethasdone( sn ) def iscollapsed(self, sn): global NPBUNCHES if self.collapse: return True return False def processSpecial(self,psn): pass # pri " Niente da fare per pacchetto No : " , psn def process(self,psn ): self.processSpecial(psn) # derived class with GLobalLock: self.free_resources( psn ) pass class ProcessingReading(ProcessingBaseClass): num_of_algorithms=1 def __init__(self, processingLevel=0, choosedAlgorithm=0 , resources={}, resources_free=[], pmanager=None, STEAM_DIRECTION=1): self.pmanager=pmanager ProcessingBaseClass.__init__(self, processingLevel, choosedAlgorithm, resources, resources_free, pmanager=pmanager , STEAM_DIRECTION = STEAM_DIRECTION ) ## ----------------- data for specific requirement. Resources know their name ----------------------------- self.requiredCpusGpus[:] = numpy.array( [0,0 ]) def processSpecial(self,psn ): if self.choosedAlgorithm==0: self.algo_0(psn ) else: raise " self.choosedAlgorithm unknown " def algo_0(self, psn ): global P,C, NPBUNCHES ## P=Parameters ## C=cspaceA ntokraw = self.resources["RawDataResources"].tokenperpacketid[ psn // NPBUNCHES] ntokraw = self.resources["RawDataResources"].tokenperpacketid[ psn // NPBUNCHES] ncpus , mygpus = self.resources["SiliciumResources"].resourcesForPacket[psn] if P.VERBOSITY>0 : print( " leggo >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>" , psn, ntokraw, NPBUNCHES) with WriteLock: C.read_chunk(psn, ntokraw, NPBUNCHES, self.STEAM_DIRECTION) if P.VERBOSITY>0 : print( " OK <<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<" ) class Processing_PreProcessing(ProcessingBaseClass): num_of_algorithms=1 def __init__(self, processingLevel=0, choosedAlgorithm=0 , resources={}, resources_free=[], pmanager=None,STEAM_DIRECTION=1): global P self.pmanager=pmanager ProcessingBaseClass.__init__(self, processingLevel, choosedAlgorithm, resources, resources_free, pmanager=pmanager , STEAM_DIRECTION = STEAM_DIRECTION ) ## ----------------- data for specific requirement. Resources know their name ----------------------------- self.requiredCpusGpus = [ [1*P.ncpus_expansion ] , 0 ] def processSpecial(self,psn ): if self.choosedAlgorithm==0: self.algo_0(psn ) else: raise " self.choosedAlgorithm unknown " def algo_0(self, psn ): global P,C, NPBUNCHES ## P=Parameters ## C=cspaceA ntokraw = self.resources["RawDataResources"].tokenperpacketid[ psn // NPBUNCHES] ntoktreated = self.resources["TreatedDataResources"].tokenperpacketid[ psn // NPBUNCHES] ncpus , mygpus = self.resources["SiliciumResources"].resourcesForPacket[psn] if P.VERBOSITY>0 : print( " chiamo C.preprocess_chunk" , psn, ntokraw, ntoktreated, NPBUNCHES, ncpus) C.preprocess_chunk(psn, ntokraw, ntoktreated, NPBUNCHES, ncpus// P.ncpus_expansion, self.STEAM_DIRECTION) class Processing_InterleavedReadingPreProcessing(ProcessingBaseClass): num_of_algorithms=1 def __init__(self, processingLevel=0, choosedAlgorithm=0 , resources={}, resources_free=[], pmanager=None,STEAM_DIRECTION=1, FF2=0): global P self.pmanager=pmanager self.FF2 = FF2 ProcessingBaseClass.__init__(self, processingLevel, choosedAlgorithm, resources, resources_free, pmanager=pmanager, STEAM_DIRECTION = STEAM_DIRECTION ) ## ----------------- data for specific requirement. Resources know their name ----------------------------- self.requiredCpusGpus = [ [1*P.ncpus_expansion ] , 0 ] def processSpecial(self,psn ): if self.choosedAlgorithm==0: self.algo_0(psn ) else: raise " self.choosedAlgorithm unknown " def algo_0(self, psn ): global P,C, NPBUNCHES ## P=Parameters ## C=cspaceA ntokraw = self.resources["RawDataResources"].tokenperpacketid[ psn // NPBUNCHES] ntoktreated = self.resources["TreatedDataResources"].tokenperpacketid[ psn // NPBUNCHES] ncpus , mygpus = self.resources["SiliciumResources"].resourcesForPacket[psn] if P.VERBOSITY>0 : print( " chiamo C.preprocess_chunk" , psn, ntokraw, ntoktreated, NPBUNCHES, ncpus) C.InterleavedReadingPreProcessing_chunk(psn, ntokraw, ntoktreated, NPBUNCHES, ncpus//P.ncpus_expansion, self.STEAM_DIRECTION, self.FF2) class Processing_Transposing(ProcessingBaseClass): num_of_algorithms=1 def __init__(self, processingLevel=0, choosedAlgorithm=0 , resources={}, resources_free=[], pmanager=None,STEAM_DIRECTION=1, Ctr = None): self.pmanager=pmanager self.pid_inprocess=-1 self.nperpid =0 self.nperpid_tr =0 self.Ctr = Ctr ProcessingBaseClass.__init__(self, processingLevel, choosedAlgorithm, resources, resources_free, pmanager=pmanager, STEAM_DIRECTION = STEAM_DIRECTION ) ## ----------------- data for specific requirement. Resources know their name ----------------------------- self.requiredCpusGpus = [ 0 , 0 ] self.event = threading.Event() self.event_tr = threading.Event() self.semaphore = threading.Semaphore() def processSpecial(self,psn ): if self.choosedAlgorithm==0: self.algo_0(psn ) else: raise " self.choosedAlgorithm unknown " def algo_0(self, psn ): global P,C, NPBUNCHES ## P=Parameters ## C=cspaceA pid = psn // NPBUNCHES pid_i = psn % NPBUNCHES with self.semaphore: if not self.pid_inprocess== pid: self.pid_inprocess = pid self.nperpid =0 self.nperpid_tr = 0 if P.STEAM_INVERTER==0 : with self.semaphore: self.nperpid_tr += 1 if self.nperpid_tr == NPBUNCHES : self.Ctr.Aspace["transposeddatatokens"][0][:]=0 self.event_tr.set() else: self.event_tr.clear() self.event_tr.wait() with self.semaphore: self.nperpid += 1 if(self.nperpid) == NPBUNCHES: ntoktreated = self.resources["TreatedDataResources"] .tokenperpacketid[pid] ntoktransposed = self.resources["TransposedDataResources"].tokenperpacketid[pid] C.transpose_chunk(psn, ntoktreated, ntoktransposed, NPBUNCHES, self.STEAM_DIRECTION ) self.event.set() else: self.event.clear() self.event.wait() class Processing_Dispenser(ProcessingBaseClass): num_of_algorithms=1 def __init__(self, processingLevel=0, choosedAlgorithm=0 , resources={}, resources_free=[], pmanager=None,STEAM_DIRECTION=1): self.pmanager=pmanager self.pid_inprocess=-1 self.nperpid =0 ProcessingBaseClass.__init__(self, processingLevel, choosedAlgorithm, resources, resources_free, pmanager=pmanager, STEAM_DIRECTION = STEAM_DIRECTION ) ## ----------------- data for specific requirement. Resources know their name ----------------------------- self.requiredCpusGpus = [ 0 , 0 ] def processSpecial(self,psn ): if self.choosedAlgorithm==0: self.algo_0(psn ) else: raise " self.choosedAlgorithm unknown " def algo_0(self, psn ): global P,C, NPBUNCHES ## P=Parameters ## C=cspaceA pid = psn // NPBUNCHES pid_i = psn % NPBUNCHES if not self.pid_inprocess== pid: self.pid_inprocess = pid self.nperpid =0 self.nperpid += 1 if(self.nperpid) == NPBUNCHES: ntoktransposed = self.resources["TransposedDataResources"].tokenperpacketid[pid] C.dispense_chunk(psn, ntoktransposed, NPBUNCHES, self.STEAM_DIRECTION ) def process(self,psn ): with GLobalLock: self.free_resources( psn ) ## deve permettere ai pacchetti di continuare e rifornirli in volo self.processSpecial(psn) pass class Processing_Reconstructing(ProcessingBaseClass): num_of_algorithms=1 def __init__(self, processingLevel=0, choosedAlgorithm=0 , resources={}, resources_free=[], pmanager=None,STEAM_DIRECTION=1): global P self.pmanager=pmanager self.pid_inprocess=-1 self.nperpid =0 ProcessingBaseClass.__init__(self, processingLevel, choosedAlgorithm, resources, resources_free, pmanager=pmanager, STEAM_DIRECTION = STEAM_DIRECTION ) ## ----------------- data for specific requirement. Resources know their name ----------------------------- self.requiredCpusGpus = [ 1 , 0 ] def processSpecial(self,psn ): if self.choosedAlgorithm==0: self.algo_0(psn ) else: raise " self.choosedAlgorithm unknown " def algo_0(self, psn ): global P,C, NPBUNCHES ## P=Parameters ## C=cspaceA pid = psn // NPBUNCHES pid_i = psn % NPBUNCHES ncpus , mygpus = self.resources["SiliciumResources"].resourcesForPacket[psn] C.reconstruct(psn, NPBUNCHES , ncpus, self.STEAM_DIRECTION ) class Processing_ReconstructingSHARED(ProcessingBaseClass): num_of_algorithms=1 def __init__(self, processingLevel=0, choosedAlgorithm=0 , resources={}, resources_free=[], pmanager=None,STEAM_DIRECTION=1): global P self.pmanager=pmanager self.pid_inprocess=-1 self.nperpid =0 ProcessingBaseClass.__init__(self, processingLevel, choosedAlgorithm, resources, resources_free, pmanager=pmanager, STEAM_DIRECTION = STEAM_DIRECTION ) ## ----------------- data for specific requirement. Resources know their name ----------------------------- self.requiredCpusGpus = [ 1 , 0 ] def processSpecial(self,psn ): if self.choosedAlgorithm==0: self.algo_0(psn ) else: raise " self.choosedAlgorithm unknown " def algo_0(self, psn ): global P,C, NPBUNCHES ## P=Parameters ## C=cspaceA pid = psn // NPBUNCHES pid_i = psn % NPBUNCHES ncpus , mygpus = self.resources["SiliciumResources"].resourcesForPacket[psn] ntoktransposed = self.resources["TransposedDataResources"].tokenperpacketid[pid] C.reconstructSHARED(psn, NPBUNCHES , ntoktransposed, ncpus, self.STEAM_DIRECTION ) class ProcessingStart(ProcessingBaseClass): num_of_algorithms=1 def __init__(self, processingLevel=0 ,choosedAlgorithm=0, resources={}, resources_free=[], pmanager=None,STEAM_DIRECTION=1): self.pmanager=pmanager ProcessingBaseClass.__init__(self, processingLevel, choosedAlgorithm, resources, resources_free, pmanager=pmanager, STEAM_DIRECTION = STEAM_DIRECTION ) self.requiredCpusGpus[:] = numpy.array( [0,0 ]) def processSpecial(self,*pars): if self.choosedAlgorithm==0: self.algo_0(*pars) else: raise " self.choosedAlgorithm unknown " def algo_0(self, *pars): pass