random-test-multi-threading-3.py 7.6 KB
Newer Older
S
Shuduo Sang 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181
###################################################################
#           Copyright (c) 2016 by TAOS Technologies, Inc.
#                     All rights reserved.
#
#  This file is proprietary and confidential to TAOS Technologies.
#  No part of this file may be reproduced, stored, transmitted,
#  disclosed or used in any form or by any means other than as
#  expressly provided by the written permission from Jianhui Tao
#
###################################################################

# -*- coding: utf-8 -*-

import sys
import random
import threading

from util.log import *
from util.cases import *
from util.sql import *
from util.dnodes import *

last_tb = ""
last_stb = ""
written = 0


class Test (threading.Thread):
    def __init__(self, threadId, name):
        threading.Thread.__init__(self)
        self.threadId = threadId
        self.name = name

        self.threadLock = threading.Lock()

    def create_table(self):
        tdLog.info("create_table")
        global last_tb
        global written

        current_tb = "tb%d" % int(round(time.time() * 1000))

        if (current_tb == last_tb):
            return
        else:
            tdLog.info("will create table %s" % current_tb)

            try:
                tdSql.execute(
                    'create table %s (ts timestamp, speed int)' %
                    current_tb)
                last_tb = current_tb
                written = 0
            except Exception as e:
                tdLog.info(repr(e))

    def insert_data(self):
        tdLog.info("insert_data")
        global last_tb
        global written

        if (last_tb == ""):
            tdLog.info("no table, create first")
            self.create_table()

        tdLog.info("will insert data to table")
        for i in range(0, 10):
            insertRows = 1000
            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)' %
                    (last_tb, j, j))
                written = written + 1

    def query_data(self):
        tdLog.info("query_data")
        global last_tb
        global written

        if (written > 0):
            tdLog.info("query data from table")
            tdSql.query("select * from %s" % last_tb)
            tdSql.checkRows(written)

    def create_stable(self):
        tdLog.info("create_stable")
        global last_tb
        global last_stb
        global written

        current_stb = "stb%d" % int(round(time.time() * 1000))

        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)
            written = 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):
        tdLog.info("restart_database")
        global last_tb
        global written

        tdDnodes.stop(1)
        tdDnodes.start(1)

    def force_restart_database(self):
        tdLog.info("force_restart_database")
        global last_tb
        global written

        tdDnodes.forcestop(1)
        tdDnodes.start(1)

    def drop_table(self):
        tdLog.info("drop_table")
        global last_tb
        global written

        for i in range(0, 10):
            if (last_tb != ""):
                tdLog.info("drop last_tb %s" % last_tb)
                tdSql.execute("drop table %s" % last_tb)
                last_tb = ""
                written = 0

    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):
        tdLog.info("reset_query_cache")
        global last_tb
        global written

        tdLog.info("reset query cache")
        tdSql.execute("reset query cache")
        tdLog.sleep(1)

    def reset_database(self):
        tdLog.info("reset_database")
        global last_tb
        global last_stb
        global written

        tdDnodes.forcestop(1)
        tdDnodes.deploy(1)
S
Shuduo Sang 已提交
182 183
        tdDnodes.start(1)
        tdSql.prepare()
S
Shuduo Sang 已提交
184 185 186 187 188 189 190
        last_tb = ""
        last_stb = ""
        written = 0

    def delete_datafiles(self):
        tdLog.info("delete_data_files")
        global last_tb
S
Shuduo Sang 已提交
191
        global last_stb
S
Shuduo Sang 已提交
192 193 194 195 196 197 198 199 200
        global written

        dnodesDir = tdDnodes.getDnodesRootDir()
        dataDir = dnodesDir + '/dnode1/*'
        deleteCmd = 'rm -rf %s' % dataDir
        os.system(deleteCmd)

        tdDnodes.start(1)
        tdSql.prepare()
S
Shuduo Sang 已提交
201 202 203
        last_tb = ""
        last_stb = ""
        written = 0
S
Shuduo Sang 已提交
204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233

    def run(self):
        dataOp = {
            1: self.insert_data,
            2: self.query_data,
            3: self.query_data_from_stable,
        }

        dbOp = {
            1: self.create_table,
            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,
        }

        queryOp = {
            1: self.query_data,
            2: self.query_data_from_stable,
        }

        if (self.threadId == 1):
            while True:
                self.threadLock.acquire()
                tdLog.notice("first thread")
                randDataOp = random.randint(1, 3)
234
                dataOp.get(randDataOp, lambda: "ERROR")()
S
Shuduo Sang 已提交
235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280
                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()
        elif (self.threadId == 3):
            while True:
                tdLog.notice("third thread")
                self.threadLock.acquire()
                randQueryOp = random.randint(1, 9)
                queryOp.get(randQueryOp, lambda: "ERROR")()
                self.threadLock.release()


class TDTestCase:
    def init(self, conn, logSql):
        tdLog.debug("start to execute %s" % __file__)
        tdSql.init(conn.cursor(), logSql)

    def run(self):
        tdSql.prepare()

        test1 = Test(1, "data operation")
        test2 = Test(2, "db operation")
        test2 = Test(3, "query operation")

        test1.start()
        test2.start()
        test3.start()
        test1.join()
        test2.join()
        test3.join()

        tdLog.info("end of test")

    def stop(self):
        tdSql.close()
        tdLog.success("%s successfully executed" % __file__)


tdCases.addWindows(__file__, TDTestCase())
tdCases.addLinux(__file__, TDTestCase())