docs: def run(self, pre_execute=True, concurrent_tasks=None, close=True, **kw):
          """
          Run the calculation and return the exported outputs.
          """
          with self._monitor:
              self._monitor.username = kw.get('username', '')
              self._monitor.hdf5 = self.datastore.hdf5
              if concurrent_tasks is None:  # use the job.ini parameter
                  ct = self.oqparam.concurrent_tasks
              else:  # used the parameter passed in the command-line
                  ct = concurrent_tasks
              if ct == 0:  # disable distribution temporarily
                  oq_distribute = os.environ.get('OQ_DISTRIBUTE')
                  os.environ['OQ_DISTRIBUTE'] = 'no'
              if ct != self.oqparam.concurrent_tasks:
                  # save the used concurrent_tasks
                  self.oqparam.concurrent_tasks = ct
              self.save_params(**kw)
              try:
                  if pre_execute:
                      self.pre_execute()
                  self.result = self.execute()
                  if self.result is not None:
                      self.post_execute(self.result)
                  self.before_export()
                  self.export(kw.get('exports', ''))
              except Exception:
                  if kw.get('pdb'):  # post-mortem debug
                      tb = sys.exc_info()[2]
                      traceback.print_tb(tb)
                      pdb.post_mortem(tb)
                  else:
                      logging.critical('', exc_info=True)
                      raise
              finally:
                  # cleanup globals
                  if ct == 0:  # restore OQ_DISTRIBUTE
                      if oq_distribute is None:  # was not set
                          del os.environ['OQ_DISTRIBUTE']
                      else:
                          os.environ['OQ_DISTRIBUTE'] = oq_distribute
                  readinput.pmap = None
                  readinput.exposure = None
                  readinput.gmfs = None
                  readinput.eids = None
                  self._monitor.flush()
  
                  if close:  # in the engine we close later
                      self.result = None
                      try:
                          self.datastore.close()
                      except (RuntimeError, ValueError):
                          # sometimes produces errors but they are difficult to
                          # reproduce
                          logging.warning('', exc_info=True)
  
          return getattr(self, 'exported', {})