docs: def _run(self):
          # type: (Uploader) -> None
          """Execute Uploader
          :param Uploader self: this
          """
          # mark start
          self._start_time = blobxfer.util.datetime_now()
          logger.info('blobxfer start time: {0}'.format(self._start_time))
          # check renames
          if not self._spec.sources.can_rename() and self._spec.options.rename:
              raise RuntimeError(
                  'cannot rename to specified destination with multiple sources')
          # initialize resume db if specified
          if self._general_options.resume_file is not None:
              self._resume = blobxfer.operations.resume.UploadResumeManager(
                  self._general_options.resume_file)
          # initialize MD5 processes
          if ((self._spec.options.store_file_properties.md5 or
               self._spec.skip_on.md5_match) and
                  self._general_options.concurrency.md5_processes > 0):
              self._md5_offload = blobxfer.operations.md5.LocalFileMd5Offload(
                  num_workers=self._general_options.concurrency.md5_processes)
              self._md5_offload.initialize_check_thread(
                  self._check_for_uploads_from_md5)
          # initialize crypto processes
          if (self._spec.options.rsa_public_key is not None and
                  self._general_options.concurrency.crypto_processes > 0):
              logger.warning(
                  'crypto offload for upload is not possible due to '
                  'sequential nature of {} and FullBlob encryption mode'.format(
                      blobxfer.models.crypto.EncryptionMetadata.
                      _ENCRYPTION_ALGORITHM)
              )
          # initialize worker threads
          self._initialize_disk_threads()
          self._initialize_transfer_threads()
          # initialize local counters
          files_processed = 0
          skipped_files = 0
          skipped_size = 0
          approx_total_bytes = 0
          # iterate through source paths to upload
          seen = set()
          for src in self._spec.sources.files(self._general_options.dry_run):
              # create a destination array for the source
              dest = [
                  (sa, ase) for sa, ase in
                  self._generate_destination_for_source(src)
              ]
              for action, lp, ase in self._vectorize_and_bind(src, dest):
                  dest_id = blobxfer.operations.upload.Uploader.\
                      create_destination_id(ase._client, ase.container, ase.name)
                  if dest_id in seen:
                      raise RuntimeError(
                          'duplicate destination entity detected: {}/{}'.format(
                              ase._client.primary_endpoint, ase.path))
                  seen.add(dest_id)
                  if self._spec.options.delete_extraneous_destination:
                      self._delete_exclude.add(dest_id)
                  files_processed += 1
                  if action == UploadAction.Skip:
                      skipped_files += 1
                      skipped_size += ase.size if ase.size is not None else 0
                      if self._general_options.dry_run:
                          logger.info('[DRY RUN] skipping: {} -> {}'.format(
                              lp.absolute_path, ase.path))
                      continue
                  approx_total_bytes += lp.size
                  if blobxfer.util.is_not_empty(ase.replica_targets):
                      approx_total_bytes += lp.size * len(ase.replica_targets)
                  # add to potential upload set
                  uid = blobxfer.operations.upload.Uploader.create_unique_id(
                      lp, ase)
                  with self._upload_lock:
                      self._upload_set.add(uid)
                      self._upload_total += 1
                  if action == UploadAction.CheckMd5:
                      self._pre_md5_skip_on_check(lp, ase)
                  elif action == UploadAction.Upload:
                      if self._general_options.dry_run:
                          logger.info('[DRY RUN] upload: {} -> {}'.format(
                              lp.absolute_path, ase.path))
                          with self._upload_lock:
                              self._upload_set.remove(uid)
                              self._upload_total -= 1
                      else:
                          self._add_to_upload_queue(lp, ase, uid)
          del seen
          # set remote files processed
          with self._md5_meta_lock:
              self._all_files_processed = True
          with self._upload_lock:
              upload_size_mib = approx_total_bytes / blobxfer.util.MEGABYTE
              logger.debug(
                  ('{0} files {1:.4f} MiB filesize and/or lmt_ge '
                   'skipped').format(
                      skipped_files, skipped_size / blobxfer.util.MEGABYTE))
              logger.debug(
                  ('{0} local files processed, waiting for upload '
                   'completion of approx. {1:.4f} MiB').format(
                       files_processed, upload_size_mib))
          del files_processed
          del skipped_files
          del skipped_size
          del upload_size_mib
          del approx_total_bytes
          # wait for uploads to complete
          self._wait_for_disk_threads(terminate=False)
          self._wait_for_transfer_threads(terminate=False)
          end_time = blobxfer.util.datetime_now()
          # update progress bar
          self._update_progress_bar()
          # check for exceptions
          if len(self._exceptions) > 0:
              logger.error('exceptions encountered while uploading')
              # raise the first one
              raise self._exceptions[0]
          # check for mismatches
          if (self._upload_sofar != self._upload_total or
                  self._upload_bytes_sofar != self._upload_bytes_total):
              raise RuntimeError(
                  'upload mismatch: [count={}/{} bytes={}/{}]'.format(
                      self._upload_sofar, self._upload_total,
                      self._upload_bytes_sofar, self._upload_bytes_total))
          # delete all remaining local files not accounted for if
          # delete extraneous enabled
          self._delete_extraneous_files()
          # delete resume file if we've gotten this far
          if self._resume is not None:
              self._resume.delete()
          # output throughput
          if self._upload_start_time is not None:
              ultime = (end_time - self._upload_start_time).total_seconds()
              if ultime == 0:  # noqa
                  ultime = 1e-9
              mibup = self._upload_bytes_total / blobxfer.util.MEGABYTE
              mibps = mibup / ultime
              logger.info(
                  ('elapsed upload + verify time and throughput of {0:.4f} '
                   'GiB: {1:.3f} sec, {2:.4f} Mbps ({3:.3f} MiB/s)').format(
                       mibup / 1024, ultime, mibps * 8, mibps))
          end_time = blobxfer.util.datetime_now()
          logger.info('blobxfer end time: {0} (elapsed: {1:.3f} sec)'.format(
              end_time, (end_time - self._start_time).total_seconds()))