Merge pull request #2125 from taosdata/feature/sangshuduo/multi-thread-random-test

add multi-threading random test

tests/pytest/random-test/random-test-multi-threading.py

@@ -20,32 +20,30 @@ from util.cases import *
 from util.sql import *
 from util.dnodes import *
 
-current_tb = ""
 last_tb = ""
+last_stb = ""
 written = 0
 
 
 class Test (threading.Thread):
-    def __init__(self, threadId, name, sleepTime):
+    def __init__(self, threadId, name):
         threading.Thread.__init__(self)
         self.threadId = threadId
         self.name = name
-        self.sleepTime = sleepTime
 
         self.threadLock = threading.Lock()
 
     def create_table(self):
-        global current_tb
+        tdLog.info("create_table")
         global last_tb
         global written
 
-        tdLog.info("create a table")
         current_tb = "tb%d" % int(round(time.time() * 1000))
-        tdLog.info("current table %s" % current_tb)
 
         if (current_tb == last_tb):
             return
         else:
+            tdLog.info("will create table %s" % current_tb)
             tdSql.execute(
                 'create table %s (ts timestamp, speed int)' %
                 current_tb)
@@ -53,30 +51,27 @@ class Test (threading.Thread):
             written = 0
 
     def insert_data(self):
-        global current_tb
+        tdLog.info("insert_data")
         global last_tb
         global written
 
-        tdLog.info("will insert data to table")
-        if (current_tb == ""):
+        if (last_tb == ""):
             tdLog.info("no table, create first")
             self.create_table()
 
-        tdLog.info("insert data to table")
+        tdLog.info("will insert data to table")
         for i in range(0, 10):
-            self.threadLock.acquire()
             insertRows = 1000
-            tdLog.info("insert %d rows to %s" % (insertRows, current_tb))
+            tdLog.info("insert %d rows to %s" % (insertRows, last_tb))
 
             for j in range(0, insertRows):
                 ret = tdSql.execute(
                     'insert into %s values (now + %dm, %d)' %
-                    (current_tb, j, j))
+                    (last_tb, j, j))
                 written = written + 1
-            self.threadLock.release()
 
     def query_data(self):
-        global current_tb
+        tdLog.info("query_data")
         global last_tb
         global written
 
@@ -86,53 +81,90 @@ class Test (threading.Thread):
             tdSql.checkRows(written)
 
     def create_stable(self):
-        global current_tb
+        tdLog.info("create_stable")
         global last_tb
+        global last_stb
         global written
+        current_stb = "stb%d" % int(round(time.time() * 1000))
 
-        tdLog.info("create a super table")
+        if (current_stb == last_stb):
+            return
+        else:
+            tdLog.info("will create stable %s" % current_stb)
+            tdSql.execute(
+                'create table %s(ts timestamp, c1 int, c2 nchar(10)) tags (t1 int, t2 nchar(10))' %
+                current_stb)
+            last_stb = current_stb
+
+            current_tb = "tb%d" % int(round(time.time() * 1000))
+            tdSql.execute(
+                "create table %s using %s tags (1, '表1')" %
+                (current_tb, last_stb))
+            last_tb = current_tb
+            tdSql.execute(
+                "insert into %s values (now, 27, '我是nchar字符串')" %
+                last_tb)
+            self.written = self.written + 1
+
+
+    def drop_stable(self):
+        tdLog.info("drop_stable")
+        global last_stb
+
+        if (last_stb == ""):
+            tdLog.info("no super table")
+            return
+        else:
+            tdLog.info("will drop last super table")
+            tdSql.execute('drop table %s' % last_stb)
+            last_stb = ""
 
     def restart_database(self):
-        global current_tb
+        tdLog.info("restart_database")
         global last_tb
         global written
 
-        tdLog.info("restart databae")
         tdDnodes.stop(1)
         tdDnodes.start(1)
         tdLog.sleep(5)
 
-    def force_restart(self):
-        global current_tb
+    def force_restart_database(self):
+        tdLog.info("force_restart_database")
         global last_tb
         global written
 
-        tdLog.info("force restart database")
         tdDnodes.forcestop(1)
         tdDnodes.start(1)
         tdLog.sleep(5)
 
     def drop_table(self):
-        global current_tb
+        tdLog.info("drop_table")
         global last_tb
         global written
 
         for i in range(0, 10):
-            self.threadLock.acquire()
-
-            tdLog.info("current_tb %s" % current_tb)
-
-            if (current_tb != ""):
-                tdLog.info("drop current tb %s" % current_tb)
-                tdSql.execute("drop table %s" % current_tb)
-                current_tb = ""
+            if (last_tb != ""):
+                tdLog.info("drop last_tb %s" % last_tb)
+                tdSql.execute("drop table %s" % last_tb)
                 last_tb = ""
                 written = 0
             tdLog.sleep(self.sleepTime)
-            self.threadLock.release()
+
+
+    def query_data_from_stable(self):
+        tdLog.info("query_data_from_stable")
+        global last_stb
+
+        if (last_stb == ""):
+            tdLog.info("no super table")
+            return
+        else:
+            tdLog.info("will query data from super table")
+            tdSql.execute('select * from %s' % last_stb)
+
 
     def reset_query_cache(self):
