B è?F[YCã@sêddlZddlZddlZddlZddlZddlZddlmZddlZddl m Z m Z m Z m Z mZddlmZmZmZeƒZe d¡ZGdd„deƒZGdd „d eƒZGd d „d ejƒZGd d „d eƒZGdd„deƒZGdd„deƒZdS)éN)ÚQueue)ÚDEFAULT_PART_SIZEÚminimum_part_sizeÚ chunk_hashesÚ tree_hashÚ bytes_to_hex)ÚUploadArchiveErrorÚDownloadArchiveErrorÚTreeHashDoesNotMatchErrorzboto.glacier.concurrentc@s2eZdZedfdd„Zdd„Zdd„Zdd „Zd S) ÚConcurrentTransfereré cCs||_||_g|_dS)N)Ú _part_sizeÚ _num_threadsÚ_threads)ÚselfÚ part_sizeÚ num_threads©rú6lib/python3.7/site-packages/boto/glacier/concurrent.pyÚ__init__+szConcurrentTransferer.__init__cCsLt|ƒ}|j|kr|j}n|}t d|j|¡tt |t|ƒ¡ƒ}||fS)NzfThe part size specified (%s) is smaller than the minimum required part size. Using a part size of: %s)rr ÚlogÚdebugÚintÚmathZceilÚfloat)rÚ total_sizeZmin_part_size_requiredrÚ total_partsrrrÚ_calculate_required_part_size0s  z2ConcurrentTransferer._calculate_required_part_sizecCsFt d¡x|jD] }d|_qWx|jD] }| ¡q(Wt d¡dS)NzShutting down threads.FzThreads have exited.)rrrÚshould_continueÚjoin)rÚthreadrrrÚ_shutdown_threads<s      z&ConcurrentTransferer._shutdown_threadscCsLt d¡xt|ƒD]}| ||f¡qWxt|jƒD]}| t¡q6WdS)NzAdding work items to queue.)rrÚrangeÚputrÚ _END_SENTINEL)rrÚ worker_queuerÚirrrÚ_add_work_items_to_queueDs  z-ConcurrentTransferer._add_work_items_to_queueN)Ú__name__Ú __module__Ú __qualname__rrrr!r'rrrrr *s r cs@eZdZdZedf‡fdd„ Zd dd„Zdd „Zd d „Z‡Z S) ÚConcurrentUploaderaConcurrently upload an archive to glacier. This class uses a thread pool to concurrently upload an archive to glacier using the multipart upload API. The threadpool is completely managed by this class and is transparent to the users of this class. r cs"tt|ƒ ||¡||_||_dS)au :type api: :class:`boto.glacier.layer1.Layer1` :param api: A layer1 glacier object. :type vault_name: str :param vault_name: The name of the vault. :type part_size: int :param part_size: The size, in bytes, of the chunks to use when uploading the archive parts. The part size must be a megabyte multiplied by a power of two. :type num_threads: int :param num_threads: The number of threads to spawn for the thread pool. The number of threads will control how much parts are being concurrently uploaded. N)Úsuperr+rÚ_apiÚ _vault_name)rÚapiÚ vault_namerr)Ú __class__rrrVszConcurrentUploader.__init__Nc Csôt |¡j}| |¡\}}dg|}tƒ}tƒ}|j |j||¡} | d} | |||¡|  || ||¡y|  |||¡Wn@t k rº} z"t   d¡|j |j| ¡| ‚Wdd} ~ XYnXt   d¡|j |j| tt|ƒƒ|¡} t   d¡| dS)a^Concurrently create an archive. The part_size value specified when the class was constructed will be used *unless* it is smaller than the minimum required part size needed for the size of the given file. In that case, the part size used will be the minimum part size required to properly upload the given file. :type file: str :param file: The filename to upload :type description: str :param description: The description of the archive. :rtype: str :return: The archive id of the newly created archive. NZUploadIdzHAn error occurred while uploading an archive, aborting multipart upload.zCompleting upload.zUpload finished.Z ArchiveId)ÚosÚstatÚst_sizerrr-Zinitiate_multipart_uploadr.r'Ú_start_upload_threadsÚ_wait_for_upload_threadsrrrZabort_multipart_uploadZcomplete_multipart_uploadrr) rÚfilenameZ descriptionrrrÚ hash_chunksr%Ú result_queueÚresponseÚ upload_idÚerrrÚuploadns2       zConcurrentUploader.uploadcCs`xRt|ƒD]F}| ¡}t|tƒr@t d|¡| ¡td|ƒ‚|\}}|||<q W| ¡dS)Nz?An error was found in the result queue, terminating threads: %sz0An error occurred while uploading an archive: %s)r"ÚgetÚ isinstanceÚ Exceptionrrr!r)rr8r9rÚ_ÚresultÚ part_numberZ tree_sha256rrrr6 s  z+ConcurrentUploader._wait_for_upload_threadscCsVt d¡xFt|jƒD]8}t|j|j||||ƒ}t d¡|  ¡|j   |¡qWdS)NzStarting threads.gš™™™™™É?) rrr"rÚUploadWorkerThreadr-r.ÚtimeÚsleepÚstartrÚappend)rr9r;r%r7rAr rrrr5°s    z(ConcurrentUploader._start_upload_threads)N) r(r)r*Ú__doc__rrr=r6r5Ú __classcell__rr)r1rr+Ls   2r+cs4eZdZ‡fdd„Zdd„Zdd„Zdd„Z‡ZS) ÚTransferThreadcs$tt|ƒ ¡||_||_d|_dS)NT)r,rKrÚ _worker_queueÚ _result_queuer)rr%r9)r1rrr¼szTransferThread.__init__cCslx^|jr^y|jjdd}Wntk r0wYnX|tkrF| ¡dS| |¡}|j |¡qW| ¡dS)Né)Ztimeout) rrLr>ZEmptyr$Ú_cleanupÚ_process_chunkrMr#)rÚworkrBrrrÚrunÄs zTransferThread.runcCsdS)Nr)rrQrrrrPÑszTransferThread._process_chunkcCsdS)Nr)rrrrrOÔszTransferThread._cleanup)r(r)r*rrRrPrOrJrr)r1rrK»s  rKcs<eZdZddef‡fdd„ Zdd„Zdd„Zdd „Z‡ZS) rDéc sLtt|ƒ ||¡||_||_||_t|dƒ|_||_||_ ||_ | |_ dS)NÚrb) r,rDrr-r.Ú _filenameÚopenÚ_fileobjÚ _upload_idÚ _num_retriesÚ_time_between_retriesÚ_retry_exceptions) rr/r0r7r;r%r9Ú num_retriesÚtime_between_retriesÚretry_exceptions)r1rrrÙs zUploadWorkerThread.__init__cCs’d}xˆt|jdƒD]v}y| |¡}PWq|jk rˆ}z@t d|d|j|d|jd|j|j|¡t   |j ¡|}Wdd}~XYqXqW|S)NrNzpException caught uploading part number %s for vault %s, attempt: (%s / %s), filename: %s, exception: %s, msg: %sr) r"rYÚ _upload_chunkr[rÚerrorr.rUr1rErFrZ)rrQrBr&r<rrrrPçs  z!UploadWorkerThread._process_chunkc Cs’|\}}||}|j |¡|j |¡}t |¡ ¡}tt|ƒƒ}||t|ƒdf}t   d||¡|j   |j |j|t|ƒ||¡} |  ¡||fS)NrNzUploading chunk %s of size %s)rWÚseekÚreadÚhashlibZsha256Z hexdigestrrÚlenrrr-Z upload_partr.rXr) rrQrCrÚ start_byteÚcontentsZ linear_hashZtree_hash_bytesÚ byte_ranger:rrrr_÷s   z UploadWorkerThread._upload_chunkcCs|j ¡dS)N)rWÚclose)rrrrrOszUploadWorkerThread._cleanup) r(r)r*r@rrPr_rOrJrr)r1rrDØs  rDcs>eZdZdZedf‡fdd„ Zdd„Zdd„Zd d „Z‡Z S) ÚConcurrentDownloaderzý Concurrently download an archive from glacier. This class uses a thread pool to concurrently download an archive from glacier. The threadpool is completely managed by this class and is transparent to the users of this class. r cstt|ƒ ||¡||_dS)a :param job: A layer2 job object for archive retrieval object. :param part_size: The size, in bytes, of the chunks to use when uploading the archive parts. The part size must be a megabyte multiplied by a power of two. N)r,rirÚ_job)rÚjobrr)r1rrrs zConcurrentDownloader.__init__c Cs|jj}| |¡\}}tƒ}tƒ}| |||¡| ||¡y| |||¡Wn2tk r€}zt  d|¡|‚Wdd}~XYnXt  d¡dS)z’ Concurrently download an archive. :param filename: The filename to download the archive to :type filename: str z2An error occurred while downloading an archive: %sNzDownload completed.) rjZ archive_sizerrr'Ú_start_download_threadsÚ_wait_for_download_threadsr rr)rr7rrrr%r9r<rrrÚdownload$s  zConcurrentDownloader.downloadc Csîdg|}t|dƒ‚}xzt|ƒD]n}| ¡}t|tƒrVt d|¡| ¡td|ƒ‚|\}} } } | ||<|| } |  | ¡|  | ¡|  ¡q WWdQRXt t |ƒƒ} t d|jj| ¡|jj| krâ| ¡td|jj| fƒ‚| ¡dS)a Waits until the result_queue is filled with all the downloaded parts This indicates that all part downloads have completed Saves downloaded parts into filename :param filename: :param result_queue: :param total_parts: NÚwbz?An error was found in the result queue, terminating threads: %sz0An error occurred while uploading an archive: %sz?Verifying final tree hash of archive, expecting: %s, actual: %szBTree hash for entire archive does not match, expected: %s, got: %s)rVr"r>r?r@rrr!r raÚwriteÚflushrrrjZsha256_treehashr )rr7r9rr8ÚfrArBrCrÚ actual_hashÚdatareZ final_hashrrrrm9s6          z/ConcurrentDownloader._wait_for_download_threadscCsNt d¡x>t|jƒD]0}t|j||ƒ}t d¡| ¡|j   |¡qWdS)NzStarting threads.gš™™™™™É?) rrr"rÚDownloadWorkerThreadrjrErFrGrrH)rr9r%rAr rrrrl`s   z,ConcurrentDownloader._start_download_threads) r(r)r*rIrrrnrmrlrJrr)r1rri s   'rics4eZdZddef‡fdd„ Zdd„Zdd„Z‡ZS)rurScs.tt|ƒ ||¡||_||_||_||_dS)a  Individual download thread that will download parts of the file from Glacier. Parts to download stored in work queue. Parts download to a temp dir with each part a separate file :param job: Glacier job object :param work_queue: A queue of tuples which include the part_number and part_size :param result_queue: A priority queue of tuples which include the part_number and the path to the temp file that holds that part's data. N)r,rurrjrYrZr[)rrkr%r9r\r]r^)r1rrrjs zDownloadWorkerThread.__init__c Csvd}xlt|jƒD]^}y| |¡}PWq|jk rl}z(t d|d|j¡t |j ¡|}Wdd}~XYqXqW|S)z‹ Attempt to download a part of the archive from Glacier Store the result in the result_queue :param work: Nz6Exception caught downloading part number %s for job %sr) r"rYÚ_download_chunkr[rr`rjrErFrZ)rrQrBrAr<rrrrPƒs  z#DownloadWorkerThread._process_chunkc Cs†|\}}||}|||df}t d||¡|j |¡}| ¡}ttt|ƒƒƒ}|d|krttd||d|fƒ‚||t   |¡|fS)z§ Downloads a chunk of archive from Glacier. Saves the data to a temp file Returns the part number and temp file location :param work: rNzDownloading chunk %s of size %sZTreeHashzBTree hash for part number %s does not match, expected: %s, got: %s) rrrjZ get_outputrbrrrr ÚbinasciiZ unhexlify) rrQrCrrergr:rtrsrrrrv–s   z$DownloadWorkerThread._download_chunk)r(r)r*r@rrPrvrJrr)r1rruis ru)r2rZ threadingrcrEZloggingZ boto.compatrrwZboto.glacier.utilsrrrrrZboto.glacier.exceptionsrr r Úobjectr$Z getLoggerrr r+ZThreadrKrDrirurrrrÚs"  "o4]