concurrent_inquiry.py 26.4 KB
Newer Older
L
liuyq-617 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14
###################################################################
#           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 threading
import taos
L
liuyq-617 已提交
15
import sys
L
liuyq-617 已提交
16 17 18
import json
import time
import random
19
import requests
L
liuyq-617 已提交
20
import argparse
21 22
import datetime
import string
23 24 25 26 27 28 29 30 31 32 33 34
from requests.auth import HTTPBasicAuth
func_list=['avg','count','twa','sum','stddev','leastsquares','min',
'max','first','last','top','bottom','percentile','apercentile',
'last_row','diff','spread']
condition_list=[
    "where _c0 > now -10d ",
    'interval(10s)',
    'limit 10',
    'group by',
    'order by',
    'fill(null)'
    
L
liuyq-617 已提交
35
]
36
where_list = ['_c0>now-10d',' <50','like',' is null']
L
liuyq-617 已提交
37
class ConcurrentInquiry:
L
liuyq-617 已提交
38 39 40 41 42
    # def __init__(self,ts=1500000001000,host='127.0.0.1',user='root',password='taosdata',dbname='test',
    #             stb_prefix='st',subtb_prefix='t',n_Therads=10,r_Therads=10,probabilities=0.05,loop=5,
    #             stableNum = 2,subtableNum = 1000,insertRows = 100):  
    def __init__(self,ts,host,user,password,dbname,
                stb_prefix,subtb_prefix,n_Therads,r_Therads,probabilities,loop,
L
liuyq-617 已提交
43
                stableNum ,subtableNum ,insertRows ,mix_table, replay):  
44 45
        self.n_numOfTherads = n_Therads
        self.r_numOfTherads = r_Therads
L
liuyq-617 已提交
46 47 48 49 50 51 52
        self.ts=ts
        self.host = host
        self.user = user
        self.password = password
        self.dbname=dbname
        self.stb_prefix = stb_prefix
        self.subtb_prefix = subtb_prefix
53 54 55 56 57 58
        self.stb_list=[]
        self.subtb_list=[]
        self.stb_stru_list=[]
        self.subtb_stru_list=[]
        self.stb_tag_list=[]
        self.subtb_tag_list=[]
59 60
        self.probabilities = [1-probabilities,probabilities]
        self.ifjoin = [1,0]
L
liuyq-617 已提交
61 62 63 64
        self.loop = loop
        self.stableNum = stableNum
        self.subtableNum = subtableNum
        self.insertRows = insertRows
L
liuyq-617 已提交
65
        self.mix_table = mix_table
66 67
        self.max_ts = datetime.datetime.now()
        self.min_ts = datetime.datetime.now() - datetime.timedelta(days=5)
L
liuyq-617 已提交
68
        self.replay = replay
L
liuyq-617 已提交
69 70
    def SetThreadsNum(self,num):
        self.numOfTherads=num
71

72 73 74 75 76 77
    def ret_fcol(self,cl,sql):                     #返回结果的第一列
        cl.execute(sql)
        fcol_list=[]
        for data in cl:
            fcol_list.append(data[0])
        return fcol_list
78 79

    def r_stb_list(self,cl):                    #返回超级表列表
80 81
        sql='show '+self.dbname+'.stables'
        self.stb_list=self.ret_fcol(cl,sql)
82 83

    def r_subtb_list(self,cl,stablename):       #每个超级表返回2个子表
84 85
        sql='select tbname from '+self.dbname+'.'+stablename+' limit 2;'
        self.subtb_list+=self.ret_fcol(cl,sql)
86 87

    def cal_struct(self,cl,tbname):             #查看表结构
88 89 90 91 92 93 94 95 96 97
        tb=[]
        tag=[]
        sql='describe '+self.dbname+'.'+tbname+';'
        cl.execute(sql)
        for data in cl:
            if data[3]:
                tag.append(data[0])
            else:
                tb.append(data[0])
        return tb,tag
98 99

    def r_stb_stru(self,cl):                    #获取所有超级表的表结构
100 101 102 103
        for i in self.stb_list:
            tb,tag=self.cal_struct(cl,i)
            self.stb_stru_list.append(tb)
            self.stb_tag_list.append(tag)
104 105

    def r_subtb_stru(self,cl):                  #返回所有子表的表结构
