def reduce(self, target_map, target_reduce, threads=<NUM_LIT:0>):
if not threads:<EOL><INDENT>threads = multiprocessing.cpu_count()<EOL><DEDENT>map_threads = threads - <NUM_LIT:1> if threads > <NUM_LIT:1> else <NUM_LIT:1><EOL>q = self.copy()<EOL>limit = q.bounds.limit<EOL>offset = q.bounds.offset<EOL>total_count = limit if limit else q.count()<EOL>limit_count = int(math.ceil(float(total_count) / float(map_threads)))<EOL>logger.info("<STR_LIT>".format(<EOL>map_threads,<EOL>limit_count,<EOL>total_count<EOL>))<EOL>queue = multiprocessing.JoinableQueue()<EOL>interfaces = get_interfaces()<EOL>for name, inter in interfaces.items():<EOL><INDENT>inter.close()<EOL><DEDENT>q.interface.close()<EOL>ts = []<EOL>for page in range(map_threads):<EOL><INDENT>q = self.copy()<EOL>q.limit(limit_count).offset(offset + (limit_count * page))<EOL>t = ReduceThread(<EOL>target=target_map,<EOL>query=q,<EOL>queue=queue,<EOL>)<EOL>t.start()<EOL>ts.append(t)<EOL><DEDENT>while ts or not queue.empty():<EOL><INDENT>try:<EOL><INDENT>val = queue.get(True, <NUM_LIT:1.0>)<EOL>target_reduce(val)<EOL><DEDENT>except queues.Empty:<EOL><INDENT>pass<EOL><DEDENT>else:<EOL><INDENT>queue.task_done()<EOL><DEDENT>ts = [t for t in ts if t.is_alive()]<EOL><DEDENT>
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