bank_client.py: introduce graceful termination of clients on stop() command.

Manually cherry-picked from

commit 14a172476b16619be66e01ed0c6cc2aca5d77892
Author: Andrey V. Lepikhov <a.lepikhov@postgrespro.ru>
Date:   Wed Feb 26 19:30:08 2020 +0500

    1. test_referee.py: provide time gap without loading for the Multimaster to clean old referee
    decision.
    2. bank_client.py: introduce graceful termination of clients on stop() command.
    Now on the stop() call the MtmClient process waits for all open transactions,
    and closes database connections.
    3. Increase value of TEST_MAX_RECOVERY_TIME up to 1200 s.
    4. Timestamps added for some test  messages to simplify log analysis.

(the rest of changes are already here or not needed)

Author: danolivo

tests/lib/bank_client.py

@@ -12,6 +12,7 @@
 import multiprocessing
 import logging
 import re
+import signal
 import pprint
 import uuid
 import traceback
@@ -285,9 +286,19 @@ def status(self):
                         serialized_aggs[node_id][aggname] = total_agg.as_dict()
 
                 yield from self.child_pipe.coro_send(serialized_aggs)
+            elif msg == 'exit':
+                break
             else:
                 print('evloop: unknown message')
 
+        # End of work. Wait for tasks and stop event loop.
+        self.running = False # mark for other coroutines to exit
+        tasks = [t for t in asyncio.Task.all_tasks() if t is not asyncio.Task.current_task()]
+        yield from asyncio.gather(*tasks)
+
+        self.loop.stop()
+
+
     @asyncio.coroutine
     def exec_tx(self, tx_block, node_id, aggname_prefix, conn_i):
         conn_name = "node_{}_{}_{}".format(node_id + 1, aggname_prefix, conn_i)
@@ -359,7 +370,12 @@ def exec_tx(self, tx_block, node_id, aggname_prefix, conn_i):
                 # back to event loop and block it
                 yield from asyncio.sleep(0.5)
 
-        print("We've count to infinity!")
+        # Close connection during client termination
+        # XXX: I tried to be polite and close the cursor beforehand, but it
+        # sometimes gives 'close cannot be used while an asynchronous query
+        # is underway'.
+        if conn:
+            conn.close()
 
     @asyncio.coroutine
     def transfer_tx(self, conn, cur, agg, conn_i):
@@ -407,14 +423,19 @@ def run(self):
 
         for i, _ in enumerate(self.dsns):
             for j in range(1):
-                asyncio.ensure_future(self.exec_tx(self.transfer_tx, i, 'transfer', j))
+                self.loop.create_task(self.exec_tx(self.transfer_tx, i, 'transfer', j))
+            self.loop.create_task(self.exec_tx(self.total_tx, i, 'sumtotal', 0))
             asyncio.ensure_future(self.exec_tx(self.total_tx, i, 'sumtotal', 0))
             for j in range(10):
-                asyncio.ensure_future(self.exec_tx(self.insert_tx, i, 'inserter', j))
+                self.loop.create_task(self.exec_tx(self.insert_tx, i, 'inserter', j))
 
-        asyncio.ensure_future(self.status())
+        self.loop.create_task(self.status())
 
-        self.loop.run_forever()
+        try:
+            self.running = True
+            self.loop.run_forever()
+        finally:
+             self.loop.close()
 
     def bgrun(self):
         print('Starting evloop in different process')
@@ -446,10 +467,9 @@ def clean_aggregates(self):
         print('aggregates cleaned')
 
     def stop(self):
-        self.running = False
-        self.evloop_process.terminate()
+        print('stopping client')
+        self.parent_pipe.send('exit')
         self.evloop_process.join()
-        time.sleep(3)
         print('client stopped')
 
     @classmethod

tests/lib/test_helper.py

@@ -77,7 +77,7 @@ def awaitOnline(dsn):
                 cur.execute("select 1")
                 one = int(cur.fetchone()[0])
                 cur.close()
-                print("{} is online!".format(dsn))
+                print("{} {} is online!".format(datetime.datetime.utcnow(), dsn))
                 return
             except Exception as e:
                 print('{} waiting for {} to get online: {}'.format(datetime.datetime.utcnow(), dsn, str(e)))