Added the -a option to crash_gen tool to automatically start/stop TDengine service

tests/pytest/crash_gen.py

@@ -1629,7 +1629,7 @@ class MyLoggingAdapter(logging.LoggerAdapter):
         # return '[%s] %s' % (self.extra['connid'], msg), kwargs
 
 class SvcManager:    
-    MAX_QUEUE_SIZE = 30
+    MAX_QUEUE_SIZE = 10000
 
     def __init__(self):
         print("Starting service manager")
@@ -1667,7 +1667,7 @@ class SvcManager:
                 break
 
             # queue.put(line)
-        print("No more output (most likely) from IO thread managing TDengine service") # meaning sub process must have died
+        print("\nNo more output (most likely) from IO thread managing TDengine service") # meaning sub process must have died
         out.close()
 
     def _doMenu(self):
@@ -1700,10 +1700,10 @@ class SvcManager:
         if choice == "1" :
             self.sigHandlerResume() # TODO: can the sub-process be blocked due to us not reading from queue?
         elif choice == "2" :
-            self._stopTaosService()
+            self.stopTaosService()
         elif choice == "3" :
-            self._stopTaosService()
-            self._startTaosService()
+            self.stopTaosService()
+            self.startTaosService()
         else:
             raise RuntimeError("Invalid menu choice: {}".format(choice))
 
@@ -1714,7 +1714,7 @@ class SvcManager:
             return
 
         self.status = MainExec.STATUS_STOPPING # immediately set our status    
-        self._stopTaosService()  
+        self.stopTaosService()  
         print("INT signal handler returning...")  
 
     def sigHandlerResume(self) :
@@ -1728,12 +1728,17 @@ class SvcManager:
         else :
             print("Joining empty thread, doing nothing")
     
+    TD_READY_MSG = "TDengine is initialized successfully"
     def _procIpcBatch(self):
         # Process all the output generated by the underlying sub process, managed by IO thread
         while True :
             try:  
                 line = self.ipcQueue.get_nowait() # getting output at fast speed   
-                print("_o", end="", flush=True)                 
+                print("_o", end="", flush=True)   
+                if self.status == MainExec.STATUS_STARTING : # we are starting, let's see if we have started
+                    if line.find(self.TD_READY_MSG) != -1 : # found
+                        self.status = MainExec.STATUS_RUNNING
+                              
             except Empty:
                 # time.sleep(2.3) # wait only if there's no output
                 # no more output
@@ -1743,13 +1748,13 @@ class SvcManager:
 
     def _procIpcAll(self):
         while True :
-            print("<<", end="", flush=True)
+            print("<", end="", flush=True)
             self._procIpcBatch() # process one batch
 
             # check if the ioThread is still running
             if (not self.ioThread) or (not self.ioThread.is_alive()):
                 print("IO Thread (with subprocess) has ended, main thread now exiting...")
-                self._stopTaosService()
+                self.stopTaosService()
                 self._procIpcBatch() # one more batch
                 return # TODO: maybe one last batch?
 
@@ -1759,10 +1764,10 @@ class SvcManager:
                 self._procIpcBatch() # one more batch
                 return
 
-            print(">>", end="", flush=True)
+            print(">", end="", flush=True)
             time.sleep(0.5)
 
-    def _startTaosService(self):
+    def startTaosService(self):
         ON_POSIX = 'posix' in sys.builtin_module_names
         svcCmd = ['../../build/build/bin/taosd', '-c', '../../build/test/cfg']
         # svcCmd = ['vmstat', '1']
@@ -1783,9 +1788,19 @@ class SvcManager:
         self.ioThread.start()
 
         self.shouldStop = False # don't let the main loop stop
-        self.status = MainExec.STATUS_RUNNING
-
-    def _stopTaosService(self):
+        self.status = MainExec.STATUS_STARTING
+
+        # wait for service to start
+        for i in range(0, 10) :
+            time.sleep(1.0)
+            self._procIpcBatch() # pump messages
+            print("_zz_", end="", flush=True)
+            if self.status == MainExec.STATUS_RUNNING :
+                print("TDengine service READY to process requests")
+                return # now we've started
+        raise RuntimeError("TDengine service did not start successfully") # TODO: handle this better?
+
+    def stopTaosService(self):
         # can be called from both main thread or signal handler
         print("Terminating TDengine service running as the sub process...")
         # Linux will send Control-C generated SIGINT to the TDengine process already, ref: https://unix.stackexchange.com/questions/176235/fork-and-how-signals-are-delivered-to-processes
@@ -1804,7 +1819,7 @@ class SvcManager:
             except subprocess.TimeoutExpired as err:
                 print("Time out waiting for TDengine service process to exit")
             else:
-                print("Sub process has ended")
+                print("TDengine service process terminated successfully from SIG_INT")
                 self.subProcess = None
 
         if self.subProcess and (not self.subProcess.poll()):
@@ -1814,7 +1829,7 @@ class SvcManager:
         self.joinIoThread()
 
     def run(self):
-        self._startTaosService()
+        self.startTaosService()
 
         # proc = subprocess.Popen(['echo', '"to stdout"'], 
         #                 stdout=subprocess.PIPE,
@@ -1878,6 +1893,10 @@ class ClientManager:
         self._printLastNumbers()
 
     def run(self):
+        if gConfig.auto_start_service :
+            svcMgr = SvcManager()
+            svcMgr.startTaosService()
+
         self._printLastNumbers()
 
         dbManager = DbManager() # Regular function
@@ -1889,6 +1908,8 @@ class ClientManager:
         # print("exec stats: {}".format(self.tc.getExecStats()))
         # print("TC failed = {}".format(self.tc.isFailed()))
         self.conclude()
+        if gConfig.auto_start_service :
+            svcMgr.stopTaosService()
         # print("TC failed (2) = {}".format(self.tc.isFailed()))
         return 1 if self.tc.isFailed() else 0 # Linux return code: ref https://shapeshed.com/unix-exit-codes/
 
@@ -1898,8 +1919,9 @@ class ClientManager:
 
 
 class MainExec:
-    STATUS_RUNNING = 1
-    STATUS_STOPPING = 2
+    STATUS_STARTING = 1
+    STATUS_RUNNING = 2
+    STATUS_STOPPING = 3
     # STATUS_STOPPED = 3 # Not used yet
 
     @classmethod
@@ -1968,6 +1990,8 @@ def main():
 
             '''))
 
+    parser.add_argument('-a', '--auto-start-service', action='store_true',                        
+                        help='Automatically start/stop the TDengine service (default: false)')
     parser.add_argument('-c', '--connector-type', action='store', default='native', type=str,
                         help='Connector type to use: native, rest, or mixed (default: 10)')
     parser.add_argument('-d', '--debug', action='store_true',