106 107 108 109
        for i in self.subtb_list:
            tb,tag=self.cal_struct(cl,i)
            self.subtb_stru_list.append(tb)
            self.subtb_tag_list.append(tag)
110

111 112 113 114 115 116 117 118
    def get_timespan(self,cl):                  #获取时间跨度(仅第一个超级表)
        sql = 'select first(_c0),last(_c0) from ' + self.dbname + '.' + self.stb_list[0] + ';'
        print(sql)
        cl.execute(sql)
        for data in cl:
            self.max_ts = data[1]
            self.min_ts = data[0]

119
    def get_full(self):                         #获取所有的表、表结构
L
liuyq-617 已提交
120 121 122
        host = self.host
        user = self.user
        password = self.password
123 124 125 126 127 128 129 130 131 132 133
        conn = taos.connect(
            host,
            user,
            password,
            )
        cl = conn.cursor()
        self.r_stb_list(cl)
        for i in self.stb_list:
            self.r_subtb_list(cl,i)
        self.r_stb_stru(cl)
        self.r_subtb_stru(cl)
134
        self.get_timespan(cl)
135
        cl.close()
136 137 138
        conn.close()  
        
    #query condition
139
    def con_where(self,tlist,col_list,tag_list):                               
140 141 142 143
        l=[]
        for i in range(random.randint(0,len(tlist))):
            c = random.choice(where_list)
            if c == '_c0>now-10d':
144 145 146 147 148 149 150 151 152 153 154
                rdate = self.min_ts + (self.max_ts - self.min_ts)/10 * random.randint(-11,11)
                conlist = ' _c0 ' + random.choice(['<','>','>=','<=','<>']) + "'" + str(rdate) + "'"
                if self.random_pick():
                    l.append(conlist)
                else: l.append(c)
            elif '<50' in c:
                conlist = ' ' + random.choice(tlist) + random.choice(['<','>','>=','<=','<>']) + str(random.randrange(-100,100))
                l.append(conlist) 
            elif 'is null' in c:
                conlist = ' ' + random.choice(tlist) + random.choice([' is null',' is not null'])
                l.append(conlist) 
155
            else:
156 157 158
                s_all = string.ascii_letters
                conlist = ' ' + random.choice(tlist) + " like \'%" + random.choice(s_all) + "%\' " 
                l.append(conlist)
159
        return 'where '+random.choice([' and ',' or ']).join(l)
160

161
    def con_interval(self,tlist,col_list,tag_list): 
L
liuyq-617 已提交
162
        interval = 'interval(' + str(random.randint(0,20)) + random.choice(['a','s','d','w','n','y'])  + ')'          
163
        return interval
164

165 166 167
    def con_limit(self,tlist,col_list,tag_list):
        rand1 = str(random.randint(0,1000))
        rand2 = str(random.randint(0,1000))
168 169 170
        return random.choice(['limit ' + rand1,'limit ' + rand1 + ' offset '+rand2,
        ' slimit ' + rand1,' slimit ' + rand1 + ' offset ' + rand2,'limit '+rand1 + ' slimit '+ rand2,
        'limit '+ rand1 + ' offset' + rand2 + ' slimit '+ rand1 + ' soffset ' + rand2 ])
171
    
172
    def con_fill(self,tlist,col_list,tag_list):
173
        return random.choice(['fill(null)','fill(prev)','fill(none)','fill(LINEAR)'])
174
    
175 176 177
    def con_group(self,tlist,col_list,tag_list):
        rand_tag = random.randint(0,5)
        rand_col = random.randint(0,1)
L
liuyq-617 已提交
178
        return 'group by '+','.join(random.sample(col_list,rand_col) + random.sample(tag_list,rand_tag))
179
    
180
    def con_order(self,tlist,col_list,tag_list):
181
        return 'order by '+random.choice(tlist)
182 183 184
    
    def gen_query_sql(self):                        #生成查询语句
        tbi=random.randint(0,len(self.subtb_list)+len(self.stb_list))  #随机决定查询哪张表
185 186 187 188 189 190 191 192 193 194 195 196 197 198
        tbname=''
        col_list=[]
        tag_list=[]
        is_stb=0
        if tbi>len(self.stb_list) :
            tbi=tbi-len(self.stb_list)
            tbname=self.subtb_list[tbi-1]
            col_list=self.subtb_stru_list[tbi-1]
            tag_list=self.subtb_tag_list[tbi-1]
        else:
            tbname=self.stb_list[tbi-1]
            col_list=self.stb_stru_list[tbi-1]
            tag_list=self.stb_tag_list[tbi-1]
            is_stb=1
