Commit 4cd1f08b authored by Cristiano Urban's avatar Cristiano Urban
Browse files

Added variables to track the progress of the job + some code cleanup. 



Signed-off-by: default avatarCristiano Urban <cristiano.urban@inaf.it>
parent b13d92a1

transfer_service/retrieve_executor.py

+10 −14
Original line number Diff line number Diff line
@@ -40,7 +40,9 @@ class RetrieveExecutor(TaskExecutor): 
        self.jobId = None

        self.nodeList = []

        self.fileList = []

        self.numBlocks = None        

        self.numBlocks = 0

        self.procBlocks = 0

        self.totalSize = 0

        super(RetrieveExecutor, self).__init__()

        

    def  buildFileList(self):
@@ -98,6 +100,7 @@ class RetrieveExecutor(TaskExecutor): 
            blockSize = 0

        for fileInfo in self.fileList:

            fileSize = fileInfo["fileSize"]

            self.totalSize += fileSize

            # controlla se il file ha dimensione superiore alla dimensione di un blocco

            if fileSize > self.maxBlockSize:

                # chiudi il blocco corrente, se non vuoto, altrimenti usa quello?
@@ -107,15 +110,9 @@ class RetrieveExecutor(TaskExecutor): 
                    blockIdx += 1

                    fileInfo["blockIdx"] = blockIdx

                    blockIdx += 1

                    #self.blockList.append(currentBlock.copy())

                    #currentBlock.clear()

                    #currentBlock.append(fileInfo.copy())

                else:

                    fileInfo["blockIdx"] = blockIdx

                    blockIdx += 1

                    #currentBlock.append(fileInfo.copy())

                    #self.blockList.append(currentBlock.copy())

                    #currentBlock.clear()

                blockSize = 0

            else:

                # il file sta dentro un blocco, quindi controlla se
@@ -124,7 +121,6 @@ class RetrieveExecutor(TaskExecutor): 
                if blockSize + fileSize <= self.maxBlockSize:

                    # se si, aggiungi il file al blocco e vai avanti col prossimo

                    fileInfo["blockIdx"] = blockIdx

                    #currentBlock.append(fileInfo.copy())

                    blockSize += fileSize

                else:

                    # se no, chiudi il blocco attuale e aggiungilo alla lista blocchi
@@ -132,13 +128,9 @@ class RetrieveExecutor(TaskExecutor): 
                    blockIdx += 1

                    fileInfo["blockIdx"] = blockIdx

                    blockSize = fileSize

                    #self.blockList.append(currentBlock.copy())

                    #currentBlock.clear()

                    #blockSize = 0

                    #currentBlock.append(fileInfo.copy())

                    #blockSize += fileSize

        if self.fileList:

            self.numBlocks = blockIdx + 1

            self.dbConn.setTotalBlocks(self.jobId, self.numBlocks)

        print(f"numBlocks = {self.numBlocks}")

        if os.path.exists("blocks.txt"):

            os.remove("blocks.txt")
@@ -185,6 +177,8 @@ class RetrieveExecutor(TaskExecutor): 
                self.tapeClient.disconnect()

                

            blockFileList.clear()

            self.procBlocks += 1

            self.dbConn.updateProcessedBlocks(self.jobId, self.procBlocks)

            

        return True
@@ -204,7 +198,9 @@ class RetrieveExecutor(TaskExecutor): 
        self.fileList.clear()

        self.nodeList.clear()

        self.storageType = None

        self.numBlocks = None

        self.numBlocks = 0

        self.procBlocks = 0

        self.totalSize = 0



    def run(self):

        print("Starting retrieve executor...")