def run(self, timeout=9000):

        index_key_sets = []
        index_query_threads = []
        queue = Queue.Queue()
        for (field, op, value) in self._filters:
            # spin off threads to do index queries
            thd = threading.Thread(target=self._filter_to_index_query,
                                   args=(field, op, value, queue))
            index_query_threads.append(thd)
            thd.start()

        # wait for the threads to finish the job
        [thd.join() for thd in index_query_threads]

        # get the results back from the queue
        while not queue.empty():
            key_set = queue.get()
            index_key_sets.append(key_set)

        # find the intersection of the key sets.
        if index_key_sets:
            mr_inputs = reduce(lambda x, y: x.intersection(y), index_key_sets)
        else:
            mr_inputs = set()

        if self._filters and not mr_inputs:
            yield []

        if not mr_inputs:
            self._mr_query = self._client.add(self._bucket)
        for key in mr_inputs:
            self._mr_query.add(self._bucket, key)

        self._mr_query.map(JS_MAP_FUNCTION)

        if self._order:
            self._mr_query.reduce(JS_REDUCE_ORDER_FUNC,
                                  {'arg': {'by': self._order[0],
                                           'order': self._order[1].lower()
                                           }
                                   })

        if self._limit:
            start = self._offset
            end = self._offset + self._limit
            if (end > len(mr_inputs)) and self._filters:
                end = 0
            self._mr_query.reduce('Riak.reduceSlice', {'arg': [start, end]})

        for result in self._mr_query.run(timeout):
            yield result
Shows the details of an worker.
dataset/ETHPy150Open mining/mining/mining/db/backends/mriak/multiquery.py/RiakMultiIndexQuery.run--crossover with--dataset/ETHPy150Open cloudControl/cctrl/cctrl/app.py/AppController.showWorker