199
        tlist=col_list+tag_list+['abc']            #增加不存在的域'abc',是否会引起新bug
200 201 202 203 204 205 206 207 208 209
        con_rand=random.randint(0,len(condition_list))
        func_rand=random.randint(0,len(func_list))
        col_rand=random.randint(0,len(col_list))
        tag_rand=random.randint(0,len(tag_list))
        t_rand=random.randint(0,len(tlist))
        sql='select '                                           #select 
        random.shuffle(col_list)
        random.shuffle(func_list)
        sel_col_list=[]
        col_rand=random.randint(0,len(col_list))
L
liuyq-617 已提交
210
        loop = 0
211
        for i,j in zip(col_list[0:col_rand],func_list):         #决定每个被查询col的函数
L
liuyq-617 已提交
212 213
            alias = ' as '+ 'taos%d ' % loop
            loop += 1
214
            pick_func = ''
215
            if j == 'leastsquares':
216
                pick_func=j+'('+i+',1,1)'
217
            elif j == 'top' or j == 'bottom' or j == 'percentile' or j == 'apercentile':
218
                pick_func=j+'('+i+',1)'
219
            else:
220 221 222 223
                pick_func=j+'('+i+')'
            if bool(random.getrandbits(1)):
                pick_func+=alias
            sel_col_list.append(pick_func)
224 225 226 227
        if col_rand == 0:
            sql = sql + '*'   
        else: 
            sql=sql+','.join(sel_col_list)         #select col & func
L
liuyq-617 已提交
228 229 230 231 232 233
        if self.mix_table == 0:
            sql = sql + ' from '+random.choice(self.stb_list+self.subtb_list)+' '         
        elif self.mix_table == 1:
            sql = sql + ' from '+random.choice(self.subtb_list)+' '
        else:
            sql = sql + ' from '+random.choice(self.stb_list)+' ' 
234 235 236 237
        con_func=[self.con_where,self.con_interval,self.con_limit,self.con_group,self.con_order,self.con_fill]
        sel_con=random.sample(con_func,random.randint(0,len(con_func)))
        sel_con_list=[]
        for i in sel_con:
238
            sel_con_list.append(i(tlist,col_list,tag_list))                                  #获取对应的条件函数
239
        sql+=' '.join(sel_con_list)                                       # condition
L
liuyq-617 已提交
240
        #print(sql)
241
        return sql
242

243
    def gen_query_join(self):                        #生成join查询语句
L
liuyq-617 已提交
244 245 246 247 248
        tbname   = []
        col_list = []
        tag_list = []
        col_intersection = []
        tag_intersection = []
249
        subtable = None
L
liuyq-617 已提交
250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266
        if self.mix_table == 0:
            if bool(random.getrandbits(1)):
                subtable = True
                tbname = random.sample(self.subtb_list,2)
                for i in tbname:
                    col_list.append(self.subtb_stru_list[self.subtb_list.index(i)])
                    tag_list.append(self.subtb_stru_list[self.subtb_list.index(i)])
                col_intersection = list(set(col_list[0]).intersection(set(col_list[1])))
                tag_intersection = list(set(tag_list[0]).intersection(set(tag_list[1])))
            else:
                tbname = random.sample(self.stb_list,2)
                for i in tbname:
                    col_list.append(self.stb_stru_list[self.stb_list.index(i)])
                    tag_list.append(self.stb_stru_list[self.stb_list.index(i)])
                col_intersection = list(set(col_list[0]).intersection(set(col_list[1])))
                tag_intersection = list(set(tag_list[0]).intersection(set(tag_list[1])))
        elif self.mix_table == 1:
267
            subtable = True
L
liuyq-617 已提交
268 269 270 271 272 273
            tbname = random.sample(self.subtb_list,2)
            for i in tbname:
                col_list.append(self.subtb_stru_list[self.subtb_list.index(i)])
                tag_list.append(self.subtb_stru_list[self.subtb_list.index(i)])
            col_intersection = list(set(col_list[0]).intersection(set(col_list[1])))
            tag_intersection = list(set(tag_list[0]).intersection(set(tag_list[1])))
