# -----!/usr/bin/python3.7 ################################################################### # Copyright (c) 2016 by TAOS Technologies, Inc. # All rights reserved. # # This file is proprietary and confidential to TAOS Technologies. # No part of this file may be reproduced, stored, transmitted, # disclosed or used in any form or by any means other than as # expressly provided by the written permission from Jianhui Tao # ################################################################### # -*- coding: utf-8 -*- # For type hinting before definition, ref: # https://stackoverflow.com/questions/33533148/how-do-i-specify-that-the-return-type-of-a-method-is-the-same-as-the-class-itsel from __future__ import annotations import taos import crash_gen from util.sql import * from util.cases import * from util.dnodes import * from util.log import * from queue import Queue, Empty from typing import IO from typing import Set from typing import Dict from typing import List from requests.auth import HTTPBasicAuth import textwrap import datetime import logging import time import random import threading import requests import copy import argparse import getopt import sys import os import io import signal import traceback # Require Python 3 if sys.version_info[0] < 3: raise Exception("Must be using Python 3") # Global variables, tried to keep a small number. # Command-line/Environment Configurations, will set a bit later # ConfigNameSpace = argparse.Namespace gConfig = argparse.Namespace() # Dummy value, will be replaced later logger = None def runThread(wt: WorkerThread): wt.run() class CrashGenError(Exception): def __init__(self, msg=None, errno=None): self.msg = msg self.errno = errno def __str__(self): return self.msg class WorkerThread: def __init__(self, pool: ThreadPool, tid, tc: ThreadCoordinator, # te: TaskExecutor, ): # note: main thread context! # self._curStep = -1 self._pool = pool self._tid = tid self._tc = tc # type: ThreadCoordinator # self.threadIdent = threading.get_ident() self._thread = threading.Thread(target=runThread, args=(self,)) self._stepGate = threading.Event() # Let us have a DB connection of our own if (gConfig.per_thread_db_connection): # type: ignore # print("connector_type = {}".format(gConfig.connector_type)) if gConfig.connector_type == 'native': self._dbConn = DbConn.createNative() elif gConfig.connector_type == 'rest': self._dbConn = DbConn.createRest() elif gConfig.connector_type == 'mixed': if Dice.throw(2) == 0: # 1/2 chance self._dbConn = DbConn.createNative() else: self._dbConn = DbConn.createRest() else: raise RuntimeError("Unexpected connector type: {}".format(gConfig.connector_type)) self._dbInUse = False # if "use db" was executed already def logDebug(self, msg): logger.debug(" TRD[{}] {}".format(self._tid, msg)) def logInfo(self, msg): logger.info(" TRD[{}] {}".format(self._tid, msg)) def dbInUse(self): return self._dbInUse def useDb(self): if (not self._dbInUse): self.execSql("use db") self._dbInUse = True def getTaskExecutor(self): return self._tc.getTaskExecutor() def start(self): self._thread.start() # AFTER the thread is recorded def run(self): # initialization after thread starts, in the thread context # self.isSleeping = False logger.info("Starting to run thread: {}".format(self._tid)) if (gConfig.per_thread_db_connection): # type: ignore logger.debug("Worker thread openning database connection") self._dbConn.open() self._doTaskLoop() # clean up if (gConfig.per_thread_db_connection): # type: ignore self._dbConn.close() def _doTaskLoop(self): # while self._curStep < self._pool.maxSteps: # tc = ThreadCoordinator(None) while True: tc = self._tc # Thread Coordinator, the overall master tc.crossStepBarrier() # shared barrier first, INCLUDING the last one logger.debug("[TRD] Worker thread [{}] exited barrier...".format(self._tid)) self.crossStepGate() # then per-thread gate, after being tapped logger.debug("[TRD] Worker thread [{}] exited step gate...".format(self._tid)) if not self._tc.isRunning(): logger.debug("[TRD] Thread Coordinator not running any more, worker thread now stopping...") break # Fetch a task from the Thread Coordinator logger.debug( "[TRD] Worker thread [{}] about to fetch task".format(self._tid)) task = tc.fetchTask() # Execute such a task logger.debug( "[TRD] Worker thread [{}] about to execute task: {}".format( self._tid, task.__class__.__name__)) task.execute(self) tc.saveExecutedTask(task) logger.debug("[TRD] Worker thread [{}] finished executing task".format(self._tid)) self._dbInUse = False # there may be changes between steps def verifyThreadSelf(self): # ensure we are called by this own thread if (threading.get_ident() != self._thread.ident): raise RuntimeError("Unexpectly called from other threads") def verifyThreadMain(self): # ensure we are called by the main thread if (threading.get_ident() != threading.main_thread().ident): raise RuntimeError("Unexpectly called from other threads") def verifyThreadAlive(self): if (not self._thread.is_alive()): raise RuntimeError("Unexpected dead thread") # A gate is different from a barrier in that a thread needs to be "tapped" def crossStepGate(self): self.verifyThreadAlive() self.verifyThreadSelf() # only allowed by ourselves # Wait again at the "gate", waiting to be "tapped" logger.debug( "[TRD] Worker thread {} about to cross the step gate".format( self._tid)) self._stepGate.wait() self._stepGate.clear() # self._curStep += 1 # off to a new step... def tapStepGate(self): # give it a tap, release the thread waiting there self.verifyThreadAlive() self.verifyThreadMain() # only allowed for main thread logger.debug("[TRD] Tapping worker thread {}".format(self._tid)) self._stepGate.set() # wake up! time.sleep(0) # let the released thread run a bit def execSql(self, sql): # TODO: expose DbConn directly if (gConfig.per_thread_db_connection): return self._dbConn.execute(sql) else: return self._tc.getDbManager().getDbConn().execute(sql) def querySql(self, sql): # TODO: expose DbConn directly if (gConfig.per_thread_db_connection): return self._dbConn.query(sql) else: return self._tc.getDbManager().getDbConn().query(sql) def getQueryResult(self): if (gConfig.per_thread_db_connection): return self._dbConn.getQueryResult() else: return self._tc.getDbManager().getDbConn().getQueryResult() def getDbConn(self): if (gConfig.per_thread_db_connection): return self._dbConn else: return self._tc.getDbManager().getDbConn() # def querySql(self, sql): # not "execute", since we are out side the DB context # if ( gConfig.per_thread_db_connection ): # return self._dbConn.query(sql) # else: # return self._tc.getDbState().getDbConn().query(sql) # The coordinator of all worker threads, mostly running in main thread class ThreadCoordinator: def __init__(self, pool: ThreadPool, dbManager): self._curStep = -1 # first step is 0 self._pool = pool # self._wd = wd self._te = None # prepare for every new step self._dbManager = dbManager self._executedTasks: List[Task] = [] # in a given step self._lock = threading.RLock() # sync access for a few things self._stepBarrier = threading.Barrier( self._pool.numThreads + 1) # one barrier for all threads self._execStats = ExecutionStats() self._runStatus = MainExec.STATUS_RUNNING def getTaskExecutor(self): return self._te def getDbManager(self) -> DbManager: return self._dbManager def crossStepBarrier(self): self._stepBarrier.wait() def requestToStop(self): self._runStatus = MainExec.STATUS_STOPPING self._execStats.registerFailure("User Interruption") def _runShouldEnd(self, transitionFailed, hasAbortedTask): maxSteps = gConfig.max_steps # type: ignore if self._curStep >= (maxSteps - 1): # maxStep==10, last curStep should be 9 return True if self._runStatus != MainExec.STATUS_RUNNING: return True if transitionFailed: return True if hasAbortedTask: return True return False def _hasAbortedTask(self): # from execution of previous step for task in self._executedTasks: if task.isAborted(): # print("Task aborted: {}".format(task)) # hasAbortedTask = True return True return False def _releaseAllWorkerThreads(self, transitionFailed): self._curStep += 1 # we are about to get into next step. TODO: race condition here! # Now not all threads had time to go to sleep logger.debug( "--\r\n\n--> Step {} starts with main thread waking up".format(self._curStep)) # A new TE for the new step self._te = None # set to empty first, to signal worker thread to stop if not transitionFailed: # only if not failed self._te = TaskExecutor(self._curStep) logger.debug("[TRD] Main thread waking up at step {}, tapping worker threads".format( self._curStep)) # Now not all threads had time to go to sleep # Worker threads will wake up at this point, and each execute it's own task self.tapAllThreads() # release all worker thread from their "gate" def _syncAtBarrier(self): # Now main thread (that's us) is ready to enter a step # let other threads go past the pool barrier, but wait at the # thread gate logger.debug("[TRD] Main thread about to cross the barrier") self.crossStepBarrier() self._stepBarrier.reset() # Other worker threads should now be at the "gate" logger.debug("[TRD] Main thread finished crossing the barrier") def _doTransition(self): transitionFailed = False try: sm = self._dbManager.getStateMachine() logger.debug("[STT] starting transitions") # at end of step, transiton the DB state sm.transition(self._executedTasks) logger.debug("[STT] transition ended") # Due to limitation (or maybe not) of the Python library, # we cannot share connections across threads if sm.hasDatabase(): for t in self._pool.threadList: logger.debug("[DB] use db for all worker threads") t.useDb() # t.execSql("use db") # main thread executing "use # db" on behalf of every worker thread except taos.error.ProgrammingError as err: if (err.msg == 'network unavailable'): # broken DB connection logger.info("DB connection broken, execution failed") traceback.print_stack() transitionFailed = True self._te = None # Not running any more self._execStats.registerFailure("Broken DB Connection") # continue # don't do that, need to tap all threads at # end, and maybe signal them to stop else: raise self.resetExecutedTasks() # clear the tasks after we are done # Get ready for next step logger.debug("<-- Step {} finished, trasition failed = {}".format(self._curStep, transitionFailed)) return transitionFailed def run(self): self._pool.createAndStartThreads(self) # Coordinate all threads step by step self._curStep = -1 # not started yet self._execStats.startExec() # start the stop watch transitionFailed = False hasAbortedTask = False while not self._runShouldEnd(transitionFailed, hasAbortedTask): if not gConfig.debug: # print this only if we are not in debug mode print(".", end="", flush=True) self._syncAtBarrier() # For now just cross the barrier # At this point, all threads should be pass the overall "barrier" and before the per-thread "gate" # We use this period to do house keeping work, when all worker # threads are QUIET. hasAbortedTask = self._hasAbortedTask() # from previous step if hasAbortedTask: logger.info("Aborted task encountered, exiting test program") self._execStats.registerFailure("Aborted Task Encountered") break # do transition only if tasks are error free # Ending previous step transitionFailed = self._doTransition() # To start, we end step -1 first # Then we move on to the next step self._releaseAllWorkerThreads(transitionFailed) if hasAbortedTask or transitionFailed : # abnormal ending, workers waiting at "gate" logger.debug("Abnormal ending of main thraed") else: # regular ending, workers waiting at "barrier" logger.debug("Regular ending, main thread waiting for all worker threads to stop...") self._syncAtBarrier() self._te = None # No more executor, time to end logger.debug("Main thread tapping all threads one last time...") self.tapAllThreads() # Let the threads run one last time logger.debug("\r\n\n--> Main thread ready to finish up...") logger.debug("Main thread joining all threads") self._pool.joinAll() # Get all threads to finish logger.info("\nAll worker threads finished") self._execStats.endExec() def printStats(self): self._execStats.printStats() def isFailed(self): return self._execStats.isFailed() def getExecStats(self): return self._execStats def tapAllThreads(self): # in a deterministic manner wakeSeq = [] for i in range(self._pool.numThreads): # generate a random sequence if Dice.throw(2) == 1: wakeSeq.append(i) else: wakeSeq.insert(0, i) logger.debug( "[TRD] Main thread waking up worker threads: {}".format( str(wakeSeq))) # TODO: set dice seed to a deterministic value for i in wakeSeq: # TODO: maybe a bit too deep?! self._pool.threadList[i].tapStepGate() time.sleep(0) # yield def isRunning(self): return self._te is not None def fetchTask(self) -> Task: if (not self.isRunning()): # no task raise RuntimeError("Cannot fetch task when not running") # return self._wd.pickTask() # Alternatively, let's ask the DbState for the appropriate task # dbState = self.getDbState() # tasks = dbState.getTasksAtState() # TODO: create every time? # nTasks = len(tasks) # i = Dice.throw(nTasks) # logger.debug(" (dice:{}/{}) ".format(i, nTasks)) # # return copy.copy(tasks[i]) # Needs a fresh copy, to save execution results, etc. # return tasks[i].clone() # TODO: still necessary? # pick a task type for current state taskType = self.getDbManager().getStateMachine().pickTaskType() return taskType( self.getDbManager(), self._execStats) # create a task from it def resetExecutedTasks(self): self._executedTasks = [] # should be under single thread def saveExecutedTask(self, task): with self._lock: self._executedTasks.append(task) # We define a class to run a number of threads in locking steps. class ThreadPool: def __init__(self, numThreads, maxSteps): self.numThreads = numThreads self.maxSteps = maxSteps # Internal class variables self.curStep = 0 self.threadList = [] # type: List[WorkerThread] # starting to run all the threads, in locking steps def createAndStartThreads(self, tc: ThreadCoordinator): for tid in range(0, self.numThreads): # Create the threads workerThread = WorkerThread(self, tid, tc) self.threadList.append(workerThread) workerThread.start() # start, but should block immediately before step 0 def joinAll(self): for workerThread in self.threadList: logger.debug("Joining thread...") workerThread._thread.join() # A queue of continguous POSITIVE integers, used by DbManager to generate continuous numbers # for new table names class LinearQueue(): def __init__(self): self.firstIndex = 1 # 1st ever element self.lastIndex = 0 self._lock = threading.RLock() # our functions may call each other self.inUse = set() # the indexes that are in use right now def toText(self): return "[{}..{}], in use: {}".format( self.firstIndex, self.lastIndex, self.inUse) # Push (add new element, largest) to the tail, and mark it in use def push(self): with self._lock: # if ( self.isEmpty() ): # self.lastIndex = self.firstIndex # return self.firstIndex # Otherwise we have something self.lastIndex += 1 self.allocate(self.lastIndex) # self.inUse.add(self.lastIndex) # mark it in use immediately return self.lastIndex def pop(self): with self._lock: if (self.isEmpty()): # raise RuntimeError("Cannot pop an empty queue") return False # TODO: None? index = self.firstIndex if (index in self.inUse): return False self.firstIndex += 1 return index def isEmpty(self): return self.firstIndex > self.lastIndex def popIfNotEmpty(self): with self._lock: if (self.isEmpty()): return 0 return self.pop() def allocate(self, i): with self._lock: # logger.debug("LQ allocating item {}".format(i)) if (i in self.inUse): raise RuntimeError( "Cannot re-use same index in queue: {}".format(i)) self.inUse.add(i) def release(self, i): with self._lock: # logger.debug("LQ releasing item {}".format(i)) self.inUse.remove(i) # KeyError possible, TODO: why? def size(self): return self.lastIndex + 1 - self.firstIndex def pickAndAllocate(self): if (self.isEmpty()): return None with self._lock: cnt = 0 # counting the interations while True: cnt += 1 if (cnt > self.size() * 10): # 10x iteration already # raise RuntimeError("Failed to allocate LinearQueue element") return None ret = Dice.throwRange(self.firstIndex, self.lastIndex + 1) if (ret not in self.inUse): self.allocate(ret) return ret class DbConn: TYPE_NATIVE = "native-c" TYPE_REST = "rest-api" TYPE_INVALID = "invalid" @classmethod def create(cls, connType): if connType == cls.TYPE_NATIVE: return DbConnNative() elif connType == cls.TYPE_REST: return DbConnRest() else: raise RuntimeError( "Unexpected connection type: {}".format(connType)) @classmethod def createNative(cls): return cls.create(cls.TYPE_NATIVE) @classmethod def createRest(cls): return cls.create(cls.TYPE_REST) def __init__(self): self.isOpen = False self._type = self.TYPE_INVALID def open(self): if (self.isOpen): raise RuntimeError("Cannot re-open an existing DB connection") # below implemented by child classes self.openByType() logger.debug( "[DB] data connection opened, type = {}".format( self._type)) self.isOpen = True def resetDb(self): # reset the whole database, etc. if (not self.isOpen): raise RuntimeError( "Cannot reset database until connection is open") # self._tdSql.prepare() # Recreate database, etc. self.execute('drop database if exists db') logger.debug("Resetting DB, dropped database") # self._cursor.execute('create database db') # self._cursor.execute('use db') # tdSql.execute('show databases') def queryScalar(self, sql) -> int: return self._queryAny(sql) def queryString(self, sql) -> str: return self._queryAny(sql) def _queryAny(self, sql): # actual query result as an int if (not self.isOpen): raise RuntimeError( "Cannot query database until connection is open") nRows = self.query(sql) if nRows != 1: raise RuntimeError( "Unexpected result for query: {}, rows = {}".format( sql, nRows)) if self.getResultRows() != 1 or self.getResultCols() != 1: raise RuntimeError( "Unexpected result set for query: {}".format(sql)) return self.getQueryResult()[0][0] def execute(self, sql): raise RuntimeError("Unexpected execution, should be overriden") def openByType(self): raise RuntimeError("Unexpected execution, should be overriden") def getQueryResult(self): raise RuntimeError("Unexpected execution, should be overriden") def getResultRows(self): raise RuntimeError("Unexpected execution, should be overriden") def getResultCols(self): raise RuntimeError("Unexpected execution, should be overriden") # Sample: curl -u root:taosdata -d "show databases" localhost:6020/rest/sql class DbConnRest(DbConn): def __init__(self): super().__init__() self._type = self.TYPE_REST self._url = "http://localhost:6020/rest/sql" # fixed for now self._result = None def openByType(self): # Open connection pass # do nothing, always open def close(self): if (not self.isOpen): raise RuntimeError( "Cannot clean up database until connection is open") # Do nothing for REST logger.debug("[DB] REST Database connection closed") self.isOpen = False def _doSql(self, sql): try: r = requests.post(self._url, data = sql, auth = HTTPBasicAuth('root', 'taosdata')) except: print("REST API Failure (TODO: more info here)") raise rj = r.json() # Sanity check for the "Json Result" if ('status' not in rj): raise RuntimeError("No status in REST response") if rj['status'] == 'error': # clearly reported error if ('code' not in rj): # error without code raise RuntimeError("REST error return without code") errno = rj['code'] # May need to massage this in the future # print("Raising programming error with REST return: {}".format(rj)) raise taos.error.ProgrammingError( rj['desc'], errno) # todo: check existance of 'desc' if rj['status'] != 'succ': # better be this raise RuntimeError( "Unexpected REST return status: {}".format( rj['status'])) nRows = rj['rows'] if ('rows' in rj) else 0 self._result = rj return nRows def execute(self, sql): if (not self.isOpen): raise RuntimeError( "Cannot execute database commands until connection is open") logger.debug("[SQL-REST] Executing SQL: {}".format(sql)) nRows = self._doSql(sql) logger.debug( "[SQL-REST] Execution Result, nRows = {}, SQL = {}".format(nRows, sql)) return nRows def query(self, sql): # return rows affected return self.execute(sql) def getQueryResult(self): return self._result['data'] def getResultRows(self): print(self._result) raise RuntimeError("TBD") # return self._tdSql.queryRows def getResultCols(self): print(self._result) raise RuntimeError("TBD") # Duplicate code from TDMySQL, TODO: merge all this into DbConnNative class MyTDSql: def __init__(self): self.queryRows = 0 self.queryCols = 0 self.affectedRows = 0 def init(self, cursor, log=True): self.cursor = cursor # if (log): # caller = inspect.getframeinfo(inspect.stack()[1][0]) # self.cursor.log(caller.filename + ".sql") def close(self): self.cursor.close() def query(self, sql): self.sql = sql try: self.cursor.execute(sql) self.queryResult = self.cursor.fetchall() self.queryRows = len(self.queryResult) self.queryCols = len(self.cursor.description) except Exception as e: # caller = inspect.getframeinfo(inspect.stack()[1][0]) # args = (caller.filename, caller.lineno, sql, repr(e)) # tdLog.exit("%s(%d) failed: sql:%s, %s" % args) raise return self.queryRows def execute(self, sql): self.sql = sql try: self.affectedRows = self.cursor.execute(sql) except Exception as e: # caller = inspect.getframeinfo(inspect.stack()[1][0]) # args = (caller.filename, caller.lineno, sql, repr(e)) # tdLog.exit("%s(%d) failed: sql:%s, %s" % args) raise return self.affectedRows class DbConnNative(DbConn): def __init__(self): super().__init__() self._type = self.TYPE_NATIVE self._conn = None self._cursor = None def getBuildPath(self): selfPath = os.path.dirname(os.path.realpath(__file__)) if ("community" in selfPath): projPath = selfPath[:selfPath.find("communit")] else: projPath = selfPath[:selfPath.find("tests")] for root, dirs, files in os.walk(projPath): if ("taosd" in files): rootRealPath = os.path.dirname(os.path.realpath(root)) if ("packaging" not in rootRealPath): buildPath = root[:len(root) - len("/build/bin")] break return buildPath connInfoDisplayed = False def openByType(self): # Open connection cfgPath = self.getBuildPath() + "/test/cfg" hostAddr = "127.0.0.1" if not self.connInfoDisplayed: logger.info("Initiating TAOS native connection to {}, using config at {}".format(hostAddr, cfgPath)) self.connInfoDisplayed = True self._conn = taos.connect( host=hostAddr, config=cfgPath) # TODO: make configurable self._cursor = self._conn.cursor() # Get the connection/cursor ready self._cursor.execute('reset query cache') # self._cursor.execute('use db') # do this at the beginning of every # step # Open connection self._tdSql = MyTDSql() self._tdSql.init(self._cursor) def close(self): if (not self.isOpen): raise RuntimeError( "Cannot clean up database until connection is open") self._tdSql.close() logger.debug("[DB] Database connection closed") self.isOpen = False def execute(self, sql): if (not self.isOpen): raise RuntimeError( "Cannot execute database commands until connection is open") logger.debug("[SQL] Executing SQL: {}".format(sql)) nRows = self._tdSql.execute(sql) logger.debug( "[SQL] Execution Result, nRows = {}, SQL = {}".format( nRows, sql)) return nRows def query(self, sql): # return rows affected if (not self.isOpen): raise RuntimeError( "Cannot query database until connection is open") logger.debug("[SQL] Executing SQL: {}".format(sql)) nRows = self._tdSql.query(sql) logger.debug( "[SQL] Query Result, nRows = {}, SQL = {}".format( nRows, sql)) return nRows # results are in: return self._tdSql.queryResult def getQueryResult(self): return self._tdSql.queryResult def getResultRows(self): return self._tdSql.queryRows def getResultCols(self): return self._tdSql.queryCols class AnyState: STATE_INVALID = -1 STATE_EMPTY = 0 # nothing there, no even a DB STATE_DB_ONLY = 1 # we have a DB, but nothing else STATE_TABLE_ONLY = 2 # we have a table, but totally empty STATE_HAS_DATA = 3 # we have some data in the table _stateNames = ["Invalid", "Empty", "DB_Only", "Table_Only", "Has_Data"] STATE_VAL_IDX = 0 CAN_CREATE_DB = 1 CAN_DROP_DB = 2 CAN_CREATE_FIXED_SUPER_TABLE = 3 CAN_DROP_FIXED_SUPER_TABLE = 4 CAN_ADD_DATA = 5 CAN_READ_DATA = 6 def __init__(self): self._info = self.getInfo() def __str__(self): # -1 hack to accomodate the STATE_INVALID case return self._stateNames[self._info[self.STATE_VAL_IDX] + 1] def getInfo(self): raise RuntimeError("Must be overriden by child classes") def equals(self, other): if isinstance(other, int): return self.getValIndex() == other elif isinstance(other, AnyState): return self.getValIndex() == other.getValIndex() else: raise RuntimeError( "Unexpected comparison, type = {}".format( type(other))) def verifyTasksToState(self, tasks, newState): raise RuntimeError("Must be overriden by child classes") def getValIndex(self): return self._info[self.STATE_VAL_IDX] def getValue(self): return self._info[self.STATE_VAL_IDX] def canCreateDb(self): return self._info[self.CAN_CREATE_DB] def canDropDb(self): return self._info[self.CAN_DROP_DB] def canCreateFixedSuperTable(self): return self._info[self.CAN_CREATE_FIXED_SUPER_TABLE] def canDropFixedSuperTable(self): return self._info[self.CAN_DROP_FIXED_SUPER_TABLE] def canAddData(self): return self._info[self.CAN_ADD_DATA] def canReadData(self): return self._info[self.CAN_READ_DATA] def assertAtMostOneSuccess(self, tasks, cls): sCnt = 0 for task in tasks: if not isinstance(task, cls): continue if task.isSuccess(): # task.logDebug("Task success found") sCnt += 1 if (sCnt >= 2): raise RuntimeError( "Unexpected more than 1 success with task: {}".format(cls)) def assertIfExistThenSuccess(self, tasks, cls): sCnt = 0 exists = False for task in tasks: if not isinstance(task, cls): continue exists = True # we have a valid instance if task.isSuccess(): sCnt += 1 if (exists and sCnt <= 0): raise RuntimeError( "Unexpected zero success for task: {}".format(cls)) def assertNoTask(self, tasks, cls): for task in tasks: if isinstance(task, cls): raise CrashGenError( "This task: {}, is not expected to be present, given the success/failure of others".format(cls.__name__)) def assertNoSuccess(self, tasks, cls): for task in tasks: if isinstance(task, cls): if task.isSuccess(): raise RuntimeError( "Unexpected successful task: {}".format(cls)) def hasSuccess(self, tasks, cls): for task in tasks: if not isinstance(task, cls): continue if task.isSuccess(): return True return False def hasTask(self, tasks, cls): for task in tasks: if isinstance(task, cls): return True return False class StateInvalid(AnyState): def getInfo(self): return [ self.STATE_INVALID, False, False, # can create/drop Db False, False, # can create/drop fixed table False, False, # can insert/read data with fixed table ] # def verifyTasksToState(self, tasks, newState): class StateEmpty(AnyState): def getInfo(self): return [ self.STATE_EMPTY, True, False, # can create/drop Db False, False, # can create/drop fixed table False, False, # can insert/read data with fixed table ] def verifyTasksToState(self, tasks, newState): if (self.hasSuccess(tasks, TaskCreateDb) ): # at EMPTY, if there's succes in creating DB if (not self.hasTask(tasks, TaskDropDb)): # and no drop_db tasks # we must have at most one. TODO: compare numbers self.assertAtMostOneSuccess(tasks, TaskCreateDb) class StateDbOnly(AnyState): def getInfo(self): return [ self.STATE_DB_ONLY, False, True, True, False, False, False, ] def verifyTasksToState(self, tasks, newState): if (not self.hasTask(tasks, TaskCreateDb)): # only if we don't create any more self.assertAtMostOneSuccess(tasks, TaskDropDb) self.assertIfExistThenSuccess(tasks, TaskDropDb) # self.assertAtMostOneSuccess(tasks, CreateFixedTableTask) # not true in massively parrallel cases # Nothing to be said about adding data task # if ( self.hasSuccess(tasks, DropDbTask) ): # dropped the DB # self.assertHasTask(tasks, DropDbTask) # implied by hasSuccess # self.assertAtMostOneSuccess(tasks, DropDbTask) # self._state = self.STATE_EMPTY # if ( self.hasSuccess(tasks, TaskCreateSuperTable) ): # did not drop db, create table success # # self.assertHasTask(tasks, CreateFixedTableTask) # tried to create table # if ( not self.hasTask(tasks, TaskDropSuperTable) ): # self.assertAtMostOneSuccess(tasks, TaskCreateSuperTable) # at most 1 attempt is successful, if we don't drop anything # self.assertNoTask(tasks, DropDbTask) # should have have tried # if ( not self.hasSuccess(tasks, AddFixedDataTask) ): # just created table, no data yet # # can't say there's add-data attempts, since they may all fail # self._state = self.STATE_TABLE_ONLY # else: # self._state = self.STATE_HAS_DATA # What about AddFixedData? # elif ( self.hasSuccess(tasks, AddFixedDataTask) ): # self._state = self.STATE_HAS_DATA # else: # no success in dropping db tasks, no success in create fixed table? read data should also fail # # raise RuntimeError("Unexpected no-success scenario") # We might just landed all failure tasks, # self._state = self.STATE_DB_ONLY # no change class StateSuperTableOnly(AnyState): def getInfo(self): return [ self.STATE_TABLE_ONLY, False, True, False, True, True, True, ] def verifyTasksToState(self, tasks, newState): if (self.hasSuccess(tasks, TaskDropSuperTable) ): # we are able to drop the table #self.assertAtMostOneSuccess(tasks, TaskDropSuperTable) # we must have had recreted it self.hasSuccess(tasks, TaskCreateSuperTable) # self._state = self.STATE_DB_ONLY # elif ( self.hasSuccess(tasks, AddFixedDataTask) ): # no success dropping the table, but added data # self.assertNoTask(tasks, DropFixedTableTask) # not true in massively parrallel cases # self._state = self.STATE_HAS_DATA # elif ( self.hasSuccess(tasks, ReadFixedDataTask) ): # no success in prev cases, but was able to read data # self.assertNoTask(tasks, DropFixedTableTask) # self.assertNoTask(tasks, AddFixedDataTask) # self._state = self.STATE_TABLE_ONLY # no change # else: # did not drop table, did not insert data, did not read successfully, that is impossible # raise RuntimeError("Unexpected no-success scenarios") # TODO: need to revamp!! class StateHasData(AnyState): def getInfo(self): return [ self.STATE_HAS_DATA, False, True, False, True, True, True, ] def verifyTasksToState(self, tasks, newState): if (newState.equals(AnyState.STATE_EMPTY)): self.hasSuccess(tasks, TaskDropDb) if (not self.hasTask(tasks, TaskCreateDb)): self.assertAtMostOneSuccess(tasks, TaskDropDb) # TODO: dicy elif (newState.equals(AnyState.STATE_DB_ONLY)): # in DB only if (not self.hasTask(tasks, TaskCreateDb) ): # without a create_db task # we must have drop_db task self.assertNoTask(tasks, TaskDropDb) self.hasSuccess(tasks, TaskDropSuperTable) # self.assertAtMostOneSuccess(tasks, DropFixedSuperTableTask) # TODO: dicy # elif ( newState.equals(AnyState.STATE_TABLE_ONLY) ): # data deleted # self.assertNoTask(tasks, TaskDropDb) # self.assertNoTask(tasks, TaskDropSuperTable) # self.assertNoTask(tasks, TaskAddData) # self.hasSuccess(tasks, DeleteDataTasks) else: # should be STATE_HAS_DATA if (not self.hasTask(tasks, TaskCreateDb) ): # only if we didn't create one # we shouldn't have dropped it self.assertNoTask(tasks, TaskDropDb) if (not self.hasTask(tasks, TaskCreateSuperTable) ): # if we didn't create the table # we should not have a task that drops it self.assertNoTask(tasks, TaskDropSuperTable) # self.assertIfExistThenSuccess(tasks, ReadFixedDataTask) class StateMechine: def __init__(self, dbConn): self._dbConn = dbConn self._curState = self._findCurrentState() # starting state # transitition target probabilities, indexed with value of STATE_EMPTY, # STATE_DB_ONLY, etc. self._stateWeights = [1, 3, 5, 15] def getCurrentState(self): return self._curState def hasDatabase(self): return self._curState.canDropDb() # ha, can drop DB means it has one # May be slow, use cautionsly... def getTaskTypes(self): # those that can run (directly/indirectly) from the current state def typesToStrings(types): ss = [] for t in types: ss.append(t.__name__) return ss allTaskClasses = StateTransitionTask.__subclasses__() # all state transition tasks firstTaskTypes = [] for tc in allTaskClasses: # t = tc(self) # create task object if tc.canBeginFrom(self._curState): firstTaskTypes.append(tc) # now we have all the tasks that can begin directly from the current # state, let's figure out the INDIRECT ones taskTypes = firstTaskTypes.copy() # have to have these for task1 in firstTaskTypes: # each task type gathered so far endState = task1.getEndState() # figure the end state if endState is None: # does not change end state continue # no use, do nothing for tc in allTaskClasses: # what task can further begin from there? if tc.canBeginFrom(endState) and (tc not in firstTaskTypes): taskTypes.append(tc) # gather it if len(taskTypes) <= 0: raise RuntimeError( "No suitable task types found for state: {}".format( self._curState)) logger.debug( "[OPS] Tasks found for state {}: {}".format( self._curState, typesToStrings(taskTypes))) return taskTypes def _findCurrentState(self): dbc = self._dbConn ts = time.time() # we use this to debug how fast/slow it is to do the various queries to find the current DB state if dbc.query("show databases") == 0: # no database?! # logger.debug("Found EMPTY state") logger.debug( "[STT] empty database found, between {} and {}".format( ts, time.time())) return StateEmpty() # did not do this when openning connection, and this is NOT the worker # thread, which does this on their own dbc.execute("use db") if dbc.query("show tables") == 0: # no tables # logger.debug("Found DB ONLY state") logger.debug( "[STT] DB_ONLY found, between {} and {}".format( ts, time.time())) return StateDbOnly() if dbc.query("SELECT * FROM db.{}".format(DbManager.getFixedSuperTableName()) ) == 0: # no regular tables # logger.debug("Found TABLE_ONLY state") logger.debug( "[STT] SUPER_TABLE_ONLY found, between {} and {}".format( ts, time.time())) return StateSuperTableOnly() else: # has actual tables # logger.debug("Found HAS_DATA state") logger.debug( "[STT] HAS_DATA found, between {} and {}".format( ts, time.time())) return StateHasData() def transition(self, tasks): if (len(tasks) == 0): # before 1st step, or otherwise empty logger.debug("[STT] Starting State: {}".format(self._curState)) return # do nothing # this should show up in the server log, separating steps self._dbConn.execute("show dnodes") # Generic Checks, first based on the start state if self._curState.canCreateDb(): self._curState.assertIfExistThenSuccess(tasks, TaskCreateDb) # self.assertAtMostOneSuccess(tasks, CreateDbTask) # not really, in # case of multiple creation and drops if self._curState.canDropDb(): self._curState.assertIfExistThenSuccess(tasks, TaskDropDb) # self.assertAtMostOneSuccess(tasks, DropDbTask) # not really in # case of drop-create-drop # if self._state.canCreateFixedTable(): # self.assertIfExistThenSuccess(tasks, CreateFixedTableTask) # Not true, DB may be dropped # self.assertAtMostOneSuccess(tasks, CreateFixedTableTask) # not # really, in case of create-drop-create # if self._state.canDropFixedTable(): # self.assertIfExistThenSuccess(tasks, DropFixedTableTask) # Not True, the whole DB may be dropped # self.assertAtMostOneSuccess(tasks, DropFixedTableTask) # not # really in case of drop-create-drop # if self._state.canAddData(): # self.assertIfExistThenSuccess(tasks, AddFixedDataTask) # not true # actually # if self._state.canReadData(): # Nothing for sure newState = self._findCurrentState() logger.debug("[STT] New DB state determined: {}".format(newState)) # can old state move to new state through the tasks? self._curState.verifyTasksToState(tasks, newState) self._curState = newState def pickTaskType(self): # all the task types we can choose from at curent state taskTypes = self.getTaskTypes() weights = [] for tt in taskTypes: endState = tt.getEndState() if endState is not None: # TODO: change to a method weights.append(self._stateWeights[endState.getValIndex()]) else: # read data task, default to 10: TODO: change to a constant weights.append(10) i = self._weighted_choice_sub(weights) # logger.debug(" (weighted random:{}/{}) ".format(i, len(taskTypes))) return taskTypes[i] # ref: # https://eli.thegreenplace.net/2010/01/22/weighted-random-generation-in-python/ def _weighted_choice_sub(self, weights): # TODO: use our dice to ensure it being determinstic? rnd = random.random() * sum(weights) for i, w in enumerate(weights): rnd -= w if rnd < 0: return i # Manager of the Database Data/Connection class DbManager(): def __init__(self, resetDb=True): self.tableNumQueue = LinearQueue() # datetime.datetime(2019, 1, 1) # initial date time tick self._lastTick = self.setupLastTick() self._lastInt = 0 # next one is initial integer self._lock = threading.RLock() # self.openDbServerConnection() self._dbConn = DbConn.createNative() if ( gConfig.connector_type == 'native') else DbConn.createRest() try: self._dbConn.open() # may throw taos.error.ProgrammingError: disconnected except taos.error.ProgrammingError as err: # print("Error type: {}, msg: {}, value: {}".format(type(err), err.msg, err)) if (err.msg == 'client disconnected'): # cannot open DB connection print( "Cannot establish DB connection, please re-run script without parameter, and follow the instructions.") sys.exit(2) else: raise except BaseException: print("[=] Unexpected exception") raise if resetDb: self._dbConn.resetDb() # drop and recreate DB # Do this after dbConn is in proper shape self._stateMachine = StateMechine(self._dbConn) def getDbConn(self): return self._dbConn def getStateMachine(self) -> StateMechine: return self._stateMachine # def getState(self): # return self._stateMachine.getCurrentState() # We aim to create a starting time tick, such that, whenever we run our test here once # We should be able to safely create 100,000 records, which will not have any repeated time stamp # when we re-run the test in 3 minutes (180 seconds), basically we should expand time duration # by a factor of 500. # TODO: what if it goes beyond 10 years into the future # TODO: fix the error as result of above: "tsdb timestamp is out of range" def setupLastTick(self): t1 = datetime.datetime(2020, 6, 1) t2 = datetime.datetime.now() # maybe a very large number, takes 69 years to exceed Python int range elSec = int(t2.timestamp() - t1.timestamp()) elSec2 = (elSec % (8 * 12 * 30 * 24 * 60 * 60 / 500)) * \ 500 # a number representing seconds within 10 years # print("elSec = {}".format(elSec)) t3 = datetime.datetime(2012, 1, 1) # default "keep" is 10 years t4 = datetime.datetime.fromtimestamp( t3.timestamp() + elSec2) # see explanation above logger.info("Setting up TICKS to start from: {}".format(t4)) return t4 def pickAndAllocateTable(self): # pick any table, and "use" it return self.tableNumQueue.pickAndAllocate() def addTable(self): with self._lock: tIndex = self.tableNumQueue.push() return tIndex @classmethod def getFixedSuperTableName(cls): return "fs_table" def releaseTable(self, i): # return the table back, so others can use it self.tableNumQueue.release(i) def getNextTick(self): with self._lock: # prevent duplicate tick if Dice.throw(10) == 0: # 1 in 10 chance return self._lastTick + datetime.timedelta(0, -100) else: # regular # add one second to it self._lastTick += datetime.timedelta(0, 1) return self._lastTick def getNextInt(self): with self._lock: self._lastInt += 1 return self._lastInt def getNextBinary(self): return "Beijing_Shanghai_Los_Angeles_New_York_San_Francisco_Chicago_Beijing_Shanghai_Los_Angeles_New_York_San_Francisco_Chicago_{}".format( self.getNextInt()) def getNextFloat(self): return 0.9 + self.getNextInt() def getTableNameToDelete(self): tblNum = self.tableNumQueue.pop() # TODO: race condition! if (not tblNum): # maybe false return False return "table_{}".format(tblNum) def cleanUp(self): self._dbConn.close() class TaskExecutor(): class BoundedList: def __init__(self, size=10): self._size = size self._list = [] def add(self, n: int): if not self._list: # empty self._list.append(n) return # now we should insert nItems = len(self._list) insPos = 0 for i in range(nItems): insPos = i if n <= self._list[i]: # smaller than this item, time to insert break # found the insertion point insPos += 1 # insert to the right if insPos == 0: # except for the 1st item, # TODO: elimiate first item as gating item return # do nothing # print("Inserting at postion {}, value: {}".format(insPos, n)) self._list.insert(insPos, n) # insert newLen = len(self._list) if newLen <= self._size: return # do nothing elif newLen == (self._size + 1): del self._list[0] # remove the first item else: raise RuntimeError("Corrupt Bounded List") def __str__(self): return repr(self._list) _boundedList = BoundedList() def __init__(self, curStep): self._curStep = curStep @classmethod def getBoundedList(cls): return cls._boundedList def getCurStep(self): return self._curStep def execute(self, task: Task, wt: WorkerThread): # execute a task on a thread task.execute(wt) def recordDataMark(self, n: int): # print("[{}]".format(n), end="", flush=True) self._boundedList.add(n) # def logInfo(self, msg): # logger.info(" T[{}.x]: ".format(self._curStep) + msg) # def logDebug(self, msg): # logger.debug(" T[{}.x]: ".format(self._curStep) + msg) class Task(): taskSn = 100 @classmethod def allocTaskNum(cls): Task.taskSn += 1 # IMPORTANT: cannot use cls.taskSn, since each sub class will have a copy # logger.debug("Allocating taskSN: {}".format(Task.taskSn)) return Task.taskSn def __init__(self, dbManager: DbManager, execStats: ExecutionStats): self._dbManager = dbManager self._workerThread = None self._err = None self._aborted = False self._curStep = None self._numRows = None # Number of rows affected # Assign an incremental task serial number self._taskNum = self.allocTaskNum() # logger.debug("Creating new task {}...".format(self._taskNum)) self._execStats = execStats self._lastSql = "" # last SQL executed/attempted def isSuccess(self): return self._err is None def isAborted(self): return self._aborted def clone(self): # TODO: why do we need this again? newTask = self.__class__(self._dbManager, self._execStats) return newTask def logDebug(self, msg): self._workerThread.logDebug( "Step[{}.{}] {}".format( self._curStep, self._taskNum, msg)) def logInfo(self, msg): self._workerThread.logInfo( "Step[{}.{}] {}".format( self._curStep, self._taskNum, msg)) def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): raise RuntimeError( "To be implemeted by child classes, class name: {}".format( self.__class__.__name__)) def execute(self, wt: WorkerThread): wt.verifyThreadSelf() self._workerThread = wt # type: ignore te = wt.getTaskExecutor() self._curStep = te.getCurStep() self.logDebug( "[-] executing task {}...".format(self.__class__.__name__)) self._err = None self._execStats.beginTaskType( self.__class__.__name__) # mark beginning try: self._executeInternal(te, wt) # TODO: no return value? except taos.error.ProgrammingError as err: errno2 = err.errno if ( err.errno > 0) else 0x80000000 + err.errno # correct error scheme if (gConfig.continue_on_exception): # user choose to continue self.logDebug( "[=] Continue after TAOS exception: errno=0x{:X}, msg: {}, SQL: {}".format( errno2, err, self._lastSql)) self._err = err elif (errno2 in [ 0x05, # TSDB_CODE_RPC_NOT_READY 0x200, 0x360, 0x362, 0x36A, 0x36B, 0x36D, 0x381, 0x380, 0x383, 0x386, # DB is being dropped?! 0x503, 0x510, # vnode not in ready state 0x600, 1000 # REST catch-all error ]): # allowed errors self.logDebug( "[=] Acceptable Taos library exception: errno=0x{:X}, msg: {}, SQL: {}".format( errno2, err, self._lastSql)) print("_", end="", flush=True) self._err = err else: errMsg = "[=] Unexpected Taos library exception: errno=0x{:X}, msg: {}, SQL: {}".format( errno2, err, self._lastSql) self.logDebug(errMsg) if gConfig.debug: # raise # so that we see full stack traceback.print_exc() print( "\n\n----------------------------\nProgram ABORTED Due to Unexpected TAOS Error: \n\n{}\n".format(errMsg) + "----------------------------\n") # sys.exit(-1) self._err = err self._aborted = True except Exception as e: self.logInfo("Non-TAOS exception encountered") self._err = e self._aborted = True traceback.print_exc() except BaseException as e: self.logInfo("Python base exception encountered") self._err = e self._aborted = True traceback.print_exc() except BaseException: self.logDebug( "[=] Unexpected exception, SQL: {}".format( self._lastSql)) raise self._execStats.endTaskType(self.__class__.__name__, self.isSuccess()) self.logDebug("[X] task execution completed, {}, status: {}".format( self.__class__.__name__, "Success" if self.isSuccess() else "Failure")) # TODO: merge with above. self._execStats.incExecCount(self.__class__.__name__, self.isSuccess()) def execSql(self, sql): self._lastSql = sql return self._dbManager.execute(sql) def execWtSql(self, wt: WorkerThread, sql): # execute an SQL on the worker thread self._lastSql = sql return wt.execSql(sql) def queryWtSql(self, wt: WorkerThread, sql): # execute an SQL on the worker thread self._lastSql = sql return wt.querySql(sql) def getQueryResult(self, wt: WorkerThread): # execute an SQL on the worker thread return wt.getQueryResult() class ExecutionStats: def __init__(self): # total/success times for a task self._execTimes: Dict[str, [int, int]] = {} self._tasksInProgress = 0 self._lock = threading.Lock() self._firstTaskStartTime = None self._execStartTime = None self._elapsedTime = 0.0 # total elapsed time self._accRunTime = 0.0 # accumulated run time self._failed = False self._failureReason = None def __str__(self): return "[ExecStats: _failed={}, _failureReason={}".format( self._failed, self._failureReason) def isFailed(self): return self._failed def startExec(self): self._execStartTime = time.time() def endExec(self): self._elapsedTime = time.time() - self._execStartTime def incExecCount(self, klassName, isSuccess): # TODO: add a lock here if klassName not in self._execTimes: self._execTimes[klassName] = [0, 0] t = self._execTimes[klassName] # tuple for the data t[0] += 1 # index 0 has the "total" execution times if isSuccess: t[1] += 1 # index 1 has the "success" execution times def beginTaskType(self, klassName): with self._lock: if self._tasksInProgress == 0: # starting a new round self._firstTaskStartTime = time.time() # I am now the first task self._tasksInProgress += 1 def endTaskType(self, klassName, isSuccess): with self._lock: self._tasksInProgress -= 1 if self._tasksInProgress == 0: # all tasks have stopped self._accRunTime += (time.time() - self._firstTaskStartTime) self._firstTaskStartTime = None def registerFailure(self, reason): self._failed = True self._failureReason = reason def printStats(self): logger.info( "----------------------------------------------------------------------") logger.info( "| Crash_Gen test {}, with the following stats:". format( "FAILED (reason: {})".format( self._failureReason) if self._failed else "SUCCEEDED")) logger.info("| Task Execution Times (success/total):") execTimesAny = 0 for k, n in self._execTimes.items(): execTimesAny += n[0] logger.info("| {0:<24}: {1}/{2}".format(k, n[1], n[0])) logger.info( "| Total Tasks Executed (success or not): {} ".format(execTimesAny)) logger.info( "| Total Tasks In Progress at End: {}".format( self._tasksInProgress)) logger.info( "| Total Task Busy Time (elapsed time when any task is in progress): {:.3f} seconds".format( self._accRunTime)) logger.info( "| Average Per-Task Execution Time: {:.3f} seconds".format(self._accRunTime / execTimesAny)) logger.info( "| Total Elapsed Time (from wall clock): {:.3f} seconds".format( self._elapsedTime)) logger.info( "| Top numbers written: {}".format( TaskExecutor.getBoundedList())) logger.info( "----------------------------------------------------------------------") class StateTransitionTask(Task): LARGE_NUMBER_OF_TABLES = 35 SMALL_NUMBER_OF_TABLES = 3 LARGE_NUMBER_OF_RECORDS = 50 SMALL_NUMBER_OF_RECORDS = 3 @classmethod def getInfo(cls): # each sub class should supply their own information raise RuntimeError("Overriding method expected") _endState = None @classmethod def getEndState(cls): # TODO: optimize by calling it fewer times raise RuntimeError("Overriding method expected") # @classmethod # def getBeginStates(cls): # return cls.getInfo()[0] # @classmethod # def getEndState(cls): # returning the class name # return cls.getInfo()[0] @classmethod def canBeginFrom(cls, state: AnyState): # return state.getValue() in cls.getBeginStates() raise RuntimeError("must be overriden") @classmethod def getRegTableName(cls, i): return "db.reg_table_{}".format(i) def execute(self, wt: WorkerThread): super().execute(wt) class TaskCreateDb(StateTransitionTask): @classmethod def getEndState(cls): return StateDbOnly() @classmethod def canBeginFrom(cls, state: AnyState): return state.canCreateDb() def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): self.execWtSql(wt, "create database db") class TaskDropDb(StateTransitionTask): @classmethod def getEndState(cls): return StateEmpty() @classmethod def canBeginFrom(cls, state: AnyState): return state.canDropDb() def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): self.execWtSql(wt, "drop database db") logger.debug("[OPS] database dropped at {}".format(time.time())) class TaskCreateSuperTable(StateTransitionTask): @classmethod def getEndState(cls): return StateSuperTableOnly() @classmethod def canBeginFrom(cls, state: AnyState): return state.canCreateFixedSuperTable() def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): if not wt.dbInUse(): # no DB yet, to the best of our knowledge logger.debug("Skipping task, no DB yet") return tblName = self._dbManager.getFixedSuperTableName() # wt.execSql("use db") # should always be in place self.execWtSql( wt, "create table db.{} (ts timestamp, speed int) tags (b binary(200), f float) ".format(tblName)) # No need to create the regular tables, INSERT will do that # automatically class TaskReadData(StateTransitionTask): @classmethod def getEndState(cls): return None # meaning doesn't affect state @classmethod def canBeginFrom(cls, state: AnyState): return state.canReadData() def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): sTbName = self._dbManager.getFixedSuperTableName() self.queryWtSql(wt, "select TBNAME from db.{}".format( sTbName)) # TODO: analyze result set later if random.randrange( 5) == 0: # 1 in 5 chance, simulate a broken connection. TODO: break connection in all situations wt.getDbConn().close() wt.getDbConn().open() else: # wt.getDbConn().getQueryResult() rTables = self.getQueryResult(wt) # print("rTables[0] = {}, type = {}".format(rTables[0], type(rTables[0]))) for rTbName in rTables: # regular tables self.execWtSql(wt, "select * from db.{}".format(rTbName[0])) # tdSql.query(" cars where tbname in ('carzero', 'carone')") class TaskDropSuperTable(StateTransitionTask): @classmethod def getEndState(cls): return StateDbOnly() @classmethod def canBeginFrom(cls, state: AnyState): return state.canDropFixedSuperTable() def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): # 1/2 chance, we'll drop the regular tables one by one, in a randomized # sequence if Dice.throw(2) == 0: tblSeq = list(range( 2 + (self.LARGE_NUMBER_OF_TABLES if gConfig.larger_data else self.SMALL_NUMBER_OF_TABLES))) random.shuffle(tblSeq) tickOutput = False # if we have spitted out a "d" character for "drop regular table" isSuccess = True for i in tblSeq: regTableName = self.getRegTableName( i) # "db.reg_table_{}".format(i) try: self.execWtSql(wt, "drop table {}".format( regTableName)) # nRows always 0, like MySQL except taos.error.ProgrammingError as err: # correcting for strange error number scheme errno2 = err.errno if ( err.errno > 0) else 0x80000000 + err.errno if (errno2 in [0x362]): # mnode invalid table name isSuccess = False logger.debug( "[DB] Acceptable error when dropping a table") continue # try to delete next regular table if (not tickOutput): tickOutput = True # Print only one time if isSuccess: print("d", end="", flush=True) else: print("f", end="", flush=True) # Drop the super table itself tblName = self._dbManager.getFixedSuperTableName() self.execWtSql(wt, "drop table db.{}".format(tblName)) class TaskAlterTags(StateTransitionTask): @classmethod def getEndState(cls): return None # meaning doesn't affect state @classmethod def canBeginFrom(cls, state: AnyState): return state.canDropFixedSuperTable() # if we can drop it, we can alter tags def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): tblName = self._dbManager.getFixedSuperTableName() dice = Dice.throw(4) if dice == 0: sql = "alter table db.{} add tag extraTag int".format(tblName) elif dice == 1: sql = "alter table db.{} drop tag extraTag".format(tblName) elif dice == 2: sql = "alter table db.{} drop tag newTag".format(tblName) else: # dice == 3 sql = "alter table db.{} change tag extraTag newTag".format( tblName) self.execWtSql(wt, sql) class TaskAddData(StateTransitionTask): # Track which table is being actively worked on activeTable: Set[int] = set() # We use these two files to record operations to DB, useful for power-off # tests fAddLogReady = None fAddLogDone = None @classmethod def prepToRecordOps(cls): if gConfig.record_ops: if (cls.fAddLogReady is None): logger.info( "Recording in a file operations to be performed...") cls.fAddLogReady = open("add_log_ready.txt", "w") if (cls.fAddLogDone is None): logger.info("Recording in a file operations completed...") cls.fAddLogDone = open("add_log_done.txt", "w") @classmethod def getEndState(cls): return StateHasData() @classmethod def canBeginFrom(cls, state: AnyState): return state.canAddData() def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): ds = self._dbManager # wt.execSql("use db") # TODO: seems to be an INSERT bug to require # this tblSeq = list( range( self.LARGE_NUMBER_OF_TABLES if gConfig.larger_data else self.SMALL_NUMBER_OF_TABLES)) random.shuffle(tblSeq) for i in tblSeq: if (i in self.activeTable): # wow already active # logger.info("Concurrent data insertion into table: {}".format(i)) # print("ct({})".format(i), end="", flush=True) # Concurrent # insertion into table print("x", end="", flush=True) else: self.activeTable.add(i) # marking it active # No need to shuffle data sequence, unless later we decide to do # non-increment insertion regTableName = self.getRegTableName( i) # "db.reg_table_{}".format(i) for j in range( self.LARGE_NUMBER_OF_RECORDS if gConfig.larger_data else self.SMALL_NUMBER_OF_RECORDS): # number of records per table nextInt = ds.getNextInt() if gConfig.record_ops: self.prepToRecordOps() self.fAddLogReady.write( "Ready to write {} to {}\n".format( nextInt, regTableName)) self.fAddLogReady.flush() os.fsync(self.fAddLogReady) sql = "insert into {} using {} tags ('{}', {}) values ('{}', {});".format( regTableName, ds.getFixedSuperTableName(), ds.getNextBinary(), ds.getNextFloat(), ds.getNextTick(), nextInt) self.execWtSql(wt, sql) # Successfully wrote the data into the DB, let's record it # somehow te.recordDataMark(nextInt) if gConfig.record_ops: self.fAddLogDone.write( "Wrote {} to {}\n".format( nextInt, regTableName)) self.fAddLogDone.flush() os.fsync(self.fAddLogDone) self.activeTable.discard(i) # not raising an error, unlike remove # Deterministic random number generator class Dice(): seeded = False # static, uninitialized @classmethod def seed(cls, s): # static if (cls.seeded): raise RuntimeError( "Cannot seed the random generator more than once") cls.verifyRNG() random.seed(s) cls.seeded = True # TODO: protect against multi-threading @classmethod def verifyRNG(cls): # Verify that the RNG is determinstic random.seed(0) x1 = random.randrange(0, 1000) x2 = random.randrange(0, 1000) x3 = random.randrange(0, 1000) if (x1 != 864 or x2 != 394 or x3 != 776): raise RuntimeError("System RNG is not deterministic") @classmethod def throw(cls, stop): # get 0 to stop-1 return cls.throwRange(0, stop) @classmethod def throwRange(cls, start, stop): # up to stop-1 if (not cls.seeded): raise RuntimeError("Cannot throw dice before seeding it") return random.randrange(start, stop) class LoggingFilter(logging.Filter): def filter(self, record: logging.LogRecord): if (record.levelno >= logging.INFO): return True # info or above always log # Commenting out below to adjust... # if msg.startswith("[TRD]"): # return False return True class MyLoggingAdapter(logging.LoggerAdapter): def process(self, msg, kwargs): return "[{}]{}".format(threading.get_ident() % 10000, msg), kwargs # return '[%s] %s' % (self.extra['connid'], msg), kwargs class SvcManager: def __init__(self): print("Starting TDengine Service Manager") signal.signal(signal.SIGTERM, self.sigIntHandler) signal.signal(signal.SIGINT, self.sigIntHandler) signal.signal(signal.SIGUSR1, self.sigUsrHandler) # different handler! self.inSigHandler = False # self._status = MainExec.STATUS_RUNNING # set inside # _startTaosService() self.svcMgrThread = None def _doMenu(self): choice = "" while True: print("\nInterrupting Service Program, Choose an Action: ") print("1: Resume") print("2: Terminate") print("3: Restart") # Remember to update the if range below # print("Enter Choice: ", end="", flush=True) while choice == "": choice = input("Enter Choice: ") if choice != "": break # done with reading repeated input if choice in ["1", "2", "3"]: break # we are done with whole method print("Invalid choice, please try again.") choice = "" # reset return choice def sigUsrHandler(self, signalNumber, frame): print("Interrupting main thread execution upon SIGUSR1") if self.inSigHandler: # already print("Ignoring repeated SIG...") return # do nothing if it's already not running self.inSigHandler = True choice = self._doMenu() if choice == "1": # TODO: can the sub-process be blocked due to us not reading from # queue? self.sigHandlerResume() elif choice == "2": self.stopTaosService() elif choice == "3": self.stopTaosService() self.startTaosService() else: raise RuntimeError("Invalid menu choice: {}".format(choice)) self.inSigHandler = False def sigIntHandler(self, signalNumber, frame): print("Sig INT Handler starting...") if self.inSigHandler: print("Ignoring repeated SIG_INT...") return self.inSigHandler = True self.stopTaosService() print("INT signal handler returning...") self.inSigHandler = False def sigHandlerResume(self): print("Resuming TDengine service manager thread (main thread)...\n\n") def _checkServiceManagerThread(self): if self.svcMgrThread: # valid svc mgr thread if self.svcMgrThread.isStopped(): # done? self.svcMgrThread.procIpcBatch() # one last time. TODO: appropriate? self.svcMgrThread = None # no more def _procIpcAll(self): while self.svcMgrThread: # for as long as the svc mgr thread is still here self.svcMgrThread.procIpcBatch() # regular processing, time.sleep(0.5) # pause, before next round self._checkServiceManagerThread() print( "Service Manager Thread (with subprocess) has ended, main thread now exiting...") def startTaosService(self): if self.svcMgrThread: raise RuntimeError( "Cannot start TAOS service when one may already be running") self.svcMgrThread = ServiceManagerThread() # create the object self.svcMgrThread.start() print("TAOS service started, printing out output...") self.svcMgrThread.procIpcBatch( trimToTarget=10, forceOutput=True) # for printing 10 lines print("TAOS service started") def stopTaosService(self, outputLines=20): print("Terminating Service Manager Thread (SMT) execution...") if not self.svcMgrThread: raise RuntimeError("Unexpected empty svc mgr thread") self.svcMgrThread.stop() if self.svcMgrThread.isStopped(): self.svcMgrThread.procIpcBatch(outputLines) # one last time self.svcMgrThread = None print("----- End of TDengine Service Output -----\n") print("SMT execution terminated") else: print("WARNING: SMT did not terminate as expected") def run(self): self.startTaosService() self._procIpcAll() # pump/process all the messages if self.svcMgrThread: # if sig handler hasn't destroyed it by now self.stopTaosService() # should have started already class ServiceManagerThread: MAX_QUEUE_SIZE = 10000 def __init__(self): self._tdeSubProcess = None self._thread = None self._status = None def getStatus(self): return self._status def isRunning(self): # return self._thread and self._thread.is_alive() return self._status == MainExec.STATUS_RUNNING def isStopping(self): return self._status == MainExec.STATUS_STOPPING def isStopped(self): return self._status == MainExec.STATUS_STOPPED # Start the thread (with sub process), and wait for the sub service # to become fully operational def start(self): if self._thread: raise RuntimeError("Unexpected _thread") if self._tdeSubProcess: raise RuntimeError("TDengine sub process already created/running") self._status = MainExec.STATUS_STARTING self._tdeSubProcess = TdeSubProcess() self._tdeSubProcess.start() self._ipcQueue = Queue() self._thread = threading.Thread( target=self.svcOutputReader, args=(self._tdeSubProcess.getStdOut(), self._ipcQueue)) self._thread.daemon = True # thread dies with the program self._thread.start() # wait for service to start for i in range(0, 10): time.sleep(1.0) # self.procIpcBatch() # don't pump message during start up print("_zz_", end="", flush=True) if self._status == MainExec.STATUS_RUNNING: logger.info("[] TDengine service READY to process requests") return # now we've started # TODO: handle this better? raise RuntimeError("TDengine service did not start successfully") def stop(self): # can be called from both main thread or signal handler print("Terminating TDengine service running as the sub process...") if self.isStopped(): print("Service already stopped") return if self.isStopping(): print("Service is already being stopped") return # Linux will send Control-C generated SIGINT to the TDengine process # already, ref: # https://unix.stackexchange.com/questions/176235/fork-and-how-signals-are-delivered-to-processes if not self._tdeSubProcess: raise RuntimeError("sub process object missing") self._status = MainExec.STATUS_STOPPING self._tdeSubProcess.stop() if self._tdeSubProcess.isRunning(): # still running print( "FAILED to stop sub process, it is still running... pid = {}".format( self.subProcess.pid)) else: self._tdeSubProcess = None # not running any more self.join() # stop the thread, change the status, etc. def join(self): # TODO: sanity check if not self.isStopping(): raise RuntimeError( "Unexpected status when ending svc mgr thread: {}".format( self._status)) if self._thread: self._thread.join() self._thread = None self._status = MainExec.STATUS_STOPPED else: print("Joining empty thread, doing nothing") def _trimQueue(self, targetSize): if targetSize <= 0: return # do nothing q = self._ipcQueue if (q.qsize() <= targetSize): # no need to trim return logger.debug("Triming IPC queue to target size: {}".format(targetSize)) itemsToTrim = q.qsize() - targetSize for i in range(0, itemsToTrim): try: q.get_nowait() except Empty: break # break out of for loop, no more trimming TD_READY_MSG = "TDengine is initialized successfully" def procIpcBatch(self, trimToTarget=0, forceOutput=False): self._trimQueue(trimToTarget) # trim if necessary # Process all the output generated by the underlying sub process, # managed by IO thread print("<", end="", flush=True) while True: try: line = self._ipcQueue.get_nowait() # getting output at fast speed self._printProgress("_o") except Empty: # time.sleep(2.3) # wait only if there's no output # no more output print(".>", end="", flush=True) return # we are done with THIS BATCH else: # got line, printing out if forceOutput: logger.info(line) else: logger.debug(line) print(">", end="", flush=True) _ProgressBars = ["--", "//", "||", "\\\\"] def _printProgress(self, msg): # TODO: assuming 2 chars print(msg, end="", flush=True) pBar = self._ProgressBars[Dice.throw(4)] print(pBar, end="", flush=True) print('\b\b\b\b', end="", flush=True) def svcOutputReader(self, out: IO, queue): # Important Reference: https://stackoverflow.com/questions/375427/non-blocking-read-on-a-subprocess-pipe-in-python # print("This is the svcOutput Reader...") # for line in out : for line in iter(out.readline, b''): # print("Finished reading a line: {}".format(line)) # print("Adding item to queue...") line = line.decode("utf-8").rstrip() # This might block, and then causing "out" buffer to block queue.put(line) self._printProgress("_i") if self._status == MainExec.STATUS_STARTING: # we are starting, let's see if we have started if line.find(self.TD_READY_MSG) != -1: # found self._status = MainExec.STATUS_RUNNING # Trim the queue if necessary: TODO: try this 1 out of 10 times self._trimQueue(self.MAX_QUEUE_SIZE * 9 // 10) # trim to 90% size if self.isStopping(): # TODO: use thread status instead # WAITING for stopping sub process to finish its outptu print("_w", end="", flush=True) # queue.put(line) # meaning sub process must have died print("\nNo more output from IO thread managing TDengine service") out.close() class TdeSubProcess: def __init__(self): self.subProcess = None def getStdOut(self): return self.subProcess.stdout def isRunning(self): return self.subProcess is not None def getBuildPath(self): selfPath = os.path.dirname(os.path.realpath(__file__)) if ("community" in selfPath): projPath = selfPath[:selfPath.find("communit")] else: projPath = selfPath[:selfPath.find("tests")] for root, dirs, files in os.walk(projPath): if ("taosd" in files): rootRealPath = os.path.dirname(os.path.realpath(root)) if ("packaging" not in rootRealPath): buildPath = root[:len(root) - len("/build/bin")] break return buildPath def start(self): ON_POSIX = 'posix' in sys.builtin_module_names taosdPath = self.getBuildPath() + "/build/bin/taosd" cfgPath = self.getBuildPath() + "/test/cfg" svcCmd = [taosdPath, '-c', cfgPath] # svcCmd = ['vmstat', '1'] if self.subProcess: # already there raise RuntimeError("Corrupt process state") self.subProcess = subprocess.Popen( svcCmd, stdout=subprocess.PIPE, # bufsize=1, # not supported in binary mode close_fds=ON_POSIX) # had text=True, which interferred with reading EOF def stop(self): if not self.subProcess: print("Sub process already stopped") return retCode = self.subProcess.poll() if retCode: # valid return code, process ended self.subProcess = None else: # process still alive, let's interrupt it print( "Sub process is running, sending SIG_INT and waiting for it to terminate...") # sub process should end, then IPC queue should end, causing IO # thread to end self.subProcess.send_signal(signal.SIGINT) try: self.subProcess.wait(10) except subprocess.TimeoutExpired as err: print("Time out waiting for TDengine service process to exit") else: print("TDengine service process terminated successfully from SIG_INT") self.subProcess = None class ClientManager: def __init__(self): print("Starting service manager") signal.signal(signal.SIGTERM, self.sigIntHandler) signal.signal(signal.SIGINT, self.sigIntHandler) self._status = MainExec.STATUS_RUNNING self.tc = None def sigIntHandler(self, signalNumber, frame): if self._status != MainExec.STATUS_RUNNING: print("Repeated SIGINT received, forced exit...") # return # do nothing if it's already not running sys.exit(-1) self._status = MainExec.STATUS_STOPPING # immediately set our status print("Terminating program...") self.tc.requestToStop() def _printLastNumbers(self): # to verify data durability dbManager = DbManager(resetDb=False) dbc = dbManager.getDbConn() if dbc.query("show databases") == 0: # no databae return if dbc.query("show tables") == 0: # no tables return dbc.execute("use db") sTbName = dbManager.getFixedSuperTableName() # get all regular tables # TODO: analyze result set later dbc.query("select TBNAME from db.{}".format(sTbName)) rTables = dbc.getQueryResult() bList = TaskExecutor.BoundedList() for rTbName in rTables: # regular tables dbc.query("select speed from db.{}".format(rTbName[0])) numbers = dbc.getQueryResult() for row in numbers: # print("<{}>".format(n), end="", flush=True) bList.add(row[0]) print("Top numbers in DB right now: {}".format(bList)) print("TDengine client execution is about to start in 2 seconds...") time.sleep(2.0) dbManager = None # release? def prepare(self): self._printLastNumbers() def run(self): if gConfig.auto_start_service: svcMgr = SvcManager() svcMgr.startTaosService() self._printLastNumbers() dbManager = DbManager() # Regular function thPool = ThreadPool(gConfig.num_threads, gConfig.max_steps) self.tc = ThreadCoordinator(thPool, dbManager) self.tc.run() # print("exec stats: {}".format(self.tc.getExecStats())) # print("TC failed = {}".format(self.tc.isFailed())) if gConfig.auto_start_service: svcMgr.stopTaosService() # Print exec status, etc., AFTER showing messages from the server self.conclude() # print("TC failed (2) = {}".format(self.tc.isFailed())) # Linux return code: ref https://shapeshed.com/unix-exit-codes/ return 1 if self.tc.isFailed() else 0 def conclude(self): self.tc.printStats() self.tc.getDbManager().cleanUp() class MainExec: STATUS_STARTING = 1 STATUS_RUNNING = 2 STATUS_STOPPING = 3 STATUS_STOPPED = 4 @classmethod def runClient(cls): clientManager = ClientManager() return clientManager.run() @classmethod def runService(cls): svcManager = SvcManager() svcManager.run() @classmethod def runTemp(cls): # for debugging purposes # # Hack to exercise reading from disk, imcreasing coverage. TODO: fix # dbc = dbState.getDbConn() # sTbName = dbState.getFixedSuperTableName() # dbc.execute("create database if not exists db") # if not dbState.getState().equals(StateEmpty()): # dbc.execute("use db") # rTables = None # try: # the super table may not exist # sql = "select TBNAME from db.{}".format(sTbName) # logger.info("Finding out tables in super table: {}".format(sql)) # dbc.query(sql) # TODO: analyze result set later # logger.info("Fetching result") # rTables = dbc.getQueryResult() # logger.info("Result: {}".format(rTables)) # except taos.error.ProgrammingError as err: # logger.info("Initial Super table OPS error: {}".format(err)) # # sys.exit() # if ( not rTables == None): # # print("rTables[0] = {}, type = {}".format(rTables[0], type(rTables[0]))) # try: # for rTbName in rTables : # regular tables # ds = dbState # logger.info("Inserting into table: {}".format(rTbName[0])) # sql = "insert into db.{} values ('{}', {});".format( # rTbName[0], # ds.getNextTick(), ds.getNextInt()) # dbc.execute(sql) # for rTbName in rTables : # regular tables # dbc.query("select * from db.{}".format(rTbName[0])) # TODO: check success failure # logger.info("Initial READING operation is successful") # except taos.error.ProgrammingError as err: # logger.info("Initial WRITE/READ error: {}".format(err)) # Sandbox testing code # dbc = dbState.getDbConn() # while True: # rows = dbc.query("show databases") # print("Rows: {}, time={}".format(rows, time.time())) return def main(): # Super cool Python argument library: # https://docs.python.org/3/library/argparse.html parser = argparse.ArgumentParser( formatter_class=argparse.RawDescriptionHelpFormatter, description=textwrap.dedent('''\ TDengine Auto Crash Generator (PLEASE NOTICE the Prerequisites Below) --------------------------------------------------------------------- 1. You build TDengine in the top level ./build directory, as described in offical docs 2. You run the server there before this script: ./build/bin/taosd -c test/cfg ''')) # parser.add_argument('-a', '--auto-start-service', action='store_true', # help='Automatically start/stop the TDengine service (default: false)') # parser.add_argument('-c', '--connector-type', action='store', default='native', type=str, # help='Connector type to use: native, rest, or mixed (default: 10)') # parser.add_argument('-d', '--debug', action='store_true', # help='Turn on DEBUG mode for more logging (default: false)') # parser.add_argument('-e', '--run-tdengine', action='store_true', # help='Run TDengine service in foreground (default: false)') # parser.add_argument('-l', '--larger-data', action='store_true', # help='Write larger amount of data during write operations (default: false)') # parser.add_argument('-p', '--per-thread-db-connection', action='store_true', # help='Use a single shared db connection (default: false)') # parser.add_argument('-r', '--record-ops', action='store_true', # help='Use a pair of always-fsynced fils to record operations performing + performed, for power-off tests (default: false)') # parser.add_argument('-s', '--max-steps', action='store', default=1000, type=int, # help='Maximum number of steps to run (default: 100)') # parser.add_argument('-t', '--num-threads', action='store', default=5, type=int, # help='Number of threads to run (default: 10)') # parser.add_argument('-x', '--continue-on-exception', action='store_true', # help='Continue execution after encountering unexpected/disallowed errors/exceptions (default: false)') parser.add_argument( '-a', '--auto-start-service', action='store_true', help='Automatically start/stop the TDengine service (default: false)') parser.add_argument( '-c', '--connector-type', action='store', default='native', type=str, help='Connector type to use: native, rest, or mixed (default: 10)') parser.add_argument( '-d', '--debug', action='store_true', help='Turn on DEBUG mode for more logging (default: false)') parser.add_argument( '-e', '--run-tdengine', action='store_true', help='Run TDengine service in foreground (default: false)') parser.add_argument( '-l', '--larger-data', action='store_true', help='Write larger amount of data during write operations (default: false)') parser.add_argument( '-p', '--per-thread-db-connection', action='store_true', help='Use a single shared db connection (default: false)') parser.add_argument( '-r', '--record-ops', action='store_true', help='Use a pair of always-fsynced fils to record operations performing + performed, for power-off tests (default: false)') parser.add_argument( '-s', '--max-steps', action='store', default=1000, type=int, help='Maximum number of steps to run (default: 100)') parser.add_argument( '-t', '--num-threads', action='store', default=5, type=int, help='Number of threads to run (default: 10)') parser.add_argument( '-x', '--continue-on-exception', action='store_true', help='Continue execution after encountering unexpected/disallowed errors/exceptions (default: false)') global gConfig gConfig = parser.parse_args() # Logging Stuff global logger _logger = logging.getLogger('CrashGen') # real logger _logger.addFilter(LoggingFilter()) ch = logging.StreamHandler() _logger.addHandler(ch) # Logging adapter, to be used as a logger logger = MyLoggingAdapter(_logger, []) if (gConfig.debug): logger.setLevel(logging.DEBUG) # default seems to be INFO else: logger.setLevel(logging.INFO) Dice.seed(0) # initial seeding of dice # Run server or client if gConfig.run_tdengine: # run server MainExec.runService() else: return MainExec.runClient() if __name__ == "__main__": exitCode = main() # print("Exiting with code: {}".format(exitCode)) sys.exit(exitCode)