diff --git a/.travis.yml b/.travis.yml index 2a0aa6372be2fc50d596e296dec1f936cf9331cd..7f8d0fdb67662f9abf63db3617f04dbc86513aa1 100644 --- a/.travis.yml +++ b/.travis.yml @@ -61,6 +61,11 @@ matrix: cd ${TRAVIS_BUILD_DIR}/tests ./test-all.sh smoke || travis_terminate $? + sleep 1 + + cd ${TRAVIS_BUILD_DIR}/tests/pytest + ./crash_gen.sh -p -t 5 -s 50|| travis_terminate $? + sleep 1 cd ${TRAVIS_BUILD_DIR}/tests/pytest ./valgrind-test.sh 2>&1 > mem-error-out.log diff --git a/tests/pytest/crash_gen.py b/tests/pytest/crash_gen.py index 49c428b7f108b785f564564915aa2f8dbf52127e..c8b34fe7f36d31cce2b80a4d0277af31d44e05a6 100755 --- a/tests/pytest/crash_gen.py +++ b/tests/pytest/crash_gen.py @@ -1,4 +1,4 @@ -#-----!/usr/bin/python3.7 +# -----!/usr/bin/python3.7 ################################################################### # Copyright (c) 2016 by TAOS Technologies, Inc. # All rights reserved. @@ -11,7 +11,31 @@ ################################################################### # -*- coding: utf-8 -*- -from __future__ import annotations # 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 +# 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 @@ -22,71 +46,48 @@ import traceback if sys.version_info[0] < 3: raise Exception("Must be using Python 3") -import getopt -import argparse -import copy -import requests - -import threading -import random -import time -import logging -import datetime -import textwrap -import requests -from requests.auth import HTTPBasicAuth - -from typing import List -from typing import Dict -from typing import Set -from typing import IO -from queue import Queue, Empty - -from util.log import * -from util.dnodes import * -from util.cases import * -from util.sql import * - -import crash_gen -import taos -# Global variables, tried to keep a small number. +# 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 +gConfig = argparse.Namespace() # Dummy value, will be replaced later logger = None -def runThread(wt: WorkerThread): + +def runThread(wt: WorkerThread): wt.run() + class CrashGenError(Exception): def __init__(self, msg=None, errno=None): - self.msg = msg + 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 + 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._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 + if (gConfig.per_thread_db_connection): # type: ignore # print("connector_type = {}".format(gConfig.connector_type)) - self._dbConn = DbConn.createNative() if (gConfig.connector_type == 'native') else DbConn.createRest() + self._dbConn = DbConn.createNative() if ( + gConfig.connector_type == 'native') else DbConn.createRest() - self._dbInUse = False # if "use db" was executed already + self._dbInUse = False # if "use db" was executed already def logDebug(self, msg): logger.debug(" TRD[{}] {}".format(self._tid, msg)) @@ -98,137 +99,153 @@ class WorkerThread: return self._dbInUse def useDb(self): - if ( not self._dbInUse ): + if (not self._dbInUse): self.execSql("use db") self._dbInUse = True def getTaskExecutor(self): - return self._tc.getTaskExecutor() + return self._tc.getTaskExecutor() def start(self): self._thread.start() # AFTER the thread is recorded - def run(self): + 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 + if (gConfig.per_thread_db_connection): # type: ignore logger.debug("Worker thread openning database connection") self._dbConn.open() - self._doTaskLoop() - + self._doTaskLoop() + # clean up - if ( gConfig.per_thread_db_connection ): # type: ignore + if (gConfig.per_thread_db_connection): # type: ignore self._dbConn.close() - def _doTaskLoop(self) : + def _doTaskLoop(self): # while self._curStep < self._pool.maxSteps: # tc = ThreadCoordinator(None) - while True: - tc = self._tc # Thread Coordinator, the overall master + 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)) + 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)) + 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...") + 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)) + 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__)) + 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)) + logger.debug( + "[TRD] Worker thread [{}] finished executing task".format( + self._tid)) + + self._dbInUse = False # there may be changes between steps - 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 ): + 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 ): + 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() ): + 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 - + 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() + 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 + def tapStepGate(self): # give it a tap, release the thread waiting there self.verifyThreadAlive() - self.verifyThreadMain() # only allowed for main thread - + 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 + 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) + 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) + 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() + 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 + 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) + # 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._curStep = -1 # first step is 0 self._pool = pool # self._wd = wd - self._te = None # prepare for every new step + 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._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._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 : + def getDbManager(self) -> DbManager: return self._dbManager def crossStepBarrier(self): @@ -238,89 +255,103 @@ class ThreadCoordinator: self._runStatus = MainExec.STATUS_STOPPING self._execStats.registerFailure("User Interruption") - def run(self): + def run(self): self._pool.createAndStartThreads(self) # Coordinate all threads step by step - self._curStep = -1 # not started yet - maxSteps = gConfig.max_steps # type: ignore - self._execStats.startExec() # start the stop watch + self._curStep = -1 # not started yet + maxSteps = gConfig.max_steps # type: ignore + self._execStats.startExec() # start the stop watch transitionFailed = False hasAbortedTask = False - while(self._curStep < maxSteps-1 and - (not transitionFailed) and - (self._runStatus==MainExec.STATUS_RUNNING) and - (not hasAbortedTask)): # maxStep==10, last curStep should be 9 - - if not gConfig.debug: - print(".", end="", flush=True) # print this only if we are not in debug mode + while(self._curStep < maxSteps - 1 and + (not transitionFailed) and + (self._runStatus == MainExec.STATUS_RUNNING) and + (not hasAbortedTask)): # maxStep==10, last curStep should be 9 + + if not gConfig.debug: + # print this only if we are not in debug mode + print(".", end="", flush=True) logger.debug("[TRD] Main thread going to sleep") # Now main thread (that's us) is ready to enter a step - self.crossStepBarrier() # let other threads go past the pool barrier, but wait at the thread gate - self._stepBarrier.reset() # Other worker threads should now be at the "gate" + # let other threads go past the pool barrier, but wait at the + # thread gate + self.crossStepBarrier() + self._stepBarrier.reset() # Other worker threads should now be at the "gate" # 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. + # We use this period to do house keeping work, when all worker + # threads are QUIET. hasAbortedTask = False - for task in self._executedTasks : - if task.isAborted() : + for task in self._executedTasks: + if task.isAborted(): print("Task aborted: {}".format(task)) hasAbortedTask = True break - if hasAbortedTask : # do transition only if tasks are error free + if hasAbortedTask: # do transition only if tasks are error free self._execStats.registerFailure("Aborted Task Encountered") - else: + else: try: sm = self._dbManager.getStateMachine() logger.debug("[STT] starting transitions") - sm.transition(self._executedTasks) # at end of step, transiton the DB state + # 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() : + # 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 + # 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 + 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._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 + # continue # don't do that, need to tap all threads at + # end, and maybe signal them to stop else: - raise + raise # finally: # pass - - self.resetExecutedTasks() # clear the tasks after we are done + + self.resetExecutedTasks() # clear the tasks after we are done # Get ready for next step logger.debug("<-- Step {} finished".format(self._curStep)) - self._curStep += 1 # we are about to get into next step. TODO: race condition here! - logger.debug("\r\n\n--> Step {} starts with main thread waking up".format(self._curStep)) # Now not all threads had time to go to sleep + 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 - if not transitionFailed: # only if not failed + 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 - self.tapAllThreads() # Worker threads will wake up at this point, and each execute it's own task + 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() logger.debug("Main thread ready to finish up...") - if not transitionFailed: # only in regular situations - self.crossStepBarrier() # Cross it one last time, after all threads finish + if not transitionFailed: # only in regular situations + self.crossStepBarrier() # Cross it one last time, after all threads finish self._stepBarrier.reset() logger.debug("Main thread in exclusive zone...") - self._te = None # No more executor, time to end + 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 + self.tapAllThreads() # Let the threads run one last time logger.debug("Main thread joining all threads") - self._pool.joinAll() # Get all threads to finish + self._pool.joinAll() # Get all threads to finish logger.info("\nAll worker threads finished") self._execStats.endExec() @@ -333,24 +364,27 @@ class ThreadCoordinator: def getExecStats(self): return self._execStats - def tapAllThreads(self): # in a deterministic manner + def tapAllThreads(self): # in a deterministic manner wakeSeq = [] - for i in range(self._pool.numThreads): # generate a random sequence - if Dice.throw(2) == 1 : + 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))) + logger.debug( + "[TRD] Main thread waking up worker threads: {}".format( + str(wakeSeq))) # TODO: set dice seed to a deterministic value for i in wakeSeq: - self._pool.threadList[i].tapStepGate() # TODO: maybe a bit too deep?! - time.sleep(0) # yield + # TODO: maybe a bit too deep?! + self._pool.threadList[i].tapStepGate() + time.sleep(0) # yield def isRunning(self): - return self._te != None + return self._te is not None - def fetchTask(self) -> Task : - if ( not self.isRunning() ): # no task + 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 @@ -361,31 +395,36 @@ class ThreadCoordinator: # 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? - taskType = self.getDbManager().getStateMachine().pickTaskType() # pick a task type for current state - return taskType(self.getDbManager(), self._execStats) # create a task from it + # 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 + 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] - + 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) + 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 + workerThread.start() # start, but should block immediately before step 0 def joinAll(self): for workerThread in self.threadList: @@ -394,21 +433,24 @@ class ThreadPool: # 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 + 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) + 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): + def push(self): with self._lock: - # if ( self.isEmpty() ): - # self.lastIndex = self.firstIndex + # if ( self.isEmpty() ): + # self.lastIndex = self.firstIndex # return self.firstIndex # Otherwise we have something self.lastIndex += 1 @@ -418,12 +460,12 @@ class LinearQueue(): def pop(self): with self._lock: - if ( self.isEmpty() ): - # raise RuntimeError("Cannot pop an empty queue") - return False # TODO: None? - + if (self.isEmpty()): + # raise RuntimeError("Cannot pop an empty queue") + return False # TODO: None? + index = self.firstIndex - if ( index in self.inUse ): + if (index in self.inUse): return False self.firstIndex += 1 @@ -441,33 +483,35 @@ class LinearQueue(): 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)) + 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? + self.inUse.remove(i) # KeyError possible, TODO: why? def size(self): return self.lastIndex + 1 - self.firstIndex def pickAndAllocate(self): - if ( self.isEmpty() ): + if (self.isEmpty()): return None with self._lock: - cnt = 0 # counting the interations + cnt = 0 # counting the interations while True: cnt += 1 - if ( cnt > self.size()*10 ): # 10x iteration already + 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 ( not ret in self.inUse ): + 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" @@ -480,7 +524,8 @@ class DbConn: elif connType == cls.TYPE_REST: return DbConnRest() else: - raise RuntimeError("Unexpected connection type: {}".format(connType)) + raise RuntimeError( + "Unexpected connection type: {}".format(connType)) @classmethod def createNative(cls): @@ -495,18 +540,21 @@ class DbConn: self._type = self.TYPE_INVALID def open(self): - if ( self.isOpen ): + 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)) + 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") + 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') @@ -515,83 +563,99 @@ class DbConn: # self._cursor.execute('use db') # tdSql.execute('show databases') - def queryScalar(self, sql) -> int : + def queryScalar(self, sql) -> int: return self._queryAny(sql) - def queryString(self, sql) -> str : + 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") + 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 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)) + 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._url = "http://localhost:6020/rest/sql" # fixed for now self._result = None - def openByType(self): # Open connection - pass # do nothing, always open - + 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") + 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): - r = requests.post(self._url, - data = sql, - auth = HTTPBasicAuth('root', 'taosdata')) + r = requests.post(self._url, + data=sql, + auth=HTTPBasicAuth('root', 'taosdata')) rj = r.json() # Sanity check for the "Json Result" - if (not 'status' in rj): + if ('status' not in rj): raise RuntimeError("No status in REST response") - if rj['status'] == 'error': # clearly reported error - if (not 'code' in rj): # error without code - raise RuntimeError("REST error return without code") - errno = rj['code'] # May need to massage this in the future + 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' + 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'])) + 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 + self._result = rj return nRows - def execute(self, sql): - if ( not self.isOpen ): - raise RuntimeError("Cannot execute database commands until connection is open") + 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)) + logger.debug( + "[SQL-REST] Execution Result, nRows = {}, SQL = {}".format(nRows, sql)) return nRows - def query(self, sql) : # return rows affected + def query(self, sql): # return rows affected return self.execute(sql) def getQueryResult(self): @@ -605,48 +669,59 @@ class DbConnRest(DbConn): def getResultCols(self): print(self._result) raise RuntimeError("TBD") - + + class DbConnNative(DbConn): def __init__(self): super().__init__() self._type = self.TYPE_REST - self._conn = None + self._conn = None self._cursor = None - - def openByType(self): # Open connection - cfgPath = "../../build/test/cfg" - self._conn = taos.connect(host="127.0.0.1", config=cfgPath) # TODO: make configurable + + def openByType(self): # Open connection + cfgPath = "../../build/test/cfg" + self._conn = taos.connect( + host="127.0.0.1", + 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 + # self._cursor.execute('use db') # do this at the beginning of every + # step # Open connection self._tdSql = TDSql() self._tdSql.init(self._cursor) - + def close(self): - if ( not self.isOpen ): - raise RuntimeError("Cannot clean up database until connection is open") + 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") + 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)) + 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") + 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)) + logger.debug( + "[SQL] Query Result, nRows = {}, SQL = {}".format( + nRows, sql)) return nRows # results are in: return self._tdSql.queryResult @@ -659,13 +734,13 @@ class DbConnNative(DbConn): 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_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 + STATE_HAS_DATA = 3 # we have some data in the table _stateNames = ["Invalid", "Empty", "DB_Only", "Table_Only", "Has_Data"] STATE_VAL_IDX = 0 @@ -680,7 +755,8 @@ class AnyState: self._info = self.getInfo() def __str__(self): - return self._stateNames[self._info[self.STATE_VAL_IDX] + 1] # -1 hack to accomodate the STATE_INVALID case + # -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") @@ -691,7 +767,9 @@ class AnyState: elif isinstance(other, AnyState): return self.getValIndex() == other.getValIndex() else: - raise RuntimeError("Unexpected comparison, type = {}".format(type(other))) + raise RuntimeError( + "Unexpected comparison, type = {}".format( + type(other))) def verifyTasksToState(self, tasks, newState): raise RuntimeError("Must be overriden by child classes") @@ -701,55 +779,65 @@ class AnyState: 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 : + 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)) + 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 : + for task in tasks: if not isinstance(task, cls): continue - exists = True # we have a valid instance + 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)) + if (exists and sCnt <= 0): + raise RuntimeError( + "Unexpected zero success for task: {}".format(cls)) def assertNoTask(self, tasks, cls): - for task in tasks : + 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__)) + 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 : + for task in tasks: if isinstance(task, cls): if task.isSuccess(): - raise RuntimeError("Unexpected successful task: {}".format(cls)) + raise RuntimeError( + "Unexpected successful task: {}".format(cls)) def hasSuccess(self, tasks, cls): - for task in tasks : + for task in tasks: if not isinstance(task, cls): continue if task.isSuccess(): @@ -757,35 +845,40 @@ class AnyState: return False def hasTask(self, tasks, cls): - for task in tasks : + 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 + 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 + 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 - self.assertAtMostOneSuccess(tasks, TaskCreateDb) # we must have at most one. TODO: compare numbers + 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): @@ -797,32 +890,34 @@ class StateDbOnly(AnyState): ] def verifyTasksToState(self, tasks, newState): - if ( not self.hasTask(tasks, TaskCreateDb) ): - self.assertAtMostOneSuccess(tasks, TaskDropDb) # only if we don't create any more + 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 + # 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) ): + # 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 + # 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, + # # 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 [ @@ -833,9 +928,11 @@ class StateSuperTableOnly(AnyState): ] def verifyTasksToState(self, tasks, newState): - if ( self.hasSuccess(tasks, TaskDropSuperTable) ): # we are able to drop the table + if (self.hasSuccess(tasks, TaskDropSuperTable) + ): # we are able to drop the table #self.assertAtMostOneSuccess(tasks, TaskDropSuperTable) - self.hasSuccess(tasks, TaskCreateSuperTable) # we must have had recreted it + # 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 @@ -849,6 +946,7 @@ class StateSuperTableOnly(AnyState): # raise RuntimeError("Unexpected no-success scenarios") # TODO: need to revamp!! + class StateHasData(AnyState): def getInfo(self): return [ @@ -859,13 +957,15 @@ class StateHasData(AnyState): ] def verifyTasksToState(self, tasks, newState): - if ( newState.equals(AnyState.STATE_EMPTY) ): + 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 - self.assertNoTask(tasks, TaskDropDb) # we must have drop_db task + 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 @@ -873,19 +973,26 @@ class StateHasData(AnyState): # 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 - self.assertNoTask(tasks, TaskDropDb) # we shouldn't have dropped it - if (not self.hasTask(tasks, TaskCreateSuperTable)) : # if we didn't create the table - self.assertNoTask(tasks, TaskDropSuperTable) # we should not have a task that drops it + 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 - self._stateWeights = [1,3,5,15] # transitition target probabilities, indexed with value of STATE_EMPTY, STATE_DB_ONLY, etc. - + 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 @@ -893,142 +1000,178 @@ class StateMechine: 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 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 + allTaskClasses = StateTransitionTask.__subclasses__() # all state transition tasks firstTaskTypes = [] for tc in allTaskClasses: - # t = tc(self) # create task object + # 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 == None: # does not change end state - continue # no use, do nothing - for tc in allTaskClasses: # what task can further begin from there? + # 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 + 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))) + 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?! + 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())) + logger.debug( + "[STT] empty database found, between {} and {}".format( + ts, time.time())) return StateEmpty() - dbc.execute("use db") # did not do this when openning connection, and this is NOT the worker thread, which does this on their own - if dbc.query("show tables") == 0 : # no tables + # 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())) + 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 + 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())) + logger.debug( + "[STT] SUPER_TABLE_ONLY found, between {} and {}".format( + ts, time.time())) return StateSuperTableOnly() - else: # has actual tables + else: # has actual tables # logger.debug("Found HAS_DATA state") - logger.debug("[STT] HAS_DATA found, between {} and {}".format(ts, time.time())) + 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 + if (len(tasks) == 0): # before 1st step, or otherwise empty logger.debug("[STT] Starting State: {}".format(self._curState)) - return # do nothing + return # do nothing - self._dbConn.execute("show dnodes") # this should show up in the server log, separating steps + # 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 + # 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 + # 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 + # 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 + # self.assertAtMostOneSuccess(tasks, DropFixedTableTask) # not + # really in case of drop-create-drop # if self._state.canAddData(): - # self.assertIfExistThenSuccess(tasks, AddFixedDataTask) # not true actually + # 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)) - self._curState.verifyTasksToState(tasks, newState) # can old state move to new state through the tasks? + # can old state move to new state through the tasks? + self._curState.verifyTasksToState(tasks, newState) self._curState = newState def pickTaskType(self): - taskTypes = self.getTaskTypes() # all the task types we can choose from at curent state + # all the task types we can choose from at curent state + taskTypes = self.getTaskTypes() weights = [] for tt in taskTypes: endState = tt.getEndState() - if endState != None : - weights.append(self._stateWeights[endState.getValIndex()]) # TODO: change to a method + if endState is not None: + # TODO: change to a method + weights.append(self._stateWeights[endState.getValIndex()]) else: - weights.append(10) # read data task, default to 10: TODO: change to a constant + # 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))) + # logger.debug(" (weighted random:{}/{}) ".format(i, len(taskTypes))) return taskTypes[i] - def _weighted_choice_sub(self, weights): # ref: https://eli.thegreenplace.net/2010/01/22/weighted-random-generation-in-python/ - rnd = random.random() * sum(weights) # TODO: use our dice to ensure it being determinstic? + # 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): + + +class DbManager(): + def __init__(self, resetDb=True): self.tableNumQueue = LinearQueue() - self._lastTick = self.setupLastTick() # datetime.datetime(2019, 1, 1) # initial date time tick - self._lastInt = 0 # next one is initial integer + # 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() + self._dbConn = DbConn.createNative() if ( + gConfig.connector_type == 'native') else DbConn.createRest() try: - self._dbConn.open() # may throw taos.error.ProgrammingError: disconnected + 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.") + 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: + raise + except BaseException: print("[=] Unexpected exception") - raise + raise + + if resetDb: + self._dbConn.resetDb() # drop and recreate DB - if resetDb : - self._dbConn.resetDb() # drop and recreate DB + # Do this after dbConn is in proper shape + self._stateMachine = StateMechine(self._dbConn) - self._stateMachine = StateMechine(self._dbConn) # Do this after dbConn is in proper shape - def getDbConn(self): return self._dbConn - def getStateMachine(self) -> StateMechine : + def getStateMachine(self) -> StateMechine: return self._stateMachine # def getState(self): @@ -1043,15 +1186,18 @@ class DbManager(): def setupLastTick(self): t1 = datetime.datetime(2020, 6, 1) t2 = datetime.datetime.now() - elSec = int(t2.timestamp() - t1.timestamp()) # maybe a very large number, takes 69 years to exceed Python int range - elSec2 = ( elSec % (8 * 12 * 30 * 24 * 60 * 60 / 500 ) ) * 500 # a number representing seconds within 10 years + # 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 + 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 + def pickAndAllocateTable(self): # pick any table, and "use" it return self.tableNumQueue.pickAndAllocate() def addTable(self): @@ -1063,15 +1209,16 @@ class DbManager(): def getFixedSuperTableName(cls): return "fs_table" - def releaseTable(self, i): # return the table back, so others can use it + 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 + with self._lock: # prevent duplicate tick + if Dice.throw(10) == 0: # 1 in 10 chance return self._lastTick + datetime.timedelta(0, -100) - else: # regular - self._lastTick += datetime.timedelta(0, 1) # add one second to it + else: # regular + # add one second to it + self._lastTick += datetime.timedelta(0, 1) return self._lastTick def getNextInt(self): @@ -1080,29 +1227,31 @@ class DbManager(): 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()) + 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 + tblNum = self.tableNumQueue.pop() # TODO: race condition! + if (not tblNum): # maybe false return False - + return "table_{}".format(tblNum) def cleanUp(self): - self._dbConn.close() + self._dbConn.close() + class TaskExecutor(): class BoundedList: - def __init__(self, size = 10): + def __init__(self, size=10): self._size = size self._list = [] - def add(self, n: int) : - if not self._list: # empty + def add(self, n: int): + if not self._list: # empty self._list.append(n) return # now we should insert @@ -1110,22 +1259,22 @@ class TaskExecutor(): 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 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 + 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 - + 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 : + 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): @@ -1143,7 +1292,7 @@ class TaskExecutor(): def getCurStep(self): return self._curStep - def execute(self, task: Task, wt: WorkerThread): # execute a task on a thread + def execute(self, task: Task, wt: WorkerThread): # execute a task on a thread task.execute(wt) def recordDataMark(self, n: int): @@ -1156,128 +1305,147 @@ class TaskExecutor(): # 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 + 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): + def __init__(self, dbManager: DbManager, execStats: ExecutionStats): self._dbManager = dbManager - self._workerThread = None + self._workerThread = None self._err = None self._aborted = False self._curStep = None - self._numRows = None # Number of rows affected + self._numRows = None # Number of rows affected - # Assign an incremental task serial number + # 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 + self._lastSql = "" # last SQL executed/attempted def isSuccess(self): - return self._err == None + return self._err is None def isAborted(self): return self._aborted - def clone(self): # TODO: why do we need this again? + 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)) + self._workerThread.logDebug( + "Step[{}.{}] {}".format( + self._curStep, self._taskNum, msg)) def logInfo(self, msg): - self._workerThread.logInfo("Step[{}.{}] {}".format(self._curStep, self._taskNum, 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__)) + 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 + self._workerThread = wt # type: ignore te = wt.getTaskExecutor() self._curStep = te.getCurStep() - self.logDebug("[-] executing task {}...".format(self.__class__.__name__)) + self.logDebug( + "[-] executing task {}...".format(self.__class__.__name__)) self._err = None - self._execStats.beginTaskType(self.__class__.__name__) # mark beginning + self._execStats.beginTaskType( + self.__class__.__name__) # mark beginning try: - self._executeInternal(te, wt) # TODO: no return value? + 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 ( errno2 in [0x200, 0x360, 0x362, 0x36A, 0x36B, 0x36D, 0x381, 0x380, 0x383, 0x503, 0x600, - 1000 # REST catch-all error - ]) : # allowed errors - self.logDebug("[=] Acceptable Taos library exception: errno=0x{:X}, msg: {}, SQL: {}".format(errno2, err, self._lastSql)) + errno2 = err.errno if ( + err.errno > 0) else 0x80000000 + err.errno # correct error scheme + if (errno2 in [0x200, 0x360, 0x362, 0x36A, 0x36B, 0x36D, 0x381, 0x380, 0x383, 0x503, 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 + self._err = err else: - errMsg = "[=] Unexpected Taos library exception: errno=0x{:X}, msg: {}, SQL: {}".format(errno2, err, self._lastSql) + 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 - else: # non-debug - print("\n\n----------------------------\nProgram ABORTED Due to Unexpected TAOS Error: \n\n{}\n".format(errMsg) + + if gConfig.debug: + raise # so that we see full stack + else: # non-debug + 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 : + except Exception as e: self.logInfo("Non-TAOS exception encountered") - self._err = e + self._err = e self._aborted = True traceback.print_exc() - except : - self.logDebug("[=] Unexpected exception, SQL: {}".format(self._lastSql)) + 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")) - self._execStats.incExecCount(self.__class__.__name__, self.isSuccess()) # TODO: merge with above. + + 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 + 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 + 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 + def getQueryResult(self, wt: WorkerThread): # execute an SQL on the worker thread return wt.getQueryResult() - class ExecutionStats: def __init__(self): - self._execTimes: Dict[str, [int, int]] = {} # total/success times for a task + # 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._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) + return "[ExecStats: _failed={}, _failureReason={}".format( + self._failed, self._failureReason) def isFailed(self): - return self._failed == True + return self._failed def startExec(self): self._execStartTime = time.time() @@ -1285,24 +1453,24 @@ class ExecutionStats: def endExec(self): self._elapsedTime = time.time() - self._execStartTime - def incExecCount(self, klassName, isSuccess): # TODO: add a lock here + 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 + 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 + 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 + 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 + if self._tasksInProgress == 0: # all tasks have stopped self._accRunTime += (time.time() - self._firstTaskStartTime) self._firstTaskStartTime = None @@ -1311,23 +1479,36 @@ class ExecutionStats: 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( + "----------------------------------------------------------------------") + 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(): + 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("----------------------------------------------------------------------") - + 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): @@ -1337,12 +1518,12 @@ class StateTransitionTask(Task): SMALL_NUMBER_OF_RECORDS = 3 @classmethod - def getInfo(cls): # each sub class should supply their own information + def getInfo(cls): # each sub class should supply their own information raise RuntimeError("Overriding method expected") - _endState = None + _endState = None @classmethod - def getEndState(cls): # TODO: optimize by calling it fewer times + def getEndState(cls): # TODO: optimize by calling it fewer times raise RuntimeError("Overriding method expected") # @classmethod @@ -1364,18 +1545,20 @@ class StateTransitionTask(Task): def execute(self, wt: WorkerThread): super().execute(wt) - + + class TaskCreateDb(StateTransitionTask): @classmethod def getEndState(cls): - return StateDbOnly() + return StateDbOnly() @classmethod def canBeginFrom(cls, state: AnyState): return state.canCreateDb() def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): - self.execWtSql(wt, "create database db") + self.execWtSql(wt, "create database db") + class TaskDropDb(StateTransitionTask): @classmethod @@ -1390,6 +1573,7 @@ class TaskDropDb(StateTransitionTask): self.execWtSql(wt, "drop database db") logger.debug("[OPS] database dropped at {}".format(time.time())) + class TaskCreateSuperTable(StateTransitionTask): @classmethod def getEndState(cls): @@ -1400,115 +1584,135 @@ class TaskCreateSuperTable(StateTransitionTask): return state.canCreateFixedSuperTable() def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): - if not wt.dbInUse(): # no DB yet, to the best of our knowledge + 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() + 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 + 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 + 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 + 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 + 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: - rTables = self.getQueryResult(wt) # wt.getDbConn().getQueryResult() + # wt.getDbConn().getQueryResult() + rTables = self.getQueryResult(wt) # print("rTables[0] = {}, type = {}".format(rTables[0], type(rTables[0]))) - for rTbName in rTables : # regular tables + 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() + 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" + # 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) + 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: - errno2 = err.errno if (err.errno > 0) else 0x80000000 + err.errno # correcting for strange error number scheme - if ( errno2 in [0x362]) : # mnode invalid table name + 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 + 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 : + tickOutput = True # Print only one time + if isSuccess: print("d", end="", flush=True) else: - print("f", end="", flush=True) + print("f", end="", flush=True) # Drop the super table itself - tblName = self._dbManager.getFixedSuperTableName() + 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 + 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 + return state.canDropFixedSuperTable() # if we can drop it, we can alter tags def _executeInternal(self, te: TaskExecutor, wt: WorkerThread): - tblName = self._dbManager.getFixedSuperTableName() + tblName = self._dbManager.getFixedSuperTableName() dice = Dice.throw(4) - if dice == 0 : + if dice == 0: sql = "alter table db.{} add tag extraTag int".format(tblName) - elif dice == 1 : + elif dice == 1: sql = "alter table db.{} drop tag extraTag".format(tblName) - elif dice == 2 : + 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) + else: # dice == 3 + sql = "alter table db.{} change tag extraTag newTag".format( + tblName) self.execWtSql(wt, sql) + class TaskAddData(StateTransitionTask): - activeTable : Set[int] = set() # Track which table is being actively worked on + # 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 + # 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 == None ): - logger.info("Recording in a file operations to be performed...") + 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 == None ): + if (cls.fAddLogDone is None): logger.info("Recording in a file operations completed...") cls.fAddLogDone = open("add_log_done.txt", "w") @@ -1519,71 +1723,84 @@ class TaskAddData(StateTransitionTask): @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 + # 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() + 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.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(), + 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 + 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.write( + "Wrote {} to {}\n".format( + nextInt, regTableName)) self.fAddLogDone.flush() os.fsync(self.fAddLogDone) - self.activeTable.discard(i) # not raising an error, unlike remove + self.activeTable.discard(i) # not raising an error, unlike remove # Deterministic random number generator class Dice(): - seeded = False # static, uninitialized + seeded = False # static, uninitialized @classmethod - def seed(cls, s): # static + def seed(cls, s): # static if (cls.seeded): - raise RuntimeError("Cannot seed the random generator more than once") + 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 + 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 ): + 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 + 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 ): + 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) @@ -1599,7 +1816,7 @@ class Dice(): # ] # def throwDice(self): -# max = len(self.tasks) - 1 +# max = len(self.tasks) - 1 # dRes = random.randint(0, max) # # logger.debug("Threw the dice in range [{},{}], and got: {}".format(0,max,dRes)) # return dRes @@ -1614,8 +1831,8 @@ class Dice(): class LoggingFilter(logging.Filter): def filter(self, record: logging.LogRecord): - if ( record.levelno >= logging.INFO ) : - return True # info or above always log + if (record.levelno >= logging.INFO): + return True # info or above always log # Commenting out below to adjust... @@ -1623,13 +1840,15 @@ class LoggingFilter(logging.Filter): # return False return True -class MyLoggingAdapter(logging.LoggerAdapter): + +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: - + +class SvcManager: + def __init__(self): print("Starting service manager") signal.signal(signal.SIGTERM, self.sigIntHandler) @@ -1641,17 +1860,18 @@ class SvcManager: def svcOutputReader(self, out: IO, queue): # print("This is the svcOutput Reader...") - for line in out : # iter(out.readline, b''): + for line in out: # iter(out.readline, b''): # print("Finished reading a line: {}".format(line)) - queue.put(line.rstrip()) # get rid of new lines - print("No more output from incoming IO") # meaning sub process must have died + queue.put(line.rstrip()) # get rid of new lines + # meaning sub process must have died + print("No more output from incoming IO") out.close() def sigIntHandler(self, signalNumber, frame): - if self.status != MainExec.STATUS_RUNNING : + if self.status != MainExec.STATUS_RUNNING: print("Ignoring repeated SIGINT...") - return # do nothing if it's already not running - self.status = MainExec.STATUS_STOPPING # immediately set our status + return # do nothing if it's already not running + self.status = MainExec.STATUS_STOPPING # immediately set our status print("Terminating program...") self.subProcess.send_signal(signal.SIGINT) @@ -1659,33 +1879,40 @@ class SvcManager: self.joinIoThread() def joinIoThread(self): - if self.ioThread : + if self.ioThread: self.ioThread.join() - self.ioThread = None + self.ioThread = None def run(self): ON_POSIX = 'posix' in sys.builtin_module_names svcCmd = ['../../build/build/bin/taosd', '-c', '../../build/test/cfg'] # svcCmd = ['vmstat', '1'] - self.subProcess = subprocess.Popen(svcCmd, stdout=subprocess.PIPE, bufsize=1, close_fds=ON_POSIX, text=True) + self.subProcess = subprocess.Popen( + svcCmd, + stdout=subprocess.PIPE, + bufsize=1, + close_fds=ON_POSIX, + text=True) q = Queue() - self.ioThread = threading.Thread(target=self.svcOutputReader, args=(self.subProcess.stdout, q)) - self.ioThread.daemon = True # thread dies with the program + self.ioThread = threading.Thread( + target=self.svcOutputReader, args=( + self.subProcess.stdout, q)) + self.ioThread.daemon = True # thread dies with the program self.ioThread.start() - # proc = subprocess.Popen(['echo', '"to stdout"'], + # proc = subprocess.Popen(['echo', '"to stdout"'], # stdout=subprocess.PIPE, # ) # stdout_value = proc.communicate()[0] # print('\tstdout: {}'.format(repr(stdout_value))) - while True : - try: - line = q.get_nowait() # getting output at fast speed + while True: + try: + line = q.get_nowait() # getting output at fast speed except Empty: # print('no output yet') - time.sleep(2.3) # wait only if there's no output - else: # got line + time.sleep(2.3) # wait only if there's no output + else: # got line print(line) # print("----end of iteration----") if self.shouldStop: @@ -1693,10 +1920,11 @@ class SvcManager: break print("end of loop") - + self.joinIoThread() print("Finished") + class ClientManager: def __init__(self): print("Starting service manager") @@ -1707,41 +1935,42 @@ class ClientManager: self.tc = None def sigIntHandler(self, signalNumber, frame): - if self.status != MainExec.STATUS_RUNNING : + if self.status != MainExec.STATUS_RUNNING: print("Ignoring repeated SIGINT...") - return # do nothing if it's already not running - self.status = MainExec.STATUS_STOPPING # immediately set our status + return # do nothing if it's already not running + self.status = MainExec.STATUS_STOPPING # immediately set our status print("Terminating program...") self.tc.requestToStop() - def _printLastNumbers(self): # to verify data durability + def _printLastNumbers(self): # to verify data durability dbManager = DbManager(resetDb=False) dbc = dbManager.getDbConn() - if dbc.query("show databases") == 0 : # no databae + if dbc.query("show databases") == 0: # no databae return - if dbc.query("show tables") == 0 : # no tables + if dbc.query("show tables") == 0: # no tables return dbc.execute("use db") - sTbName = dbManager.getFixedSuperTableName() + sTbName = dbManager.getFixedSuperTableName() # get all regular tables - dbc.query("select TBNAME from db.{}".format(sTbName)) # TODO: analyze result set later + # 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 + for rTbName in rTables: # regular tables dbc.query("select speed from db.{}".format(rTbName[0])) numbers = dbc.getQueryResult() - for row in numbers : + 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? + dbManager = None # release? def prepare(self): self._printLastNumbers() @@ -1749,21 +1978,22 @@ class ClientManager: def run(self): self._printLastNumbers() - dbManager = DbManager() # Regular function - Dice.seed(0) # initial seeding of dice + dbManager = DbManager() # Regular function + Dice.seed(0) # initial seeding of dice 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())) self.conclude() # print("TC failed (2) = {}".format(self.tc.isFailed())) - return 1 if self.tc.isFailed() else 0 # Linux return code: ref https://shapeshed.com/unix-exit-codes/ + # 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() + self.tc.getDbManager().cleanUp() class MainExec: @@ -1782,13 +2012,13 @@ class MainExec: svcManager.run() @classmethod - def runTemp(cls): # for debugging purposes + def runTemp(cls): # for debugging purposes # # Hack to exercise reading from disk, imcreasing coverage. TODO: fix # dbc = dbState.getDbConn() - # sTbName = dbState.getFixedSuperTableName() + # sTbName = dbState.getFixedSuperTableName() # dbc.execute("create database if not exists db") # if not dbState.getState().equals(StateEmpty()): - # dbc.execute("use db") + # dbc.execute("use db") # rTables = None # try: # the super table may not exist @@ -1800,7 +2030,7 @@ class MainExec: # 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]))) @@ -1809,24 +2039,26 @@ class MainExec: # ds = dbState # logger.info("Inserting into table: {}".format(rTbName[0])) # sql = "insert into db.{} values ('{}', {});".format( - # rTbName[0], + # rTbName[0], # ds.getNextTick(), ds.getNextInt()) # dbc.execute(sql) - # for rTbName in rTables : # regular tables + # 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") + # logger.info("Initial READING operation is successful") # except taos.error.ProgrammingError as err: - # logger.info("Initial WRITE/READ error: {}".format(err)) - + # logger.info("Initial WRITE/READ error: {}".format(err)) + # Sandbox testing code # dbc = dbState.getDbConn() # while True: - # rows = dbc.query("show databases") + # rows = dbc.query("show databases") # print("Rows: {}, time={}".format(rows, time.time())) - return + return + def main(): - # Super cool Python argument library: https://docs.python.org/3/library/argparse.html + # Super cool Python argument library: + # https://docs.python.org/3/library/argparse.html parser = argparse.ArgumentParser( formatter_class=argparse.RawDescriptionHelpFormatter, description=textwrap.dedent('''\ @@ -1837,22 +2069,52 @@ def main(): ''')) - 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( + '-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)') global gConfig gConfig = parser.parse_args() @@ -1860,31 +2122,40 @@ def main(): # if len(sys.argv) == 1: # parser.print_help() # sys.exit() - + # Logging Stuff global logger - _logger = logging.getLogger('CrashGen') # real logger - _logger.addFilter(LoggingFilter()) + _logger = logging.getLogger('CrashGen') # real logger + _logger.addFilter(LoggingFilter()) ch = logging.StreamHandler() _logger.addHandler(ch) - logger = MyLoggingAdapter(_logger, []) # Logging adapter, to be used as a logger + # Logging adapter, to be used as a logger + logger = MyLoggingAdapter(_logger, []) - if ( gConfig.debug ): - logger.setLevel(logging.DEBUG) # default seems to be INFO + if (gConfig.debug): + logger.setLevel(logging.DEBUG) # default seems to be INFO else: logger.setLevel(logging.INFO) - + # Run server or client - if gConfig.run_tdengine : # run server + if gConfig.run_tdengine: # run server MainExec.runService() - else : + else: return MainExec.runClient() - # logger.info("Crash_Gen execution finished") + if __name__ == "__main__": + tdDnodes.init("") + tdDnodes.setTestCluster(False) + tdDnodes.setValgrind(False) + tdDnodes.stopAll() + tdDnodes.deploy(1) + tdDnodes.start(1) + tdLog.sleep(5) + exitCode = main() # print("Exiting with code: {}".format(exitCode)) sys.exit(exitCode) diff --git a/tests/pytest/crash_gen.sh b/tests/pytest/crash_gen.sh index de80361aa3ca655e81f5e36d511a6104e1c96aa6..3c04cd619860d355528e9c40af192954a802bf0b 100755 --- a/tests/pytest/crash_gen.sh +++ b/tests/pytest/crash_gen.sh @@ -31,11 +31,23 @@ then exit -1 fi +CURR_DIR=`pwd` +IN_TDINTERNAL="community" +if [[ "$CURR_DIR" == *"$IN_TDINTERNAL"* ]]; then + TAOS_DIR=$CURR_DIR/../../.. +else + TAOS_DIR=$CURR_DIR/../.. +fi +TAOSD_DIR=`find $TAOS_DIR -name "taosd"|grep bin|head -n1` + +LIB_DIR=`echo $TAOSD_DIR|rev|cut -d '/' -f 3,4,5,6|rev`/lib +echo $LIB_DIR + # First we need to set up a path for Python to find our own TAOS modules, so that "import" can work. export PYTHONPATH=$(pwd)/../../src/connector/python/linux/python3 # Then let us set up the library path so that our compiled SO file can be loaded by Python -export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$(pwd)/../../build/build/lib +export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$LIB_DIR # Now we are all let, and let's see if we can find a crash. Note we pass all params python3 ./crash_gen.py $@