274
        else:
L
liuyq-617 已提交
275 276 277 278 279 280
            tbname = random.sample(self.stb_list,2)
            for i in tbname:
                col_list.append(self.stb_stru_list[self.stb_list.index(i)])
                tag_list.append(self.stb_stru_list[self.stb_list.index(i)])
            col_intersection = list(set(col_list[0]).intersection(set(col_list[1])))
            tag_intersection = list(set(tag_list[0]).intersection(set(tag_list[1])))
281 282 283 284
        con_rand=random.randint(0,len(condition_list))
        col_rand=random.randint(0,len(col_list))
        tag_rand=random.randint(0,len(tag_list))
        sql='select '                                           #select 
L
liuyq-617 已提交
285 286
        
        sel_col_tag=[]
287
        col_rand=random.randint(0,len(col_list))
L
liuyq-617 已提交
288 289 290 291 292
        if bool(random.getrandbits(1)):
            sql += '*'
        else:
            sel_col_tag.append('t1.' + str(random.choice(col_list[0] + tag_list[0])))
            sel_col_tag.append('t2.' + str(random.choice(col_list[1] + tag_list[1])))
293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312
            sel_col_list = []
            random.shuffle(func_list)
            if self.random_pick():
                loop = 0
                for i,j in zip(sel_col_tag,func_list):         #决定每个被查询col的函数
                    alias = ' as '+ 'taos%d ' % loop
                    loop += 1
                    pick_func = ''
                    if j == 'leastsquares':
                        pick_func=j+'('+i+',1,1)'
                    elif j == 'top' or j == 'bottom' or j == 'percentile' or j == 'apercentile':
                        pick_func=j+'('+i+',1)'
                    else:
                        pick_func=j+'('+i+')'
                    if bool(random.getrandbits(1)):
                        pick_func+=alias
                    sel_col_list.append(pick_func)
                sql += ','.join(sel_col_list)
            else:
                sql += ','.join(sel_col_tag)
L
liuyq-617 已提交
313

314 315
        sql = sql + ' from '+ str(tbname[0]) +' t1,' + str(tbname[1]) + ' t2 '                        #select col & func
        join_section = None
L
liuyq-617 已提交
316
        temp = None
317
        if subtable:
L
liuyq-617 已提交
318 319 320
            temp = random.choices(col_intersection)
            join_section = temp.pop()
            sql += 'where t1._c0 = t2._c0 and ' + 't1.' + str(join_section) + '=t2.' + str(join_section)
321
        else:
L
liuyq-617 已提交
322 323 324
            temp = random.choices(col_intersection+tag_intersection)
            join_section = temp.pop()
            sql += 'where t1._c0 = t2._c0 and ' + 't1.' + str(join_section) + '=t2.' + str(join_section)
325 326 327 328 329 330 331 332 333 334
        return sql

    def random_pick(self): 
        x = random.uniform(0,1) 
        cumulative_probability = 0.0 
        for item, item_probability in zip(self.ifjoin, self.probabilities): 
            cumulative_probability += item_probability 
            if x < cumulative_probability:break 
        return item
        
L
liuyq-617 已提交
335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352
    def gen_data(self):
        stableNum = self.stableNum
        subtableNum = self.subtableNum
        insertRows = self.insertRows
        t0 = self.ts
        host = self.host
        user = self.user
        password = self.password
        conn = taos.connect(
            host,
            user,
            password,
            )
        cl = conn.cursor()
        cl.execute("drop database if  exists %s;" %self.dbname)
        cl.execute("create database if not exists %s;" %self.dbname)
        cl.execute("use %s" % self.dbname)
        for k in range(stableNum):
L
liuyq-617 已提交
353 354
            sql="create table %s (ts timestamp, c1 int, c2 float, c3 bigint, c4 smallint, c5 tinyint, c6 double, c7 bool,c8 binary(20),c9 nchar(20),c11 int unsigned,c12 smallint unsigned,c13 tinyint unsigned,c14 bigint unsigned) \
            tags(t1 int, t2 float, t3 bigint, t4 smallint, t5 tinyint, t6 double, t7 bool,t8 binary(20),t9 nchar(20), t11 int unsigned , t12 smallint unsigned , t13 tinyint unsigned , t14 bigint unsigned)" % (self.stb_prefix+str(k))
