code: def run(self, timeout=None):
          
          query, link_results_flag = self._normalize_query()
  
          try:
              result = self._client.mapred(self._inputs, query, timeout)
          except riak.RiakError as e:
              if 'worker_startup_failed' in e.value:
                  for phase in self._phases:
                      if phase._language == 'erlang':
                          if type(phase._function) is str:
                              raise riak.RiakError(
                                      'May have tried erlang strfun '
                                      'when not allowed\n'
                                      'original error: ' + e.value)
              raise e
  
          # If the last phase is NOT a link phase, then return the result.
          if not (link_results_flag or
                  isinstance(self._phases[-1], RiakLinkPhase)):
              return result
  
          # If there are no results, then return an empty list.
          if result is None:
              return []
  
          # Otherwise, if the last phase IS a link phase, then convert the
          # results to link tuples.
          a = []
          for r in result:
              if (len(r) == 2):
                  link = (r[0], r[1], None)
              elif (len(r) == 3):
                  link = (r[0], r[1], r[2])
              a.append(link)
  
          return a
docs: Run the map/reduce operation synchronously. Returns a list of
          results, or a list of links if the last phase is a link phase.
          Shortcut for :meth:`riak.client.RiakClient.mapred`.
  
          :param timeout: Timeout in milliseconds
          :type timeout: integer, None
          :rtype: list