-        global current_tb
+        tdLog.info("reset_query_cache")
         global last_tb
         global written
 
@@ -141,51 +173,66 @@ class Test (threading.Thread):
         tdLog.sleep(1)
 
     def reset_database(self):
-        global current_tb
+        tdLog.info("reset_database")
         global last_tb
         global written
 
-        tdLog.info("reset database")
         tdDnodes.forcestop(1)
         tdDnodes.deploy(1)
-        current_tb = ""
         last_tb = ""
         written = 0
         tdDnodes.start(1)
         tdSql.prepare()
 
     def delete_datafiles(self):
-        global current_tb
+        tdLog.info("delete_data_files")
         global last_tb
         global written
 
-        tdLog.info("delete data files")
         dnodesDir = tdDnodes.getDnodesRootDir()
         dataDir = dnodesDir + '/dnode1/*'
         deleteCmd = 'rm -rf %s' % dataDir
         os.system(deleteCmd)
 
-        current_tb = ""
         last_tb = ""
         written = 0
         tdDnodes.start(1)
         tdSql.prepare()
 
     def run(self):
-        switch = {
+        dataOp = {
+            1: self.insert_data,
+            2: self.query_data,
+            3: self.query_data_from_stable,
+        }
+
+        dbOp = {
             1: self.create_table,
-            2: self.insert_data,
-            3: self.query_data,
-            4: self.create_stable,
-            5: self.restart_database,
-            6: self.force_restart,
-            7: self.drop_table,
-            8: self.reset_query_cache,
-            9: self.reset_database,
-            10: self.delete_datafiles,
+            2: self.create_stable,
+            3: self.restart_database,
+            4: self.force_restart_database,
+            5: self.drop_table,
+            6: self.reset_query_cache,
+            7: self.reset_database,
+            8: self.delete_datafiles,
+            9: self.drop_stable,
         }
 
-        switch.get(self.threadId, lambda: "ERROR")()
+        if (self.threadId == 1):
+            while True:
+                self.threadLock.acquire()
+                tdLog.notice("first thread")
+                randDataOp = random.randint(1, 3)
+                dataOp.get(randDataOp , lambda: "ERROR")()
+                self.threadLock.release()
+
+        elif (self.threadId == 2):
+            while True:
+                tdLog.notice("second thread")
+                self.threadLock.acquire()
+                randDbOp = random.randint(1, 9)
+                dbOp.get(randDbOp, lambda: "ERROR")()
+                self.threadLock.release()
 
 
 class TDTestCase:
@@ -196,8 +243,8 @@ class TDTestCase:
     def run(self):
         tdSql.prepare()
 
-        test1 = Test(2, "insert_data", 1)
-        test2 = Test(7, "drop_table", 2)
+        test1 = Test(1, "data operation")
+        test2 = Test(2, "db operation")
 
         test1.start()
         test2.start()

tests/pytest/random-test/random-test.py

@@ -21,101 +21,152 @@ from util.dnodes import *
 
 class Test:
     def __init__(self):
-        self.current_tb = ""
         self.last_tb = ""
+        self.last_stb = ""
         self.written = 0
 
     def create_table(self):
-        tdLog.info("create a table")
-        self.current_tb = "tb%d" % int(round(time.time() * 1000))
-        tdLog.info("current table %s" % self.current_tb)
+        tdLog.info("create_table")
+        current_tb = "tb%d" % int(round(time.time() * 1000))
 
-        if (self.current_tb == self.last_tb):
+        if (current_tb == self.last_tb):
             return
         else:
+            tdLog.info("will create table %s" % current_tb)
             tdSql.execute(
-                'create table %s (ts timestamp, speed int)' %
-                self.current_tb)
-            self.last_tb = self.current_tb
+                'create table %s (ts timestamp, c1 int, c2 nchar(10))' %
+                current_tb)
+            self.last_tb = current_tb
             self.written = 0
 
     def insert_data(self):
-        tdLog.info("will insert data to table")
-        if (self.current_tb == ""):
+        tdLog.info("insert_data")
+        if (self.last_tb == ""):
             tdLog.info("no table, create first")
             self.create_table()
 
-        tdLog.info("insert data to table")
+        tdLog.info("will insert data to table")
         insertRows = 10
         tdLog.info("insert %d rows to %s" % (insertRows, self.last_tb))
         for i in range(0, insertRows):
             ret = tdSql.execute(
-                'insert into %s values (now + %dm, %d)' %
-                (self.last_tb, i, i))
+                'insert into %s values (now + %dm, %d, "%s")' %
+                (self.last_tb, i, i, "->" + str(i)))
             self.written = self.written + 1
 
         tdLog.info("insert earlier data")
-        tdSql.execute('insert into %s values (now - 5m , 10)' % self.last_tb)
+        tdSql.execute(
+            'insert into %s values (now - 5m , 10, " - 5m")' %
+            self.last_tb)
         self.written = self.written + 1