L
liuyq-617 已提交
355 356
            cl.execute(sql)
            for j in range(subtableNum):
L
liuyq-617 已提交
357 358 359 360 361 362
                if j % 100 == 0:
                    sql = "create table %s using %s tags(NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL)" % \
                            (self.subtb_prefix+str(k)+'_'+str(j),self.stb_prefix+str(k))
                else:
                    sql = "create table %s using %s tags(%d,%d,%d,%d,%d,%d,%d,'%s','%s',%d,%d,%d,%d)" % \
                            (self.subtb_prefix+str(k)+'_'+str(j),self.stb_prefix+str(k),j,j/2.0,j%41,j%51,j%53,j*1.0,j%2,'taos'+str(j),'涛思'+str(j), j%43, j%23 , j%17 , j%3167)
L
liuyq-617 已提交
363 364 365
                print(sql)
                cl.execute(sql)
                for i in range(insertRows):
L
liuyq-617 已提交
366 367 368 369 370 371 372 373
                    if i % 100 == 0 :
                        ret = cl.execute(
                        "insert into %s values (%d , NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL)" %
                        (self.subtb_prefix+str(k)+'_'+str(j), t0+i))
                    else:
                        ret = cl.execute(
                            "insert into %s values (%d , %d,%d,%d,%d,%d,%d,%d,'%s','%s',%d,%d,%d,%d)" %
                            (self.subtb_prefix+str(k)+'_'+str(j), t0+i, i%100, i/2.0, i%41, i%51, i%53, i*1.0, i%2,'taos'+str(i),'涛思'+str(i), i%43, i%23 , i%17 , i%3167))
L
liuyq-617 已提交
374 375 376
        cl.close()
        conn.close()
        
377
    def rest_query(self,sql):                                       #rest 接口
L
liuyq-617 已提交
378 379 380
        host = self.host
        user = self.user
        password = self.password
381 382 383 384
        port =6041
        url = "http://{}:{}/rest/sql".format(host, port )
        try:
            r = requests.post(url, 
L
liuyq-617 已提交
385
                data = 'use %s' % self.dbname,
386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411
                auth = HTTPBasicAuth('root', 'taosdata'))  
            r = requests.post(url, 
                data = sql,
                auth = HTTPBasicAuth('root', 'taosdata'))         
        except:
            print("REST API Failure (TODO: more info here)")
            raise
        rj = r.json()
        if ('status' not in rj):
            raise RuntimeError("No status in REST response")

        if rj['status'] == 'error':  # clearly reported error
            if ('code' not in rj):  # error without code
                raise RuntimeError("REST error return without code")
            errno = rj['code']  # May need to massage this in the future
            # print("Raising programming error with REST return: {}".format(rj))
            raise taos.error.ProgrammingError(
                rj['desc'], errno)  # todo: check existance of 'desc'

        if rj['status'] != 'succ':  # better be this
            raise RuntimeError(
                "Unexpected REST return status: {}".format(
                    rj['status']))

        nRows = rj['rows'] if ('rows' in rj) else 0
        return nRows
412

413
    
414
    def query_thread_n(self,threadID):                      #使用原生python接口查询
L
liuyq-617 已提交
415 416 417
        host = self.host
        user = self.user
        password = self.password
L
liuyq-617 已提交
418 419 420 421 422 423
        conn = taos.connect(
            host,
            user,
            password,
            )
        cl = conn.cursor()
L
liuyq-617 已提交
424
        cl.execute("use %s;" % self.dbname)
L
liuyq-617 已提交
425
        fo = open('bak_sql_n_%d'%threadID,'w+')
L
liuyq-617 已提交
426
        print("Thread %d: starting" % threadID)
L
liuyq-617 已提交
427 428
        loop = self.loop
        while loop:
429
            
L
liuyq-617 已提交
430
                try:
431 432 433 434
                    if self.random_pick():
                        sql=self.gen_query_sql()
                    else:
                        sql=self.gen_query_join()
435
                    print("sql is ",sql)
L
liuyq-617 已提交
436
                    fo.write(sql+'\n')
L
liuyq-617 已提交
437
                    start = time.time()
438 439
                    cl.execute(sql)
                    cl.fetchall()
