docs: def reduce(self, target_map, target_reduce, threads=0):
          """map/reduce this query among a bunch of processes
  
          :param target_map: callable, this function will be called once for each 
              row this query pulls out of the db, if you want something about the row
              to be seen by the target_reduce function return that value from this function
              and it will be queued for the target_reduce function to process it
          :param target_reduce: callable, this function will be called for any non 
              None value that the target_map function returns
          :param threads: integer, if not passed in this will be pegged to how many
              cpus python detects, which is almost always what you want
          """
          if not threads:
              threads = multiprocessing.cpu_count()
  
          # we subtract one for the main process
          map_threads = threads - 1 if threads > 1 else 1
  
          q = self.copy()
          limit = q.bounds.limit
          offset = q.bounds.offset
  
          total_count = limit if limit else q.count()
          limit_count = int(math.ceil(float(total_count) / float(map_threads)))
          logger.info("{} processes will handle {} rows each for a total of {}".format(
              map_threads,
              limit_count,
              total_count
          ))
  
          queue = multiprocessing.JoinableQueue()
  
          # close all open db global connections just in case, because we can't be sure
          # what the target_map methods are going to do, we want them to re-open connections
          # that they need
          interfaces = get_interfaces()
          for name, inter in interfaces.items():
              inter.close()
  
          # just in case we also close the query connection since it can in theory
          # be non-global
          q.interface.close()
  
          ts = []
          for page in range(map_threads):
              q = self.copy()
              q.limit(limit_count).offset(offset + (limit_count * page))
              t = ReduceThread(
                  target=target_map,
                  query=q,
                  queue=queue,
              )
              t.start()
              ts.append(t)
  
          while ts or not queue.empty():
              try:
                  val = queue.get(True, 1.0)
                  target_reduce(val)
  
              except queues.Empty:
                  pass
  
              else:
                  queue.task_done()
  
              # faster than using any((t.is_alive() for t in mts))
              ts = [t for t in ts if t.is_alive()]