-        tdSql.execute('insert into %s values (now - 6m , 10)' % self.last_tb)
+        tdSql.execute(
+            'insert into %s values (now - 6m , 10, " - 6m")' %
+            self.last_tb)
         self.written = self.written + 1
-        tdSql.execute('insert into %s values (now - 7m , 10)' % self.last_tb)
+        tdSql.execute(
+            'insert into %s values (now - 7m , 10, " - 7m")' %
+            self.last_tb)
         self.written = self.written + 1
-        tdSql.execute('insert into %s values (now - 8m , 10)' % self.last_tb)
+        tdSql.execute(
+            'insert into %s values (now - 8m , 10, " - 8m")' %
+            self.last_tb)
         self.written = self.written + 1
 
     def query_data(self):
+        tdLog.info("query_data")
         if (self.written > 0):
             tdLog.info("query data from table")
             tdSql.query("select * from %s" % self.last_tb)
             tdSql.checkRows(self.written)
 
     def create_stable(self):
-        tdLog.info("create a super table")
+        tdLog.info("create_stable")
+        current_stb = "stb%d" % int(round(time.time() * 1000))
+
+        if (current_stb == self.last_stb):
+            return
+        else:
+            tdLog.info("will create stable %s" % current_stb)
+            tdSql.execute(
+                'create table %s(ts timestamp, c1 int, c2 nchar(10)) tags (t1 int, t2 nchar(10))' %
+                current_stb)
+            self.last_stb = current_stb
+
+            current_tb = "tb%d" % int(round(time.time() * 1000))
+            tdSql.execute(
+                "create table %s using %s tags (1, '表1')" %
+                (current_tb, self.last_stb))
+            self.last_tb = current_tb
+            tdSql.execute(
+                "insert into %s values (now, 27, '我是nchar字符串')" %
+                self.last_tb)
+            self.written = self.written + 1
+
+    def drop_stable(self):
+        tdLog.info("drop_stable")
+        if (self.last_stb == ""):
+            tdLog.info("no super table")
+            return
+        else:
+            tdLog.info("will drop last super table")
+            tdSql.execute('drop table %s' % self.last_stb)
+            self.last_stb = ""
+
+
+    def query_data_from_stable(self):
+        tdLog.info("query_data_from_stable")
+        if (self.last_stb == ""):
+            tdLog.info("no super table")
+            return
+        else:
+            tdLog.info("will query data from super table")
+            tdSql.execute('select * from %s' % self.last_stb)
+
 
     def restart_database(self):
-        tdLog.info("restart databae")
+        tdLog.info("restart_databae")
         tdDnodes.stop(1)
         tdDnodes.start(1)
         tdLog.sleep(5)
 
-    def force_restart(self):
-        tdLog.info("force restart database")
+
+    def force_restart_database(self):
+        tdLog.info("force_restart_database")
         tdDnodes.forcestop(1)
         tdDnodes.start(1)
         tdLog.sleep(5)
+        tdSql.prepare()
 
     def drop_table(self):
-        if (self.current_tb != ""):
-            tdLog.info("drop current tb %s" % self.current_tb)
-            tdSql.execute("drop table %s" % self.current_tb)
-            self.current_tb = ""
+        tdLog.info("drop_table")
+        if (self.last_tb != ""):
+            tdLog.info("drop last tb %s" % self.last_tb)
+            tdSql.execute("drop table %s" % self.last_tb)
             self.last_tb = ""
             self.written = 0
 
     def reset_query_cache(self):
-        tdLog.info("reset query cache")
+        tdLog.info("reset_query_cache")
         tdSql.execute("reset query cache")
         tdLog.sleep(1)
 
     def reset_database(self):
-        tdLog.info("reset database")
+        tdLog.info("reset_database")
         tdDnodes.forcestop(1)
         tdDnodes.deploy(1)
-        self.current_tb = ""
         self.last_tb = ""
         self.written = 0
         tdDnodes.start(1)
+        tdLog.sleep(5)
         tdSql.prepare()
 
     def delete_datafiles(self):
-        tdLog.info("delete data files")
+        tdLog.info("delete_datafiles")
         dnodesDir = tdDnodes.getDnodesRootDir()
         dataDir = dnodesDir + '/dnode1/*'
         deleteCmd = 'rm -rf %s' % dataDir
         os.system(deleteCmd)
 
-        self.current_tb = ""
         self.last_tb = ""
         self.written = 0
         tdDnodes.start(1)
@@ -138,15 +189,17 @@ class TDTestCase:
             3: test.query_data,
             4: test.create_stable,
             5: test.restart_database,
-            6: test.force_restart,
+            6: test.force_restart_database,
             7: test.drop_table,
             8: test.reset_query_cache,
             9: test.reset_database,
             10: test.delete_datafiles,
+            11: test.query_data_from_stable,
+            12: test.drop_stable,
         }
 
-        for x in range(1, 100):
-            r = random.randint(1, 10)
+        for x in range(1, 1000):
+            r = random.randint(1, 12)
             tdLog.notice("iteration %d run func %d" % (x, r))
             switch.get(r, lambda: "ERROR")()