21
21
#include "miscadmin.h"
22
22
#include "storage/latch.h"
23
23
#include "utils/hsearch.h"
24
+ #include "utils/inval.h"
24
25
#include "utils/memutils.h"
25
26
#include "utils/syscache.h"
26
27
@@ -47,11 +48,15 @@ typedef struct ConnCacheEntry
47
48
{
48
49
ConnCacheKey key ; /* hash key (must be first) */
49
50
PGconn * conn ; /* connection to foreign server, or NULL */
51
+ /* Remaining fields are invalid when conn is NULL: */
50
52
int xact_depth ; /* 0 = no xact open, 1 = main xact open, 2 =
51
53
* one level of subxact open, etc */
52
54
bool have_prep_stmt ; /* have we prepared any stmts in this xact? */
53
55
bool have_error ; /* have any subxacts aborted in this xact? */
54
56
bool changing_xact_state ; /* xact state change in process */
57
+ bool invalidated ; /* true if reconnect is pending */
58
+ uint32 server_hashvalue ; /* hash value of foreign server OID */
59
+ uint32 mapping_hashvalue ; /* hash value of user mapping OID */
55
60
} ConnCacheEntry ;
56
61
57
62
/*
@@ -68,6 +73,7 @@ static bool xact_got_connection = false;
68
73
69
74
/* prototypes of private functions */
70
75
static PGconn * connect_pg_server (ForeignServer * server , UserMapping * user );
76
+ static void disconnect_pg_server (ConnCacheEntry * entry );
71
77
static void check_conn_params (const char * * keywords , const char * * values );
72
78
static void configure_remote_session (PGconn * conn );
73
79
static void do_sql_command (PGconn * conn , const char * sql );
@@ -77,6 +83,7 @@ static void pgfdw_subxact_callback(SubXactEvent event,
77
83
SubTransactionId mySubid ,
78
84
SubTransactionId parentSubid ,
79
85
void * arg );
86
+ static void pgfdw_inval_callback (Datum arg , int cacheid , uint32 hashvalue );
80
87
static void pgfdw_reject_incomplete_xact_state_change (ConnCacheEntry * entry );
81
88
static bool pgfdw_cancel_query (PGconn * conn );
82
89
static bool pgfdw_exec_cleanup_query (PGconn * conn , const char * query ,
@@ -94,13 +101,6 @@ static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime,
94
101
* will_prep_stmt must be true if caller intends to create any prepared
95
102
* statements. Since those don't go away automatically at transaction end
96
103
* (not even on error), we need this flag to cue manual cleanup.
97
- *
98
- * XXX Note that caching connections theoretically requires a mechanism to
99
- * detect change of FDW objects to invalidate already established connections.
100
- * We could manage that by watching for invalidation events on the relevant
101
- * syscaches. For the moment, though, it's not clear that this would really
102
- * be useful and not mere pedantry. We could not flush any active connections
103
- * mid-transaction anyway.
104
104
*/
105
105
PGconn *
106
106
GetConnection (UserMapping * user , bool will_prep_stmt )
@@ -129,6 +129,10 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
129
129
*/
130
130
RegisterXactCallback (pgfdw_xact_callback , NULL );
131
131
RegisterSubXactCallback (pgfdw_subxact_callback , NULL );
132
+ CacheRegisterSyscacheCallback (FOREIGNSERVEROID ,
133
+ pgfdw_inval_callback , (Datum ) 0 );
134
+ CacheRegisterSyscacheCallback (USERMAPPINGOID ,
135
+ pgfdw_inval_callback , (Datum ) 0 );
132
136
}
133
137
134
138
/* Set flag that we did GetConnection during the current transaction */
@@ -143,17 +147,27 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
143
147
entry = hash_search (ConnectionHash , & key , HASH_ENTER , & found );
144
148
if (!found )
145
149
{
146
- /* initialize new hashtable entry (key is already filled in) */
150
+ /*
151
+ * We need only clear "conn" here; remaining fields will be filled
152
+ * later when "conn" is set.
153
+ */
147
154
entry -> conn = NULL ;
148
- entry -> xact_depth = 0 ;
149
- entry -> have_prep_stmt = false;
150
- entry -> have_error = false;
151
- entry -> changing_xact_state = false;
152
155
}
153
156
154
157
/* Reject further use of connections which failed abort cleanup. */
155
158
pgfdw_reject_incomplete_xact_state_change (entry );
156
159
160
+ /*
161
+ * If the connection needs to be remade due to invalidation, disconnect as
162
+ * soon as we're out of all transactions.
163
+ */
164
+ if (entry -> conn != NULL && entry -> invalidated && entry -> xact_depth == 0 )
165
+ {
166
+ elog (DEBUG3 , "closing connection %p for option changes to take effect" ,
167
+ entry -> conn );
168
+ disconnect_pg_server (entry );
169
+ }
170
+
157
171
/*
158
172
* We don't check the health of cached connection here, because it would
159
173
* require some overhead. Broken connection will be detected when the
@@ -163,15 +177,26 @@ GetConnection(UserMapping *user, bool will_prep_stmt)
163
177
/*
164
178
* If cache entry doesn't have a connection, we have to establish a new
165
179
* connection. (If connect_pg_server throws an error, the cache entry
166
- * will be left in a valid empty state.)
180
+ * will remain in a valid empty state, ie conn == NULL .)
167
181
*/
168
182
if (entry -> conn == NULL )
169
183
{
170
184
ForeignServer * server = GetForeignServer (user -> serverid );
171
185
172
- entry -> xact_depth = 0 ; /* just to be sure */
186
+ /* Reset all transient state fields, to be sure all are clean */
187
+ entry -> xact_depth = 0 ;
173
188
entry -> have_prep_stmt = false;
174
189
entry -> have_error = false;
190
+ entry -> changing_xact_state = false;
191
+ entry -> invalidated = false;
192
+ entry -> server_hashvalue =
193
+ GetSysCacheHashValue1 (FOREIGNSERVEROID ,
194
+ ObjectIdGetDatum (server -> serverid ));
195
+ entry -> mapping_hashvalue =
196
+ GetSysCacheHashValue1 (USERMAPPINGOID ,
197
+ ObjectIdGetDatum (user -> umid ));
198
+
199
+ /* Now try to make the connection */
175
200
entry -> conn = connect_pg_server (server , user );
176
201
177
202
elog (DEBUG3 , "new postgres_fdw connection %p for server \"%s\" (user mapping oid %u, userid %u)" ,
@@ -285,6 +310,19 @@ connect_pg_server(ForeignServer *server, UserMapping *user)
285
310
return conn ;
286
311
}
287
312
313
+ /*
314
+ * Disconnect any open connection for a connection cache entry.
315
+ */
316
+ static void
317
+ disconnect_pg_server (ConnCacheEntry * entry )
318
+ {
319
+ if (entry -> conn != NULL )
320
+ {
321
+ PQfinish (entry -> conn );
322
+ entry -> conn = NULL ;
323
+ }
324
+ }
325
+
288
326
/*
289
327
* For non-superusers, insist that the connstr specify a password. This
290
328
* prevents a password from being picked up from .pgpass, a service file,
@@ -786,9 +824,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
786
824
entry -> changing_xact_state )
787
825
{
788
826
elog (DEBUG3 , "discarding connection %p" , entry -> conn );
789
- PQfinish (entry -> conn );
790
- entry -> conn = NULL ;
791
- entry -> changing_xact_state = false;
827
+ disconnect_pg_server (entry );
792
828
}
793
829
}
794
830
@@ -905,6 +941,47 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
905
941
}
906
942
}
907
943
944
+ /*
945
+ * Connection invalidation callback function
946
+ *
947
+ * After a change to a pg_foreign_server or pg_user_mapping catalog entry,
948
+ * mark connections depending on that entry as needing to be remade.
949
+ * We can't immediately destroy them, since they might be in the midst of
950
+ * a transaction, but we'll remake them at the next opportunity.
951
+ *
952
+ * Although most cache invalidation callbacks blow away all the related stuff
953
+ * regardless of the given hashvalue, connections are expensive enough that
954
+ * it's worth trying to avoid that.
955
+ *
956
+ * NB: We could avoid unnecessary disconnection more strictly by examining
957
+ * individual option values, but it seems too much effort for the gain.
958
+ */
959
+ static void
960
+ pgfdw_inval_callback (Datum arg , int cacheid , uint32 hashvalue )
961
+ {
962
+ HASH_SEQ_STATUS scan ;
963
+ ConnCacheEntry * entry ;
964
+
965
+ Assert (cacheid == FOREIGNSERVEROID || cacheid == USERMAPPINGOID );
966
+
967
+ /* ConnectionHash must exist already, if we're registered */
968
+ hash_seq_init (& scan , ConnectionHash );
969
+ while ((entry = (ConnCacheEntry * ) hash_seq_search (& scan )))
970
+ {
971
+ /* Ignore invalid entries */
972
+ if (entry -> conn == NULL )
973
+ continue ;
974
+
975
+ /* hashvalue == 0 means a cache reset, must clear all state */
976
+ if (hashvalue == 0 ||
977
+ (cacheid == FOREIGNSERVEROID &&
978
+ entry -> server_hashvalue == hashvalue ) ||
979
+ (cacheid == USERMAPPINGOID &&
980
+ entry -> mapping_hashvalue == hashvalue ))
981
+ entry -> invalidated = true;
982
+ }
983
+ }
984
+
908
985
/*
909
986
* Raise an error if the given connection cache entry is marked as being
910
987
* in the middle of an xact state change. This should be called at which no
@@ -922,9 +999,14 @@ pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
922
999
Form_pg_user_mapping umform ;
923
1000
ForeignServer * server ;
924
1001
925
- if (!entry -> changing_xact_state )
1002
+ /* nothing to do for inactive entries and entries of sane state */
1003
+ if (entry -> conn == NULL || !entry -> changing_xact_state )
926
1004
return ;
927
1005
1006
+ /* make sure this entry is inactive */
1007
+ disconnect_pg_server (entry );
1008
+
1009
+ /* find server name to be shown in the message below */
928
1010
tup = SearchSysCache1 (USERMAPPINGOID ,
929
1011
ObjectIdGetDatum (entry -> key ));
930
1012
if (!HeapTupleIsValid (tup ))
0 commit comments