mql_run.go 41 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338
  1. package odbcmql
  2. import (
  3. "bufio"
  4. "context"
  5. "encoding/json"
  6. "fmt"
  7. "math/rand"
  8. "os"
  9. "path/filepath"
  10. "regexp"
  11. "runtime"
  12. "strconv"
  13. "strings"
  14. "sync"
  15. "sync/atomic"
  16. "testing"
  17. "time"
  18. odb "git.wecise.com/wecise/odb-go/odb"
  19. "git.wecise.com/wecise/odb-go/odb/mql/qstru"
  20. "gitee.com/wecisecode/util/cast"
  21. mcfg "gitee.com/wecisecode/util/cfg"
  22. "gitee.com/wecisecode/util/filewalker"
  23. "gitee.com/wecisecode/util/pqc"
  24. "gitee.com/wecisecode/util/set/strset"
  25. "gitee.com/wecisecode/util/spliter"
  26. "github.com/stretchr/testify/assert"
  27. )
  28. var ODBC odb.Client
  29. var ODBError error
  30. var debug bool
  31. // var checkCassSchema bool
  32. var default_keyspace = `oktest`
  33. var default_odbpaths = `127.0.0.1:11001`
  34. var ksnative = default_keyspace + "_native"
  35. func init() {
  36. logprefix := config.GetString("log.prefix", "")
  37. if logprefix != "" {
  38. logprefix += " "
  39. }
  40. logger.SetFormat(logprefix+"yyyy-MM-dd HH:mm:ss.SSSSSS [pid] [level] msg", "\n")
  41. odbpaths := strset.New(strings.Split(mcfg.CommandArgs.GetString("odb",
  42. strings.Join(mcfg.Environs.GetStrings("ODBPATH",
  43. config.GetStrings("odbc.odbpath", default_odbpaths)...), ",")), ",")...).List()
  44. keyspace := mcfg.CommandArgs.GetString("keyspace",
  45. mcfg.Environs.GetString("KEYSPACE", config.GetString("odbc.keyspace", default_keyspace)))
  46. cassdc := mcfg.CommandArgs.GetString("cassandra.dc",
  47. mcfg.Environs.GetString("CASSANDRADC", config.GetString("cassandra.dc", "")))
  48. if cassdc == "" || cassdc == "dc1" {
  49. ksnative = keyspace + "_native"
  50. } else {
  51. ksnative = keyspace + "_" + cassdc + "_native"
  52. }
  53. ksnative = mcfg.CommandArgs.GetString("ksnative",
  54. mcfg.Environs.GetString("KSNATIVE", config.GetString("odbc.ksnative", ksnative)))
  55. debug = mcfg.CommandArgs.GetBool("debug", false)
  56. // checkCassSchema = mcfg.CommandArgs.GetBool("ccs", true)
  57. ODBC, ODBError = odb.NewClient(&odb.Config{
  58. Keyspace: keyspace,
  59. Hosts: odbpaths,
  60. Debug: debug,
  61. })
  62. }
  63. func Run(t *testing.T) {
  64. if mcfg.CommandArgs.GetString("keyspace?") == "keyspace?" {
  65. if ODBC != nil {
  66. fmt.Println(ODBC.Config().Keyspace)
  67. os.Exit(0)
  68. }
  69. os.Exit(0)
  70. }
  71. if mcfg.CommandArgs.GetString("odbpath?") == "odbpath?" {
  72. if ODBC != nil {
  73. fmt.Println(ODBC.Config().Hosts)
  74. os.Exit(0)
  75. }
  76. os.Exit(0)
  77. }
  78. spath := strings.Split(config.GetString("test.mql.path"), " ")
  79. if len(os.Args) > 1 {
  80. if len(os.Args) > 2 && os.Args[1] == "-" {
  81. s := strings.Join(os.Args[2:], " ")
  82. spath = []string{"/tmp/1.mql"}
  83. os.WriteFile(spath[0], []byte(s+"/**output()**/"), os.ModePerm)
  84. } else {
  85. for i := 1; i < len(os.Args); i++ {
  86. ap := os.Args[i]
  87. if strings.HasPrefix(ap, "-test.") {
  88. break
  89. }
  90. if strings.Contains(ap, "=") {
  91. continue
  92. }
  93. if (ap == "-" || ap == "--") && i+1 < len(os.Args) {
  94. s := strings.Join(os.Args[i+1:], " ")
  95. spath = []string{"/tmp/1.mql"}
  96. os.WriteFile(spath[0], []byte(s+"\n/**output()**/"), os.ModePerm)
  97. break
  98. }
  99. spath = append(spath, ap)
  100. }
  101. }
  102. }
  103. if len(spath) == 0 || len(spath) == 1 && len(spath[0]) == 0 {
  104. fmt.Print(`循环遍历执行指定路径下的 mql 文件
  105. usage: mql path [options] -- mqlstatement
  106. path 指定mql文件所在路径,通配符 ** 表示任意字符,* 表示除分隔符以外的任意字符, . 表示递归当前目录下的所有子目录
  107. 为避免 shell 自动将 * 转换为文件名列表,可以将指定的 path 用引号包含
  108. mqlstatement 指定要执行的 mql 语句,支持多条语句,语句之间用分号 ; 分隔
  109. options:
  110. match=^\\d+.* 文件名匹配正则表达式,默认为所有以数字开头命名的文件
  111. odb=` + default_odbpaths + ` 指定odbserver路径,默认通过环境变量ODBPATH或通过ETCD相关配置获取
  112. keyspace=` + default_keyspace + ` 指定keyspace,默认通过环境变量KEYSPACE获取
  113. ksnative=` + default_keyspace + `_native 指定native keyspace,默认通过环境变量KSNATIVE获取
  114. debug=true 开启调试模式,输出更多信息
  115. 环境变量需求:
  116. KEYSPACE=` + default_keyspace + `
  117. KSNATIVE=` + default_keyspace + `_native
  118. ODBPATH=` + default_odbpaths + `
  119. ETCDPATH=127.0.0.1:2379
  120. ETCDUSER=
  121. ETCDPASS=
  122. CASSANDRAPATH=127.0.0.1
  123. CASSANDRALOOKUP=false
  124. NATSPATH=nats://user:N5JbKeT1C3uOUh317OVXjg==@127.0.0.1:4222
  125. LANG=zh_CN.utf8
  126. LC_ALL=zh_CN.utf8
  127. 可通过mql.conf配置运行参数:
  128. [odbc]
  129. ;指定odbserver路径
  130. odbpath=` + default_odbpaths + `
  131. ;指定keyspace
  132. keyspace=` + default_keyspace + `
  133. ;指定native keyspace
  134. ksnative=` + default_keyspace + `_native
  135. mql语句扩展说明:
  136. mql语句中的内容替换:
  137. 随机字符替换:
  138. mql语句中的 '...N Bytes...' 会被替换为 N 个随机可见字符,N 为自然数
  139. 变量替换:
  140. mql语句中 '{%<f>,<varname>}' 形式的内容会被替换为,以 %<f> 作为格式化标记的 <varname> 变量的内容
  141. 目前支持的 <varname> 包括:
  142. keyspace 当前指定的 Keyspace
  143. ksnative Native Keyspace
  144. mqli 当前mql语句的循环值,参考 loop(N)
  145. filei 当前文件的循环值,参考 scope(file) loop(N)
  146. diri 当前目录的循环值,参考 scope(dir) loop(N)
  147. topi 整个执行进程的循环值,参考 scope(top) loop(N)
  148. mqlcount 当前mql语句的总计数
  149. filecount 当前文件的总计数
  150. dircount 当前目录的总计数
  151. topcount 整个执行进程的总计数
  152. mql语句中第一个 /** 到 最后一个 **/ 之间的内容可以标注执行若干预定义动作,多个动作标注用空格分隔
  153. 如 /** output() sleep(1) **/ 表示忽略当前语句执行过程中的任何报错,执行完成后等待一秒后再继续执行下一语句
  154. 目前支持的预定义动作如下:
  155. skip() 跳过当前语句,以及其它的所有预定义动作
  156. 可用于暂时屏蔽一条语句的执行
  157. params(V) 用于提供 prepare 语句中 ? 的对应值,V 为 JSON 格式编码的数组
  158. 对应值中不能有包含逗号和右括号,如果需要逗号用 \u002c 代替,右括号用 \u0029 代替
  159. subscribe(S) 订阅指定主题的消息通知 S,输出收到的通知信息,S 为要订阅主题的的名称字符串,同一主题只会被订阅一次
  160. unsubscribe(S) 取消订阅指定主题的消息通知 S,S 为要订阅主题的名称字符串
  161. fork(G) 新建并行线程分支,执行当前语句的同时,继续执行后续语句,G 为线程组名,配合 wait 使用
  162. wait(G) 执行当前语句前,等待之前创建的分支结束,G 为线程组名,对应由 fork 创建的所有同名线程分支,G 为空表示所有
  163. beforerun(F,...) 执行当前语句前,执行一系列预定义函数,以通过一些简单逻辑处理产生模拟数据,目前支持的函数包括
  164. set(S,V) 设置变量,S 为变量名,V为变量值
  165. add(S,V,u) 累加变量,S 为变量名,V为增量值,u为可选的变量单位,如时间段单位 second,minute,hour
  166. mod(S,V) 变量取模,相当于 S = S % V
  167. case(C,F) 根据条件 C 执行函数 F
  168. ------------- 针对错误处理的相关动作
  169. retry(N) 出错重试 N 次,N为自然数,默认为 0,
  170. onerror(O,E,...) retry后仍然有错时的处理方法,E为特定的错误信息关键字,不指定则为任意错误,O为处理方法,包括:
  171. break 中断当前语句的循环,继续顺序执行其它语句,没有循环时与pass无区别
  172. continue 忽略报错,继续执行,在循环中时,继续下一轮循环
  173. must 报错才正常,否则中断执行,报告期待错误信息不符,退出当前进程
  174. exit 停止执行,退出当前进程,此为默认处理方式
  175. noerrinfo() 存在 retry 或 onerror 处理时,不输出错误信息
  176. ------------- 以上的预定义动作只对当前语句起作用,下面的预定义动作可以指定其作用范围
  177. scope(S) 相关动作的作用范围 S 包括 top,dir,file,mql,默认作用范围为 mql 当前语句
  178. 如:scope(file) parallel(3) loop(5) 并发循环执行本文件五次,最大并行数限制为3
  179. qmeta(O) mql执行时设置特定的选项,O为json对象
  180. timeout(D) mql执行超时设置,D 为时间段,默认为一分钟
  181. 时间段 D 支持单位 d(天),h(时),m(分),s(秒),ms(毫秒),us(微秒),ns(纳秒),默认为毫秒
  182. loop(N) 循环执行 N 次,N 为正整数,默认执行一次
  183. 执行前会替换 mql 语句及 params 参数中的循环次数标记
  184. loopfrom(N) 设置循环执行计数起始值 N,默认为 1,该值仅影响替换 mql 语句及 params 参数中的循环次数标记
  185. loopstep(N) 设置循环执行计数步长值 N,默认为 1,该值仅影响替换 mql 语句及 params 参数中的循环次数标记
  186. parallel(N) 并发执行,N 为最大并行数,不指定 N 表示不限制并行数
  187. set(S,V) 设置变量,S 为变量名,V为变量值
  188. ------------- 以上为执行语句前的预定义动作,下面是执行完成后的预定义动作
  189. sleep(D) 执行完成后等待时间段 D 后再继续执行下一语句
  190. output() 输出执行结果
  191. outputcount() 输出执行结果记录条数
  192. metainfo() 输出执行结果相关的元信息
  193. schema(C) 执行 schema 命令检查指定类 C 是否存在
  194. count(N) 检查返回结果中的数据记录数是否为 N,N 为自然数
  195. equal(N,F,V) 判断返回结果中第 N 条数据的字段 F 的值是否为 V
  196. match(Kn,Mn) 检查返回结果中字段 Kn 值为 Mn 的记录是否存在,参数 Kn,Mn 可以有多个,需要成对出现
  197. matchcount(Kn,Mn,N) 检查返回结果中字段 Kn 值为 Mn 的记录数是否为 N,参数 Kn,Mn 可以有多个,需要成对出现,N为自然数
  198. 例:循环遍历执行当前路径下的所有 mql 文件
  199. ./mql .
  200. `)
  201. os.Exit(0)
  202. return
  203. }
  204. // 确定MODB连接
  205. if !assert.Nil(t, ODBError) {
  206. return
  207. }
  208. logger.Info("odbpath :", ODBC.Config().Hosts, ODBC.Config().Port)
  209. logger.Info("keyspace :", ODBC.Config().Keyspace)
  210. logger.Info("version :", ODBC.Versline())
  211. logger.Info("ksnative :", ksnative)
  212. logger.Info("debug :", debug)
  213. logger.Info("spath :", spath)
  214. // if checkCassSchema && !cass.CheckCassandraSchema(t) {
  215. // assert.True(t, false, "cassandra schemda is different")
  216. // return
  217. // }
  218. // 文件名以数字开头
  219. fnmatch := mcfg.CommandArgs.GetString("match", `^\d+.*`)
  220. fw, err := filewalker.NewFileWalker(spath, fnmatch) // orderby: dirfirst, filefirst, fullpath
  221. if !assert.Nil(t, err, err) {
  222. return
  223. }
  224. logger.Info("walkdir: ", fw.WalkDir)
  225. logger.Info("pathmatch: ", fw.RePath)
  226. logger.Info("filematch: ", fw.ReFile)
  227. // test
  228. fns := []string{}
  229. fw.List(func(basedir, fpath string) bool { fns = append(fns, filepath.Join(basedir, fpath)); return true })
  230. logger.Info(fmt.Sprint("fns:", "\n", strings.Join(fns, "\n")))
  231. if len(fns) == 0 {
  232. logger.Info("没有找到 MQL 文件")
  233. return
  234. }
  235. logger.Info("共找到", len(fns), "个文件")
  236. // 顺序读取当前目录下文件名为数字开头的文件,执行其中的mql
  237. (&MQLTest{}).Run(t, fw)
  238. // 等待延迟显示信息输出
  239. // time.Sleep(1 * time.Second)
  240. logger.Info("关闭与服务器的连接")
  241. ODBC.Close()
  242. // 等待正常结束,只是为了发现隐含的问题
  243. // for n := 0; runtime.NumGoroutine() > 60 && n < 10; n++ {
  244. // fmt.Println("剩余协程数:", runtime.NumGoroutine())
  245. // time.Sleep(1000 * time.Millisecond)
  246. // }
  247. logger.Info("剩余协程数:", runtime.NumGoroutine())
  248. }
  249. // action(a1,a2...)
  250. // --注释后的大括号之间的内容 -- {}
  251. // mql query之后执行的动作
  252. var reactions = regexp.MustCompile(`(\w+)\s*\(([^\)]*)\)`)
  253. // 替换为随机可见字符
  254. var rereplace_nbytes = regexp.MustCompile(`\.\.\.\s*(\d+)\s*[Bb][Yy][Tt][Ee]s?\s*\.\.\.`)
  255. var reactioneexprs = regexp.MustCompile(`(?s)\/\*\*(?:\s*(\w+\s*\(\s*.*\)\s*)*)*\*\*\/`)
  256. var reactioneexprs_2 = regexp.MustCompile(`(?s)(?:(?:^|\n)\-\-\s*\{(.*)\}\s*)+$`)
  257. var reactioneexprs_3 = regexp.MustCompile(`(?s)(?:^\-\-\s*\{(.*)\}\s*)+\n`)
  258. var commentexprs = regexp.MustCompile(`(?s)\/\*(?:[^\*]|\*+[^\*\/])*\*+\/`)
  259. var commentexprs_2 = regexp.MustCompile(`(?ms)(?:^|\n)\-\-[^\n]*(?:\n|$)`)
  260. var commentexprs_3 = regexp.MustCompile(`(?ms)(?:^|\n)//[^\n]*(?:\n|$)`)
  261. type MQLTest struct {
  262. t *testing.T
  263. fw *filewalker.FileWalker
  264. scopevars *ScopeVars
  265. }
  266. type ScopeVars struct {
  267. sync.RWMutex
  268. top *Variables
  269. dir map[string]*Variables
  270. file map[string]*Variables
  271. mql map[string]*Variables
  272. }
  273. type Variables struct {
  274. vars map[string]interface{}
  275. loop_count int
  276. loop_from int
  277. loop_step int
  278. timeout time.Duration
  279. qmeta odb.QueryMeta
  280. parallel_max int
  281. }
  282. type CurrentVars struct {
  283. loop_i int
  284. ch_parallel_count chan<- int
  285. mqlcount int32
  286. sleeptime time.Duration
  287. totalusetime time.Duration
  288. maxusetime time.Duration
  289. minusetime time.Duration
  290. }
  291. type GlobalVars struct {
  292. sync.RWMutex
  293. *CurrentVars
  294. wg_wait_fork_routine map[string]*sync.WaitGroup
  295. ch_wait_mql_done map[string]chan bool
  296. }
  297. func (mt *MQLTest) Run(t *testing.T, fw *filewalker.FileWalker) (ok bool) {
  298. mt.t = t
  299. mt.fw = fw
  300. global := &GlobalVars{
  301. CurrentVars: &CurrentVars{},
  302. wg_wait_fork_routine: make(map[string]*sync.WaitGroup),
  303. ch_wait_mql_done: make(map[string]chan bool),
  304. }
  305. mt.scopevars = &ScopeVars{
  306. top: &Variables{
  307. vars: map[string]interface{}{},
  308. loop_count: 1,
  309. loop_from: 1,
  310. loop_step: 1},
  311. dir: map[string]*Variables{},
  312. file: map[string]*Variables{},
  313. mql: map[string]*Variables{}}
  314. ctx, cancel := context.WithCancel(context.Background())
  315. defer cancel()
  316. var wg sync.WaitGroup
  317. st := time.Now()
  318. loop_i := 0
  319. parallel_queue := pqc.NewQueue[any](0)
  320. mqlcount := int32(0)
  321. for {
  322. mt.scopevars.Lock()
  323. ok := loop_i < mt.scopevars.top.loop_count
  324. mt.scopevars.Unlock()
  325. if !ok {
  326. break
  327. }
  328. loop_i++
  329. //
  330. ch_parallel_count := make(chan int)
  331. topvars := &CurrentVars{
  332. loop_i: loop_i,
  333. ch_parallel_count: ch_parallel_count,
  334. }
  335. //
  336. ok_chan := make(chan bool, 1)
  337. parallel_chan := make(chan bool, 1)
  338. parallel := false
  339. parallelcount := 0
  340. done := false
  341. wg.Add(1)
  342. go func() {
  343. defer func() {
  344. atomic.AddInt32(&mqlcount, topvars.mqlcount)
  345. wg.Done()
  346. }()
  347. ch_ok := make(chan bool)
  348. go func() {
  349. for {
  350. select {
  351. case <-ch_parallel_count:
  352. if !done && !parallel {
  353. parallel = true
  354. // 加入并发控制队列
  355. if parallelcount > 0 {
  356. if parallelcount > parallel_queue.Size() {
  357. parallel_queue.Growth(parallelcount)
  358. }
  359. parallel_queue.Push(1)
  360. }
  361. parallel_chan <- true
  362. }
  363. case ok := <-ch_ok:
  364. ok_chan <- ok
  365. if parallel {
  366. if parallelcount > 0 {
  367. // 从并发控制队列中移除
  368. parallel_queue.Pop()
  369. }
  370. } else {
  371. done = true
  372. }
  373. return
  374. }
  375. }
  376. }()
  377. ok := mt.RunAll(t, ctx,
  378. global,
  379. topvars)
  380. ch_ok <- ok
  381. if !ok {
  382. cancel() // 并发测试,有一个线程出错,就全停
  383. return
  384. }
  385. }()
  386. success := true
  387. select {
  388. case success = <-ok_chan: // 非并发,等待完成
  389. case <-parallel_chan: // 并发,执行继续下一次
  390. logger.Info("第", topvars.loop_i, "次并发执行继续")
  391. }
  392. if !success {
  393. return false // 失败,不等,直接返回
  394. }
  395. }
  396. wg.Wait()
  397. mt.scopevars.RLock()
  398. loop_count := mt.scopevars.top.loop_count
  399. mt.scopevars.RUnlock()
  400. ut := time.Since(st)
  401. aut := time.Duration(0)
  402. if mqlcount > 0 {
  403. aut = global.totalusetime / time.Duration(mqlcount)
  404. }
  405. logger.Info("完成 ", loop_count, " 次执行,共", mqlcount, "次 MQL 请求,耗时", ut, "单条响应时间", global.minusetime, "~", global.maxusetime, "/", aut, "平均每秒吞吐量", (int64(mqlcount)*int64(time.Second))/int64(ut))
  406. return
  407. }
  408. func (mt *MQLTest) RunAll(t *testing.T, ctx context.Context, global *GlobalVars, topvars *CurrentVars) bool {
  409. logger.Info("开始第", topvars.loop_i, "次执行")
  410. st := time.Now()
  411. // 读取文件列表
  412. listdirs := []string{}
  413. dirfiles := map[string][]string{}
  414. err := mt.fw.List(func(basedir, filename string) bool {
  415. if dirfiles[basedir] == nil {
  416. listdirs = append(listdirs, basedir)
  417. }
  418. dirfiles[basedir] = append(dirfiles[basedir], filename)
  419. return true
  420. })
  421. if !assert.Nil(t, err, err) {
  422. return false
  423. }
  424. // 执行所有目录
  425. for _, basedir := range listdirs {
  426. // 循环执行目录下所有文件
  427. mt.scopevars.Lock()
  428. if mt.scopevars.dir[basedir] == nil {
  429. mt.scopevars.dir[basedir] = &Variables{
  430. vars: map[string]interface{}{},
  431. loop_count: 1,
  432. loop_from: 1,
  433. loop_step: 1}
  434. }
  435. mt.scopevars.Unlock()
  436. var wg sync.WaitGroup
  437. st := time.Now()
  438. loop_i := 0
  439. parallel_queue := pqc.NewQueue[any](0)
  440. mqlcount := int32(0)
  441. for {
  442. mt.scopevars.Lock()
  443. ok := loop_i < mt.scopevars.dir[basedir].loop_count
  444. mt.scopevars.Unlock()
  445. if !ok {
  446. break
  447. }
  448. loop_i++
  449. //
  450. ch_parallel_count := make(chan int)
  451. dirvars := &CurrentVars{
  452. loop_i: loop_i,
  453. ch_parallel_count: ch_parallel_count,
  454. }
  455. //
  456. ok_chan := make(chan bool, 1)
  457. parallel_chan := make(chan bool, 1)
  458. parallel := false
  459. parallelcount := 0
  460. done := false
  461. wg.Add(1)
  462. go func(basedir string) {
  463. defer func() {
  464. atomic.AddInt32(&mqlcount, dirvars.mqlcount)
  465. wg.Done()
  466. }()
  467. ch_ok := make(chan bool)
  468. go func() {
  469. for {
  470. select {
  471. case <-ch_parallel_count:
  472. if !done && !parallel {
  473. parallel = true
  474. // 加入并发控制队列
  475. if parallelcount > 0 {
  476. if parallelcount > parallel_queue.Size() {
  477. parallel_queue.Growth(parallelcount)
  478. }
  479. parallel_queue.Push(1)
  480. }
  481. parallel_chan <- true
  482. }
  483. case ok := <-ch_ok:
  484. ok_chan <- ok
  485. if parallel {
  486. if parallelcount > 0 {
  487. // 从并发控制队列中移除
  488. parallel_queue.Pop()
  489. }
  490. } else {
  491. done = true
  492. }
  493. return
  494. }
  495. }
  496. }()
  497. // logger.Info("dir", basedir, "第", fmt.Sprint(topvars.loop_i, ".", dirvars.loop_i), "次执行开始")
  498. files := dirfiles[basedir]
  499. ch_ok <- mt.RunDir(t, ctx,
  500. global,
  501. topvars,
  502. dirvars,
  503. basedir, files)
  504. // logger.Info("dir", basedir, "第", fmt.Sprint(topvars.loop_i, ".", dirvars.loop_i), "次执行结束")
  505. }(basedir)
  506. success := true
  507. select {
  508. case success = <-ok_chan: // 非并发,等待完成
  509. logger.Info("dir", basedir, "第", fmt.Sprint(topvars.loop_i, ".", dirvars.loop_i), "次顺序执行完成")
  510. case <-parallel_chan: // 并发,执行继续下一次
  511. logger.Info("dir", basedir, "第", fmt.Sprint(topvars.loop_i, ".", dirvars.loop_i), "次并发执行继续")
  512. }
  513. if !success {
  514. return false
  515. }
  516. }
  517. wg.Wait()
  518. mt.scopevars.RLock()
  519. loop_count := mt.scopevars.dir[basedir].loop_count
  520. mt.scopevars.RUnlock()
  521. if loop_count > 1 {
  522. ut := time.Since(st)
  523. logger.Info(fmt.Sprint("dir ", basedir, " loop ", loop_count, " times, run ", mqlcount, " mqls, usetime ", ut))
  524. }
  525. }
  526. logger.Info("完成第", topvars.loop_i, "次执行,共", topvars.mqlcount, "次 MQL 请求,耗时", time.Since(st))
  527. return true
  528. }
  529. func (mt *MQLTest) RunDir(t *testing.T, ctx context.Context,
  530. global *GlobalVars,
  531. topvars *CurrentVars,
  532. dirvars *CurrentVars,
  533. basedir string, files []string) bool {
  534. for _, filename := range files {
  535. ok := mt.RunFile(t, ctx, global, topvars, dirvars, basedir, filename)
  536. if !ok {
  537. return false
  538. }
  539. }
  540. return true
  541. }
  542. func (mt *MQLTest) RunFile(t *testing.T, ctx context.Context,
  543. global *GlobalVars,
  544. topvars *CurrentVars,
  545. dirvars *CurrentVars,
  546. basedir, filename string) bool {
  547. // 读取文件内容
  548. ffpath := filepath.Join(basedir, filename)
  549. bs, err := os.ReadFile(ffpath)
  550. if !assert.Nil(t, err, err) {
  551. return false
  552. }
  553. // mql语句切分
  554. mqgr := NewMQLGroupRequest()
  555. var multilines *MQLRequest
  556. mqs := spliter.NewMQLSpliter(bufio.NewReader(strings.NewReader(string(bs))))
  557. for {
  558. mql, fromline, toline, fromchar, tochar, hasnext, _ := mqs.NextMQL()
  559. if !hasnext {
  560. break
  561. }
  562. if multilines != nil {
  563. if strings.TrimSpace(mql) == "multilines end" {
  564. e := mqgr.Append(multilines)
  565. if !assert.Nil(t, e, e) {
  566. return false
  567. }
  568. multilines = nil
  569. } else if multilines.OriginQueryString == "" {
  570. multilines.OriginQueryString = mql
  571. multilines.Fromline = fromline
  572. multilines.Toline = toline
  573. multilines.Fromchar = fromchar
  574. multilines.Tochar = tochar
  575. } else {
  576. multilines.OriginQueryString += ";" + mql
  577. multilines.Toline = toline
  578. multilines.Tochar = tochar
  579. }
  580. } else if strings.TrimSpace(mql) == "multilines begin" {
  581. multilines = &MQLRequest{FilePath: ffpath}
  582. } else {
  583. mqr := &MQLRequest{OriginQueryString: mql, FilePath: ffpath, Fromline: fromline, Toline: toline, Fromchar: fromchar, Tochar: tochar}
  584. e := mqgr.Append(mqr)
  585. if !assert.Nil(t, e, e) {
  586. return false
  587. }
  588. }
  589. }
  590. // mqls := spliter.MQLSplit(string(bs))
  591. mt.scopevars.Lock()
  592. if mt.scopevars.file[ffpath] == nil {
  593. mt.scopevars.file[ffpath] = &Variables{
  594. vars: map[string]interface{}{},
  595. loop_count: 1,
  596. loop_from: 1,
  597. loop_step: 1}
  598. }
  599. mt.scopevars.Unlock()
  600. var wg sync.WaitGroup
  601. st := time.Now()
  602. loop_i := 0
  603. parallel_queue := pqc.NewQueue[any](0)
  604. mqlcount := int32(0)
  605. for {
  606. mt.scopevars.Lock()
  607. ok := loop_i < mt.scopevars.file[ffpath].loop_count
  608. mt.scopevars.Unlock()
  609. if !ok {
  610. break
  611. }
  612. loop_i++
  613. //
  614. ch_parallel_count := make(chan int)
  615. filevars := &CurrentVars{
  616. loop_i: loop_i,
  617. ch_parallel_count: ch_parallel_count,
  618. }
  619. //
  620. ok_chan := make(chan bool, 1)
  621. parallel_chan := make(chan bool, 1)
  622. parallel := false
  623. parallelcount := 0
  624. done := false
  625. wg.Add(1)
  626. go func() {
  627. defer func() {
  628. atomic.AddInt32(&mqlcount, filevars.mqlcount)
  629. wg.Done()
  630. }()
  631. ch_ok := make(chan bool)
  632. go func() {
  633. for {
  634. select {
  635. case <-ch_parallel_count:
  636. if !done && !parallel {
  637. parallel = true
  638. // 加入并发控制队列
  639. if parallelcount > 0 {
  640. if parallelcount > parallel_queue.Size() {
  641. parallel_queue.Growth(parallelcount)
  642. }
  643. parallel_queue.Push(1)
  644. }
  645. parallel_chan <- true
  646. }
  647. case ok := <-ch_ok:
  648. ok_chan <- ok
  649. if parallel {
  650. if parallelcount > 0 {
  651. // 从并发控制队列中移除
  652. parallel_queue.Pop()
  653. }
  654. } else {
  655. done = true
  656. }
  657. return
  658. }
  659. }
  660. }()
  661. // logger.Info("file", ffpath, "第", fmt.Sprint(topvars.loop_i, ".", dirvars.loop_i, ".", filevars.loop_i), "次执行开始")
  662. ch_ok <- mt.RunMQLGroup(t, ctx,
  663. global,
  664. topvars,
  665. dirvars,
  666. filevars,
  667. basedir, ffpath, mqgr)
  668. // logger.Info("file", ffpath, "第", fmt.Sprint(topvars.loop_i, ".", dirvars.loop_i, ".", filevars.loop_i), "次执行结束")
  669. }()
  670. success := true
  671. select {
  672. case success = <-ok_chan: // 非并发,等待完成
  673. logger.Info("file", ffpath, "第", fmt.Sprint(topvars.loop_i, ".", dirvars.loop_i, ".", filevars.loop_i), "次顺序执行完成")
  674. case <-parallel_chan: // 并发,执行继续下一次
  675. logger.Info("file", ffpath, "第", fmt.Sprint(topvars.loop_i, ".", dirvars.loop_i, ".", filevars.loop_i), "次并发执行继续")
  676. }
  677. if !success {
  678. return false
  679. }
  680. }
  681. wg.Wait()
  682. mt.scopevars.RLock()
  683. loop_count := mt.scopevars.file[ffpath].loop_count
  684. mt.scopevars.RUnlock()
  685. if loop_count > 1 {
  686. ut := time.Since(st)
  687. sn := fmt.Sprint(topvars.loop_i, ".", dirvars.loop_i)
  688. logger.Info(fmt.Sprint("file ", ffpath+"/"+sn, " loop ", loop_count, " times, run ", mqlcount, " mqls, usetime ", ut))
  689. }
  690. return true
  691. }
  692. func (mt *MQLTest) RunMQLGroup(t *testing.T, ctx context.Context,
  693. global *GlobalVars,
  694. topvars *CurrentVars,
  695. dirvars *CurrentVars,
  696. filevars *CurrentVars,
  697. basedir, ffpath string, mqgr *MQLGroupRequest) (pass bool) {
  698. pass = true
  699. var wg sync.WaitGroup
  700. for _, mqs := range mqgr.mqrs {
  701. wg.Add(1)
  702. go func(mqs []*MQLRequest) {
  703. defer wg.Done()
  704. if len(mqs) == 0 {
  705. return
  706. }
  707. if mqs[0].StaticActions.ForkName != nil {
  708. forkname := *mqs[0].StaticActions.ForkName
  709. global.Lock()
  710. wg := global.wg_wait_fork_routine[forkname]
  711. if wg == nil {
  712. wg = &sync.WaitGroup{}
  713. global.wg_wait_fork_routine[forkname] = wg
  714. }
  715. global.Unlock()
  716. wg.Add(1)
  717. defer wg.Done()
  718. }
  719. ok := mt.RunMQLs(t, ctx, global, topvars, dirvars, filevars, basedir, ffpath, mqs)
  720. if !ok {
  721. pass = false
  722. }
  723. }(mqs)
  724. }
  725. wg.Wait()
  726. return
  727. }
  728. func (mt *MQLTest) RunMQLs(t *testing.T, ctx context.Context,
  729. global *GlobalVars,
  730. topvars *CurrentVars,
  731. dirvars *CurrentVars,
  732. filevars *CurrentVars,
  733. basedir, ffpath string, mqrs []*MQLRequest) bool {
  734. for _, mqr := range mqrs {
  735. mqlkey := mqr.Key
  736. mqlstr := mqr.OriginQueryString
  737. if mqlstr == "" {
  738. continue
  739. }
  740. staticactions := mqr.StaticActions
  741. // 设置执行过程中的控制参数
  742. mt.InitScopeVars(basedir, ffpath, mqlkey, staticactions)
  743. ch_test_run_one_mql_result := make(chan bool)
  744. go func() {
  745. mqrinst := fmt.Sprint(mqlkey, "/", topvars.loop_i, ".", dirvars.loop_i, ".", filevars.loop_i)
  746. global.Lock()
  747. mqrdone := global.ch_wait_mql_done[mqrinst]
  748. if mqrdone == nil {
  749. mqrdone = make(chan bool)
  750. global.ch_wait_mql_done[mqrinst] = mqrdone
  751. }
  752. var waitmqrdone chan bool
  753. if mqr.WaitMQLRequest != nil {
  754. waitmqrkey := mqr.WaitMQLRequest.Key
  755. waitmqrinst := fmt.Sprint(waitmqrkey, "/", topvars.loop_i, ".", dirvars.loop_i, ".", filevars.loop_i)
  756. waitmqrdone = global.ch_wait_mql_done[waitmqrinst]
  757. if waitmqrdone == nil {
  758. waitmqrdone = make(chan bool)
  759. global.ch_wait_mql_done[waitmqrinst] = waitmqrdone
  760. }
  761. }
  762. global.Unlock()
  763. var ret bool
  764. defer func() { mqrdone <- ret }()
  765. if waitmqrdone != nil {
  766. v := <-waitmqrdone
  767. waitmqrdone <- v
  768. if !v {
  769. // 依赖MQR失败
  770. ch_test_run_one_mql_result <- false
  771. return
  772. }
  773. }
  774. ret = mt.RunMQR(t, ctx, global, topvars, dirvars, filevars, basedir, ffpath, mqr)
  775. ch_test_run_one_mql_result <- ret
  776. }()
  777. ret := <-ch_test_run_one_mql_result
  778. if !ret {
  779. return ret
  780. }
  781. // continue
  782. }
  783. return true
  784. }
  785. func (mt *MQLTest) RunMQR(t *testing.T, ctx context.Context,
  786. global *GlobalVars,
  787. topvars *CurrentVars,
  788. dirvars *CurrentVars,
  789. filevars *CurrentVars,
  790. basedir, ffpath string, mqr *MQLRequest) bool {
  791. mqlkey := mqr.Key
  792. staticactionexprs := mqr.StaticActionExprs
  793. staticactions := mqr.StaticActions
  794. return t.Run(mqlkey, func(t *testing.T) {
  795. var wg sync.WaitGroup
  796. st := time.Now()
  797. loop_i := 0
  798. parallel_queue := pqc.NewQueue[any](0)
  799. mqlcount := int32(0)
  800. for {
  801. mt.scopevars.Lock()
  802. ok := loop_i < mt.scopevars.mql[mqlkey].loop_count
  803. mt.scopevars.Unlock()
  804. if !ok {
  805. break
  806. }
  807. loop_i++
  808. //
  809. ch_parallel_count := make(chan int)
  810. mqlvars := &CurrentVars{
  811. loop_i: loop_i,
  812. ch_parallel_count: ch_parallel_count,
  813. }
  814. // 运行实例
  815. mqlsn := fmt.Sprint(mqlkey, "/", topvars.loop_i, ".", dirvars.loop_i, ".", filevars.loop_i, ".", mqlvars.loop_i)
  816. mt.scopevars.Lock()
  817. if mt.scopevars.mql[mqlsn] == nil {
  818. mt.scopevars.mql[mqlsn] = &Variables{
  819. vars: map[string]interface{}{},
  820. loop_count: 1,
  821. loop_from: 1,
  822. loop_step: 1}
  823. }
  824. mt.scopevars.Unlock()
  825. //
  826. mt.scopevars.Lock()
  827. top_loopi := mt.scopevars.top.loop_from + (topvars.loop_i-1)*mt.scopevars.top.loop_step
  828. mt.scopevars.top.vars["topi"] = top_loopi
  829. dir_loopi := mt.scopevars.dir[basedir].loop_from + (dirvars.loop_i-1)*mt.scopevars.dir[basedir].loop_step
  830. mt.scopevars.dir[basedir].vars["diri"] = dir_loopi
  831. mt.scopevars.dir[basedir].vars["topi"] = top_loopi
  832. file_loopi := mt.scopevars.file[ffpath].loop_from + (filevars.loop_i-1)*mt.scopevars.file[ffpath].loop_step
  833. mt.scopevars.file[ffpath].vars["filei"] = file_loopi
  834. mt.scopevars.file[ffpath].vars["diri"] = dir_loopi
  835. mt.scopevars.file[ffpath].vars["topi"] = top_loopi
  836. mql_loopi := mt.scopevars.mql[mqlkey].loop_from + (mqlvars.loop_i-1)*mt.scopevars.mql[mqlkey].loop_step
  837. mt.scopevars.mql[mqlkey].vars["mqli"] = mql_loopi
  838. mt.scopevars.mql[mqlkey].vars["filei"] = file_loopi
  839. mt.scopevars.mql[mqlkey].vars["diri"] = dir_loopi
  840. mt.scopevars.mql[mqlkey].vars["topi"] = top_loopi
  841. //
  842. mt.scopevars.mql[mqlsn].vars["mqli"] = mql_loopi
  843. mt.scopevars.mql[mqlsn].vars["filei"] = file_loopi
  844. mt.scopevars.mql[mqlsn].vars["diri"] = dir_loopi
  845. mt.scopevars.mql[mqlsn].vars["topi"] = top_loopi
  846. mt.scopevars.Unlock()
  847. mt.BeforeRunAction(basedir, ffpath, mqlkey, mqlsn, staticactions)
  848. formatedmqlstr := mqr.FormatedQueryString
  849. mql := mt.ReplaceLoopSN(formatedmqlstr,
  850. global, topvars, dirvars, filevars, mqlvars,
  851. top_loopi, dir_loopi, file_loopi, mql_loopi,
  852. basedir, ffpath, mqlkey, mqlsn,
  853. )
  854. mqlreplace := rereplace_nbytes.FindAllStringSubmatch(mql, -1)
  855. for _, mqr := range mqlreplace {
  856. if len(mqr) == 2 {
  857. bs := make([]byte, cast.ToInt(mqr[1]))
  858. for i := 0; i < len(bs); i++ {
  859. bs[i] = byte(32 + rand.Intn(91))
  860. }
  861. s := string(bs)
  862. s = strings.ReplaceAll(s, ")", string([]byte{123}))
  863. s = strings.ReplaceAll(s, "'", string([]byte{124}))
  864. s = strings.ReplaceAll(s, "\"", string([]byte{125}))
  865. s = strings.ReplaceAll(s, "\\", string([]byte{126}))
  866. mql = strings.Replace(mql, mqr[0], s, 1)
  867. }
  868. }
  869. for i, sat := range staticactionexprs {
  870. mql = strings.Replace(mql, "["+strconv.Itoa(i)+"]", sat.SourceText, 1)
  871. }
  872. mql = strings.ReplaceAll(mql, "[[]", "[")
  873. mqri := &MQLRequestInstance{
  874. MQLRequest: mqr,
  875. PreparedQueryString: mql,
  876. }
  877. //
  878. ok_chan := make(chan error, 1)
  879. parallel_chan := make(chan bool, 1)
  880. parallel := false
  881. parallelcount := 0
  882. done := false
  883. wg.Add(1)
  884. go func(mqlvars *CurrentVars) {
  885. defer func() {
  886. atomic.AddInt32(&mqlcount, mqlvars.mqlcount)
  887. wg.Done()
  888. }()
  889. ch_ok := make(chan error)
  890. go func() {
  891. for {
  892. select {
  893. case parallelcount = <-ch_parallel_count:
  894. if !done && !parallel {
  895. parallel = true
  896. // 加入并发控制队列
  897. if parallelcount > 0 {
  898. if parallelcount > parallel_queue.Size() {
  899. parallel_queue.Growth(parallelcount)
  900. }
  901. parallel_queue.Push(1)
  902. }
  903. parallel_chan <- true
  904. }
  905. case ok := <-ch_ok:
  906. ok_chan <- ok
  907. if parallel {
  908. if parallelcount > 0 {
  909. // 从并发控制队列中移除
  910. parallel_queue.Pop()
  911. }
  912. } else {
  913. done = true
  914. }
  915. return
  916. }
  917. }
  918. }()
  919. nstaticactionexprs := map[int]*Action{}
  920. for i, sat := range staticactionexprs {
  921. nstaticactionexprs[i] = sat
  922. }
  923. // logger.Info(mqlkey, "第", fmt.Sprint(all_loop_i, ".", dir_loop_i, ".", file_loop_i, ".", curval.mql_i), "次执行开始")
  924. ch_ok <- mt.RunMQL(t, ctx,
  925. global,
  926. topvars,
  927. dirvars,
  928. filevars,
  929. mqlvars,
  930. basedir, ffpath, mqlkey, mqlsn, mqri, staticactions, nstaticactionexprs)
  931. // logger.Info(mqlkey, "第", fmt.Sprint(all_loop_i, ".", dir_loop_i, ".", file_loop_i, ".", curval.mql_i), "次执行结束")
  932. }(mqlvars)
  933. var err error
  934. select {
  935. case err = <-ok_chan: // 非并发,等待完成
  936. // logger.Info(mqlkey, "第", fmt.Sprint(all_loop_i, ".", dir_loop_i, ".", file_loop_i, ".", curval.mql_i), "次顺序执行完成")
  937. case <-parallel_chan: // 并发,执行继续下一次
  938. // logger.Info(mqlkey, "第", fmt.Sprint(all_loop_i, ".", dir_loop_i, ".", file_loop_i, ".", curval.mql_i), "次并发执行继续")
  939. }
  940. if err != nil {
  941. // logger.Error(err)
  942. break
  943. }
  944. }
  945. wg.Wait()
  946. mt.scopevars.RLock()
  947. loop_count := mt.scopevars.mql[mqlkey].loop_count
  948. mt.scopevars.RUnlock()
  949. if loop_count > 1 {
  950. ut := time.Since(st)
  951. sn := fmt.Sprint(topvars.loop_i, ".", dirvars.loop_i, ".", filevars.loop_i)
  952. as := ""
  953. if topvars.sleeptime > 0 {
  954. as = fmt.Sprint(", sleep ", topvars.sleeptime)
  955. }
  956. logger.Info(fmt.Sprint("mql ", mqlkey+"/"+sn, " loop ", loop_i, " times, usetime ", ut, as))
  957. }
  958. })
  959. }
  960. func (mt *MQLTest) getValue(basedir, ffpath, mqlkey, mqlsn string,
  961. staticactions *StaticActions, a any) (ak string, va any) {
  962. switch av := a.(type) {
  963. case *Action:
  964. switch av.Name {
  965. case "<EQ>":
  966. if len(av.Args) >= 2 {
  967. _, x := mt.getValue(basedir, ffpath, mqlkey, mqlsn, staticactions, av.Args[0])
  968. _, y := mt.getValue(basedir, ffpath, mqlkey, mqlsn, staticactions, av.Args[1])
  969. if cast.ToString(x) == cast.ToString(y) {
  970. return "", true
  971. }
  972. }
  973. return "", false
  974. }
  975. case *qstru.Expr:
  976. ak = av.VarName
  977. if ak == "now" {
  978. return "", time.Now()
  979. }
  980. mt.scopevars.Lock()
  981. switch staticactions.Scope {
  982. case "top":
  983. va = mt.scopevars.top.vars[ak]
  984. case "dir":
  985. va = mt.scopevars.dir[basedir].vars[ak]
  986. case "file":
  987. va = mt.scopevars.file[ffpath].vars[ak]
  988. default:
  989. var has bool
  990. va, has = mt.scopevars.mql[mqlsn].vars[ak]
  991. if !has {
  992. va = mt.scopevars.mql[mqlkey].vars[ak]
  993. }
  994. }
  995. mt.scopevars.Unlock()
  996. case string:
  997. va = av
  998. for _, tf := range time_layouts {
  999. t, e := time.Parse(tf, av)
  1000. if e == nil {
  1001. va = t
  1002. break
  1003. }
  1004. }
  1005. default:
  1006. va = a
  1007. }
  1008. return
  1009. }
  1010. func (mt *MQLTest) setValue(basedir, ffpath, mqlkey, mqlsn string,
  1011. staticactions *StaticActions, ak string, va any) {
  1012. mt.scopevars.Lock()
  1013. switch staticactions.Scope {
  1014. case "top":
  1015. mt.scopevars.top.vars[ak] = va
  1016. case "dir":
  1017. mt.scopevars.dir[basedir].vars[ak] = va
  1018. case "file":
  1019. mt.scopevars.file[ffpath].vars[ak] = va
  1020. default:
  1021. _, has := mt.scopevars.mql[mqlsn].vars[ak]
  1022. if has {
  1023. mt.scopevars.mql[mqlsn].vars[ak] = va
  1024. } else {
  1025. mt.scopevars.mql[mqlkey].vars[ak] = va
  1026. }
  1027. }
  1028. mt.scopevars.Unlock()
  1029. }
  1030. func (mt *MQLTest) BeforeRunAction(basedir, ffpath, mqlkey, mqlsn string,
  1031. staticactions *StaticActions,
  1032. ) {
  1033. mt.runAction(basedir, ffpath, mqlkey, mqlsn, staticactions, staticactions.BeforeRunActions...)
  1034. }
  1035. func (mt *MQLTest) runAction(basedir, ffpath, mqlkey, mqlsn string,
  1036. staticactions *StaticActions, actions ...*Action,
  1037. ) {
  1038. for _, act := range actions {
  1039. switch act.Name {
  1040. case "set":
  1041. var a, b any
  1042. var ak string
  1043. if len(act.Args) > 1 {
  1044. a = act.Args[0]
  1045. b = act.Args[1]
  1046. }
  1047. ak, _ = mt.getValue(basedir, ffpath, mqlkey, mqlsn, staticactions, a)
  1048. _, vb := mt.getValue(basedir, ffpath, mqlkey, mqlsn, staticactions, b)
  1049. mt.setValue(basedir, ffpath, mqlkey, mqlsn, staticactions, ak, vb)
  1050. case "add":
  1051. var a, b, va, vb any
  1052. var ak string
  1053. if len(act.Args) > 1 {
  1054. a = act.Args[0]
  1055. b = act.Args[1]
  1056. }
  1057. ak, va = mt.getValue(basedir, ffpath, mqlkey, mqlsn, staticactions, a)
  1058. _, vb = mt.getValue(basedir, ffpath, mqlkey, mqlsn, staticactions, b)
  1059. switch v := va.(type) {
  1060. case time.Time:
  1061. unit := "second"
  1062. if len(act.Args) > 2 {
  1063. unit = cast.ToString(act.Args[2])
  1064. }
  1065. vb := cast.ToInt(vb)
  1066. d := time.Duration(0)
  1067. switch unit {
  1068. case "second", "seconds", "s", "秒":
  1069. d = time.Duration(vb) * time.Second
  1070. case "minute", "minutes", "m", "分":
  1071. d = time.Duration(vb) * time.Minute
  1072. case "hour", "hours", "h", "时":
  1073. d = time.Duration(vb) * time.Hour
  1074. }
  1075. v = v.Add(d)
  1076. mt.setValue(basedir, ffpath, mqlkey, mqlsn, staticactions, ak, v)
  1077. }
  1078. case "mod":
  1079. var a, b, va, vb any
  1080. var ak string
  1081. if len(act.Args) > 1 {
  1082. a = act.Args[0]
  1083. b = act.Args[1]
  1084. }
  1085. ak, va = mt.getValue(basedir, ffpath, mqlkey, mqlsn, staticactions, a)
  1086. _, vb = mt.getValue(basedir, ffpath, mqlkey, mqlsn, staticactions, b)
  1087. av := cast.ToInt(va)
  1088. bv := cast.ToInt(vb)
  1089. v := av % bv
  1090. mt.setValue(basedir, ffpath, mqlkey, mqlsn, staticactions, ak, v)
  1091. case "case":
  1092. var a, b, va any
  1093. if len(act.Args) > 1 {
  1094. a = act.Args[0]
  1095. b = act.Args[1]
  1096. }
  1097. _, va = mt.getValue(basedir, ffpath, mqlkey, mqlsn, staticactions, a)
  1098. if cast.ToBool(va) {
  1099. switch vb := b.(type) {
  1100. case *Action:
  1101. mt.runAction(basedir, ffpath, mqlkey, mqlsn, staticactions, vb)
  1102. }
  1103. }
  1104. }
  1105. }
  1106. }
  1107. func (mt *MQLTest) InitScopeVars(basedir, ffpath, mqlkey string,
  1108. staticactions *StaticActions,
  1109. ) {
  1110. mt.scopevars.Lock()
  1111. if mt.scopevars.mql[mqlkey] == nil {
  1112. mt.scopevars.mql[mqlkey] = &Variables{
  1113. vars: map[string]interface{}{},
  1114. loop_count: 1,
  1115. loop_from: 1,
  1116. loop_step: 1}
  1117. }
  1118. mt.scopevars.Unlock()
  1119. for k, v := range staticactions.Variables {
  1120. mt.scopevars.Lock()
  1121. switch staticactions.Scope {
  1122. case "top":
  1123. mt.scopevars.top.vars[k] = v
  1124. case "dir":
  1125. mt.scopevars.dir[basedir].vars[k] = v
  1126. case "file":
  1127. mt.scopevars.file[ffpath].vars[k] = v
  1128. default:
  1129. mt.scopevars.mql[mqlkey].vars[k] = v
  1130. }
  1131. mt.scopevars.Unlock()
  1132. }
  1133. if staticactions.Timeout != nil {
  1134. mt.scopevars.Lock()
  1135. switch staticactions.Scope {
  1136. case "top":
  1137. mt.scopevars.top.timeout = *staticactions.Timeout
  1138. case "dir":
  1139. mt.scopevars.dir[basedir].timeout = *staticactions.Timeout
  1140. case "file":
  1141. mt.scopevars.file[ffpath].timeout = *staticactions.Timeout
  1142. default:
  1143. mt.scopevars.mql[mqlkey].timeout = *staticactions.Timeout
  1144. }
  1145. mt.scopevars.Unlock()
  1146. }
  1147. if staticactions.Qmeta != nil {
  1148. mt.scopevars.Lock()
  1149. switch staticactions.Scope {
  1150. case "top":
  1151. mt.scopevars.top.qmeta = staticactions.Qmeta
  1152. case "dir":
  1153. mt.scopevars.dir[basedir].qmeta = staticactions.Qmeta
  1154. case "file":
  1155. mt.scopevars.file[ffpath].qmeta = staticactions.Qmeta
  1156. default:
  1157. mt.scopevars.mql[mqlkey].qmeta = staticactions.Qmeta
  1158. }
  1159. mt.scopevars.Unlock()
  1160. }
  1161. if staticactions.LoopCount != nil {
  1162. mt.scopevars.Lock()
  1163. switch staticactions.Scope {
  1164. case "top":
  1165. mt.scopevars.top.loop_count = *staticactions.LoopCount
  1166. case "dir":
  1167. mt.scopevars.dir[basedir].loop_count = *staticactions.LoopCount
  1168. case "file":
  1169. mt.scopevars.file[ffpath].loop_count = *staticactions.LoopCount
  1170. default:
  1171. mt.scopevars.mql[mqlkey].loop_count = *staticactions.LoopCount
  1172. }
  1173. mt.scopevars.Unlock()
  1174. }
  1175. if staticactions.LoopFrom != nil {
  1176. mt.scopevars.Lock()
  1177. switch staticactions.Scope {
  1178. case "top":
  1179. mt.scopevars.top.loop_from = *staticactions.LoopFrom
  1180. case "dir":
  1181. mt.scopevars.dir[basedir].loop_from = *staticactions.LoopFrom
  1182. case "file":
  1183. mt.scopevars.file[ffpath].loop_from = *staticactions.LoopFrom
  1184. default:
  1185. mt.scopevars.mql[mqlkey].loop_from = *staticactions.LoopFrom
  1186. }
  1187. mt.scopevars.Unlock()
  1188. }
  1189. if staticactions.LoopStep != nil {
  1190. mt.scopevars.Lock()
  1191. switch staticactions.Scope {
  1192. case "top":
  1193. mt.scopevars.top.loop_step = *staticactions.LoopStep
  1194. case "dir":
  1195. mt.scopevars.dir[basedir].loop_step = *staticactions.LoopStep
  1196. case "file":
  1197. mt.scopevars.file[ffpath].loop_step = *staticactions.LoopStep
  1198. default:
  1199. mt.scopevars.mql[mqlkey].loop_step = *staticactions.LoopStep
  1200. }
  1201. mt.scopevars.Unlock()
  1202. }
  1203. }
  1204. func (mt *MQLTest) RunMQL(t *testing.T, ctx context.Context,
  1205. global *GlobalVars,
  1206. topvars *CurrentVars,
  1207. dirvars *CurrentVars,
  1208. filevars *CurrentVars,
  1209. mqlvars *CurrentVars,
  1210. basedir, ffpath, mqlkey, mqlsn string, mqri *MQLRequestInstance, staticactions *StaticActions, staticactionexprs map[int]*Action) error {
  1211. mqlstr := mqri.PreparedQueryString
  1212. if staticactions.WaitName != nil {
  1213. global.Lock()
  1214. wgs := []*sync.WaitGroup{}
  1215. if *staticactions.WaitName == "" {
  1216. for _, wg := range global.wg_wait_fork_routine {
  1217. wgs = append(wgs, wg)
  1218. }
  1219. } else {
  1220. wg := global.wg_wait_fork_routine[*staticactions.WaitName]
  1221. if wg != nil {
  1222. wgs = append(wgs, wg)
  1223. }
  1224. }
  1225. global.Unlock()
  1226. for _, wg := range wgs {
  1227. wg.Wait()
  1228. }
  1229. }
  1230. if staticactions.ParallelCount != nil {
  1231. mt.scopevars.Lock()
  1232. switch staticactions.Scope {
  1233. case "top":
  1234. topvars.ch_parallel_count <- *staticactions.ParallelCount
  1235. case "dir":
  1236. dirvars.ch_parallel_count <- *staticactions.ParallelCount
  1237. case "file":
  1238. filevars.ch_parallel_count <- *staticactions.ParallelCount
  1239. default:
  1240. mqlvars.ch_parallel_count <- *staticactions.ParallelCount
  1241. }
  1242. mt.scopevars.Unlock()
  1243. }
  1244. // 重新获取修正后的动作
  1245. actionexprs, e := getActionExprs(mqlstr, staticactionexprs)
  1246. if !assert.Nil(t, e, e) {
  1247. return e
  1248. }
  1249. dynamicactions := actionexprs.DynamicActions()
  1250. if len(dynamicactions.SubscribeArgs) > 0 {
  1251. subscribe(dynamicactions.SubscribeArgs...)
  1252. }
  1253. if len(dynamicactions.UnsubscribeArgs) > 0 {
  1254. unsubscribe(dynamicactions.UnsubscribeArgs...)
  1255. }
  1256. values := []interface{}{}
  1257. if len(dynamicactions.Params) > 0 {
  1258. err := json.Unmarshal([]byte(dynamicactions.Params), &values)
  1259. if err != nil {
  1260. assert.Nil(t, fmt.Sprint("params参数只支持JSON Array,", dynamicactions.Params, mqlstr), err)
  1261. return err
  1262. }
  1263. }
  1264. x := atomic.AddInt32(&global.mqlcount, 1)
  1265. atomic.AddInt32(&topvars.mqlcount, 1)
  1266. atomic.AddInt32(&dirvars.mqlcount, 1)
  1267. atomic.AddInt32(&filevars.mqlcount, 1)
  1268. atomic.AddInt32(&mqlvars.mqlcount, 1)
  1269. err := mt.RunMQLTryDo(t, ctx,
  1270. global,
  1271. topvars,
  1272. dirvars,
  1273. filevars,
  1274. mqlvars,
  1275. basedir, ffpath, mqlkey, mqlsn+"("+strconv.Itoa(int(x))+")", mqri, values, staticactions, actionexprs)
  1276. if err != nil {
  1277. // 执行过程有错,停止继续执行
  1278. return err
  1279. }
  1280. if staticactions.SleepTime != nil {
  1281. global.sleeptime += *staticactions.SleepTime
  1282. topvars.sleeptime += *staticactions.SleepTime
  1283. dirvars.sleeptime += *staticactions.SleepTime
  1284. filevars.sleeptime += *staticactions.SleepTime
  1285. mqlvars.sleeptime += *staticactions.SleepTime
  1286. time.Sleep(*staticactions.SleepTime)
  1287. }
  1288. return nil
  1289. }