!4076 【代码同步】同步3.0.5版本，增加 pg_terminate_active_session_socket 函数

Merge pull request !4076 from zhangxubo/5.0-termsession

src/include/catalog/upgrade_sql/rollback_catalog_maindb/rollback-post_catalog_maindb_92_853.sql

@@ -0,0 +1 @@
+DROP FUNCTION IF EXISTS pg_catalog.pg_terminate_active_session_socket() cascade;

src/include/catalog/upgrade_sql/rollback_catalog_otherdb/rollback-post_catalog_otherdb_92_853.sql

@@ -0,0 +1 @@
+DROP FUNCTION IF EXISTS pg_catalog.pg_terminate_active_session_socket() cascade;

src/include/catalog/upgrade_sql/upgrade_catalog_maindb/upgrade-post_catalog_maindb_92_853.sql

@@ -0,0 +1,5 @@
+set local inplace_upgrade_next_system_object_oids = IUO_PROC, 3148;
+CREATE OR REPLACE FUNCTION pg_catalog.pg_terminate_active_session_socket
+ (IN threadid BIGINT,
+  IN sessionid BIGINT)
+RETURNS bool LANGUAGE INTERNAL NOT FENCED as 'pg_terminate_active_session_socket';

src/include/catalog/upgrade_sql/upgrade_catalog_otherdb/upgrade-post_catalog_otherdb_92_853.sql

@@ -0,0 +1,5 @@
+set local inplace_upgrade_next_system_object_oids = IUO_PROC, 3148;
+CREATE OR REPLACE FUNCTION pg_catalog.pg_terminate_active_session_socket
+ (IN threadid BIGINT,
+  IN sessionid BIGINT)
+RETURNS bool LANGUAGE INTERNAL NOT FENCED as 'pg_terminate_active_session_socket';

src/include/knl/knl_session.h

@@ -200,6 +200,7 @@ typedef struct knl_u_sig_context {
      */
     volatile sig_atomic_t got_pool_reload;
     volatile sig_atomic_t cp_PoolReload;
+    volatile sig_atomic_t got_terminate_sess_socket;
 } knl_u_sig_context;
 
 class AutonomousSession;

src/include/knl/knl_thread.h

@@ -1425,6 +1425,7 @@ typedef struct knl_t_interrupt_context {
 
     volatile bool ignoreBackendSignal; /* ignore signal for threadpool worker */
 
+    volatile bool ignoreSessionBackendSignal; /* ignore signal for u_session */
 } knl_t_interrupt_context;
 
 typedef int64 pg_time_t;

src/include/postgres.h

@@ -959,6 +959,7 @@ extern size_t mmap_threshold;
 void HandlePoolerReload(void);
 void HandleMemoryContextDump(void);
 void HandleExecutorFlag(void);
+void handle_terminate_active_sess_socket();
 
 extern void start_xact_command(void);
 extern void finish_xact_command(void);

src/include/storage/procarray.h

@@ -114,13 +114,14 @@ extern VirtualTransactionId* GetCurrentVirtualXIDs(
 extern VirtualTransactionId *GetConflictingVirtualXIDs(TransactionId limitXmin, Oid dbOid, XLogRecPtr lsn = 0,
                                                        CommitSeqNo limitXminCSN = InvalidCommitSeqNo,
                                                        TransactionId* xminArray = NULL);
-extern ThreadId CancelVirtualTransaction(const VirtualTransactionId& vxid, ProcSignalReason sigmode);
+extern ThreadId CancelVirtualTransaction(const VirtualTransactionId& vxid, ProcSignalReason sigmode,
+    int retry_count);
 
 extern bool MinimumActiveBackends(int min);
 extern int CountDBBackends(Oid database_oid);
 extern int CountDBActiveBackends(Oid database_oid);
 extern int CountSingleNodeActiveBackends(Oid databaseOid, Oid userOid);
-extern void CancelDBBackends(Oid databaseid, ProcSignalReason sigmode, bool conflictPending);
+extern void CancelDBBackends(Oid databaseid, ProcSignalReason sigmode, bool conflictPending, int retry_count);
 extern void CancelSingleNodeBackends(Oid databaseOid, Oid userOid, ProcSignalReason sigmode, bool conflictPending);
 extern int CountUserBackends(Oid roleid);
 extern bool CountOtherDBBackends(Oid databaseId, int* nbackends, int* nprepared);

src/include/storage/procsignal.h

@@ -56,6 +56,9 @@ typedef enum {
     PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK,
     PROCSIG_EXECUTOR_FLAG,
 
+    /* close active session socket */
+    PROCSIG_COMM_CLOSE_ACTIVE_SESSION_SOCKET,
+
     NUM_PROCSIGNALS /* Must be last! */
 } ProcSignalReason;
 

src/include/storage/standby.h

@@ -30,8 +30,8 @@ void ResolveRecoveryConflictWithSnapshotOid(TransactionId latestRemovedXid, Oid
 extern void ResolveRecoveryConflictWithTablespace(Oid tsid);
 extern void ResolveRecoveryConflictWithDatabase(Oid dbid);
 
-extern void ResolveRecoveryConflictWithBufferPin(void);
-extern void SendRecoveryConflictWithBufferPin(ProcSignalReason reason);
+extern void ResolveRecoveryConflictWithBufferPin(int retry_count);
+extern void SendRecoveryConflictWithBufferPin(ProcSignalReason reason, int retry_count = 0);
 extern void CheckRecoveryConflictDeadlock(void);
 
 /*

src/include/threadpool/threadpool_listener.h

@@ -46,9 +46,10 @@ public:
     bool TryFeedWorker(ThreadPoolWorker* worker);
     void AddNewSession(knl_session_context* session);
     void WaitTask();
-    void DelSessionFromEpoll(knl_session_context* session);
+    void DelSessionFromEpoll(knl_session_context* session, bool sub_count);
     void RemoveWorkerFromList(ThreadPoolWorker* worker);
     void AddEpoll(knl_session_context* session);
+    void dispatch_socked_closed_session(knl_session_context* session);
     void SendShutDown();
     void ReaperAllSession();
     void ShutDown() const;

src/include/threadpool/threadpool_sessctl.h

@@ -59,6 +59,7 @@ public:
     bool ValidDBoidAndUseroid(Oid dbOid, Oid userOid, knl_sess_control* ctrl);
     void SigHupHandler();
     void HandlePoolerReload();
+    int terminate_session_socket(ThreadId tid, uint64 session_id);
     void CheckSessionTimeout();
 #ifndef ENABLE_MULTIPLE_NODES
     void CheckIdleInTransactionSessionTimeout();

src/include/utils/builtins.h

@@ -682,6 +682,7 @@ extern Datum pg_cancel_session(PG_FUNCTION_ARGS);
 extern Datum pg_cancel_invalid_query(PG_FUNCTION_ARGS);
 extern Datum pg_terminate_backend(PG_FUNCTION_ARGS);
 extern Datum pg_terminate_session(PG_FUNCTION_ARGS);
+extern Datum pg_terminate_active_session_socket(PG_FUNCTION_ARGS);
 extern Datum pg_reload_conf(PG_FUNCTION_ARGS);
 extern Datum pg_tablespace_databases(PG_FUNCTION_ARGS);
 extern Datum pg_tablespace_location(PG_FUNCTION_ARGS);