L
liuyq-617 已提交
440 441
                    end = time.time()
                    print("time cost :",end-start)
L
liuyq-617 已提交
442
                except Exception as e:
L
liuyq-617 已提交
443
                    print('-'*40)
L
liuyq-617 已提交
444
                    print(
L
liuyq-617 已提交
445
                "Failure thread%d, sql: %s \nexception: %s" %
446
                (threadID, str(sql),str(e)))
L
liuyq-617 已提交
447 448 449
                    err_uec='Unable to establish connection'
                    if err_uec in str(e) and loop >0:
                        exit(-1)
L
liuyq-617 已提交
450 451
                loop -= 1
                if loop == 0: break
L
liuyq-617 已提交
452
        fo.close()            
L
liuyq-617 已提交
453 454
        cl.close()
        conn.close()       
455
        print("Thread %d: finishing" % threadID)
L
liuyq-617 已提交
456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490

    def query_thread_nr(self,threadID):                      #使用原生python接口进行重放
        host = self.host
        user = self.user
        password = self.password
        conn = taos.connect(
            host,
            user,
            password,
            )
        cl = conn.cursor()
        cl.execute("use %s;" % self.dbname)
        replay_sql = []
        with  open('bak_sql_n_%d'%threadID,'r') as f:
            replay_sql = f.readlines()
        print("Replay Thread %d: starting" % threadID)
        for sql in replay_sql:
            try:
                print("sql is ",sql)
                start = time.time()
                cl.execute(sql)
                cl.fetchall()
                end = time.time()
                print("time cost :",end-start)
            except Exception as e:
                print('-'*40)
                print(
            "Failure thread%d, sql: %s \nexception: %s" %
            (threadID, str(sql),str(e)))
                err_uec='Unable to establish connection'
                if err_uec in str(e) and loop >0:
                    exit(-1)         
        cl.close()
        conn.close()       
        print("Replay Thread %d: finishing" % threadID)
L
liuyq-617 已提交
491
          
492
    def query_thread_r(self,threadID):                      #使用rest接口查询
493
        print("Thread %d: starting" % threadID)
L
liuyq-617 已提交
494
        fo = open('bak_sql_r_%d'%threadID,'w+')
L
liuyq-617 已提交
495 496 497 498 499 500 501 502
        loop = self.loop
        while loop:
            try:
                if self.random_pick():
                    sql=self.gen_query_sql()
                else:
                    sql=self.gen_query_join()
                print("sql is ",sql)
L
liuyq-617 已提交
503
                fo.write(sql+'\n')
L
liuyq-617 已提交
504 505 506 507 508 509 510 511 512
                start = time.time()
                self.rest_query(sql)
                end = time.time()
                print("time cost :",end-start)
            except Exception as e:
                print('-'*40)
                print(
            "Failure thread%d, sql: %s \nexception: %s" %
            (threadID, str(sql),str(e)))
L
liuyq-617 已提交
513 514 515
                err_uec='Unable to establish connection'
                if err_uec in str(e) and loop >0:
                    exit(-1)
L
liuyq-617 已提交
516 517
            loop -= 1    
            if loop == 0: break
L
liuyq-617 已提交
518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542
        fo.close()      
        print("Thread %d: finishing" % threadID)   

    def query_thread_rr(self,threadID):                      #使用rest接口重放
        print("Replay Thread %d: starting" % threadID)
        replay_sql = []
        with  open('bak_sql_r_%d'%threadID,'r') as f:
            replay_sql = f.readlines()

        for sql in replay_sql:
            try:
                print("sql is ",sql)
                start = time.time()
                self.rest_query(sql)
                end = time.time()
                print("time cost :",end-start)
            except Exception as e:
                print('-'*40)
                print(
            "Failure thread%d, sql: %s \nexception: %s" %
            (threadID, str(sql),str(e)))
                err_uec='Unable to establish connection'
                if err_uec in str(e) and loop >0:
                    exit(-1)  
        print("Replay Thread %d: finishing" % threadID)  
L
liuyq-617 已提交
543 544

    def run(self):
545
        print(self.n_numOfTherads,self.r_numOfTherads)  
L
liuyq-617 已提交
546
        threads = []
