downloader.py 文件源码

def shard_download(self, pointer, shard_index, bucket_id, file_id):
        self.__logger.debug('Beginning download proccess...')

        try:
            self.__logger.debug('Starting download threads...')
            self.__logger.debug('Downloading shard at index %s ...',
                                shard_index)

            url = 'http://{address}:{port}/shards/{hash}?token={token}'.format(
                address=pointer.get('farmer')['address'],
                port=str(pointer.get('farmer')['port']),
                hash=pointer['hash'],
                token=pointer['token'])
            self.__logger.debug(url)

            tp = ThreadPool(processes=1)
            async_result = tp.apply_async(
                self.retrieve_shard_file,
                (url, shard_index))  # tuple of args for foo
            shard = async_result.get(self.client.timeout)  # get the return value

            # shard = self.retrieve_shard_file(url, shard_index)
            self.__logger.debug('Shard downloaded')
            self.__logger.debug('Shard at index %s downloaded successfully',
                                shard_index)
            return shard

        except IOError as e:
            self.__logger.error('Perm error %s', e)
            if str(e) == str(13):
                self.__logger.error("""Error while saving or reading file or
                temporary file.
                Probably this is caused by insufficient permisions.
                Please check if you have permissions to write or
                read from selected directories.""")

        except TimeoutError:
            self.__logger.warning('Aborting shard %s download due to timeout' %
                                  shard_index)
            tp.terminate()
            self.__logger.warning('Try with a new pointer')
            new_pointer = self.client.file_pointers(
                bucket_id=bucket_id,
                file_id=file_id,
                limit='1',
                skip=str(shard_index),
                exclude=str([pointer['farmer']['nodeID']]))
            self.__logger.debug('Found new pointer')
            return self.shard_download(new_pointer[0], shard_index,
                                       bucket_id, file_id)

        except Exception as e:
            self.__logger.error(e)
            self.__logger.error('Unhandled')