test_consistency.py 文件源码

def _query(self, session, keyspace, count, consistency_level=ConsistencyLevel.ONE):
        routing_key = struct.pack('>i', 0)
        for i in range(count):
            ss = SimpleStatement('SELECT * FROM cf WHERE k = 0',
                                 consistency_level=consistency_level,
                                 routing_key=routing_key)
            tries = 0
            while True:
                if tries > 100:
                    raise RuntimeError("Failed to execute query after 100 attempts: {0}".format(ss))
                try:
                    self.coordinator_stats.add_coordinator(session.execute_async(ss))
                    break
                except (OperationTimedOut, ReadTimeout):
                    ex_type, ex, tb = sys.exc_info()
                    log.warn("{0}: {1} Backtrace: {2}".format(ex_type.__name__, ex, traceback.extract_tb(tb)))
                    del tb
                    tries += 1
                    time.sleep(1)