L
liuyq-617 已提交
547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564
        if self.replay:  #whether replay 
            for i in range(self.n_numOfTherads):
                thread = threading.Thread(target=self.query_thread_nr, args=(i,))
                threads.append(thread)
                thread.start()  
            for i in range(self.r_numOfTherads):
                thread = threading.Thread(target=self.query_thread_rr, args=(i,))
                threads.append(thread)
                thread.start()
        else:
            for i in range(self.n_numOfTherads):
                thread = threading.Thread(target=self.query_thread_n, args=(i,))
                threads.append(thread)
                thread.start()  
            for i in range(self.r_numOfTherads):
                thread = threading.Thread(target=self.query_thread_r, args=(i,))
                threads.append(thread)
                thread.start()
L
liuyq-617 已提交
565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671
 
parser = argparse.ArgumentParser()
parser.add_argument(
    '-H',
    '--host-name',
    action='store',
    default='127.0.0.1',
    type=str,
    help='host name to be connected (default: 127.0.0.1)')
parser.add_argument(
    '-S',
    '--ts',
    action='store',
    default=1500000000000,
    type=int,
    help='insert data from timestamp (default: 1500000000000)')
parser.add_argument(
    '-d',
    '--db-name',
    action='store',
    default='test',
    type=str,
    help='Database name to be created (default: test)')
parser.add_argument(
    '-t',
    '--number-of-native-threads',
    action='store',
    default=10,
    type=int,
    help='Number of native threads (default: 10)')
parser.add_argument(
    '-T',
    '--number-of-rest-threads',
    action='store',
    default=10,
    type=int,
    help='Number of rest threads (default: 10)')
parser.add_argument(
    '-r',
    '--number-of-records',
    action='store',
    default=100,
    type=int,
    help='Number of record to be created for each table  (default: 100)')
parser.add_argument(
    '-c',
    '--create-table',
    action='store',
    default='0',
    type=int,
    help='whether gen data (default: 0)')
parser.add_argument(
    '-p',
    '--subtb-name-prefix',
    action='store',
    default='t',
    type=str,
    help='subtable-name-prefix (default: t)')
parser.add_argument(
    '-P',
    '--stb-name-prefix',
    action='store',
    default='st',
    type=str,
    help='stable-name-prefix (default: st)')
parser.add_argument(
    '-b',
    '--probabilities',
    action='store',
    default='0.05',
    type=float,
    help='probabilities of join (default: 0.05)')
parser.add_argument(
    '-l',
    '--loop-per-thread',
    action='store',
    default='100',
    type=int,
    help='loop per thread (default: 100)')
parser.add_argument(
    '-u',
    '--user',
    action='store', 
    default='root',
    type=str,
    help='user name')
parser.add_argument(
    '-w',
    '--password',
    action='store', 
    default='root',
    type=str,
    help='user name')
parser.add_argument(
    '-n',
    '--number-of-tables',
    action='store',
    default=1000,
    type=int,
    help='Number of subtales per stable (default: 1000)')
parser.add_argument(
    '-N',
    '--number-of-stables',
    action='store',
    default=2,
    type=int,
    help='Number of stables  (default: 2)')
L
liuyq-617 已提交
672 673 674 675 676 677 678
parser.add_argument(
    '-m',
    '--mix-stable-subtable',
    action='store',
    default=0,
    type=int,
    help='0:stable & substable ,1:subtable ,2:stable (default: 0)')
L
liuyq-617 已提交
679 680 681 682 683 684 685
parser.add_argument(
    '-R',
    '--replay',
    action='store',
    default=0,
    type=int,
    help='0:not replay ,1:replay  (default: 0)')
L
liuyq-617 已提交
686 687 688 689 690

args = parser.parse_args()
q = ConcurrentInquiry(
    args.ts,args.host_name,args.user,args.password,args.db_name,
                args.stb_name_prefix,args.subtb_name_prefix,args.number_of_native_threads,args.number_of_rest_threads,
L
liuyq-617 已提交
691
                args.probabilities,args.loop_per_thread,args.number_of_stables,args.number_of_tables ,args.number_of_records,
L
liuyq-617 已提交
692
                args.mix_stable_subtable, args.replay )
L
liuyq-617 已提交
693 694 695

if args.create_table: 
    q.gen_data()
696
q.get_full()
L
liuyq-617 已提交
697

698
#q.gen_query_sql()
L
liuyq-617 已提交
699
q.run()
L
liuyq-617 已提交
700