19
19
#include "access/xlog.h"
20
20
#include "catalog/namespace.h"
21
21
#include "commands/async.h"
22
+ #include "executor/execParallel.h"
22
23
#include "libpq/libpq.h"
23
24
#include "libpq/pqformat.h"
24
25
#include "libpq/pqmq.h"
60
61
#define PARALLEL_KEY_TRANSACTION_SNAPSHOT UINT64CONST(0xFFFFFFFFFFFF0006)
61
62
#define PARALLEL_KEY_ACTIVE_SNAPSHOT UINT64CONST(0xFFFFFFFFFFFF0007)
62
63
#define PARALLEL_KEY_TRANSACTION_STATE UINT64CONST(0xFFFFFFFFFFFF0008)
63
- #define PARALLEL_KEY_EXTENSION_TRAMPOLINE UINT64CONST(0xFFFFFFFFFFFF0009)
64
+ #define PARALLEL_KEY_ENTRYPOINT UINT64CONST(0xFFFFFFFFFFFF0009)
64
65
65
66
/* Fixed-size parallel state. */
66
67
typedef struct FixedParallelState
@@ -76,7 +77,7 @@ typedef struct FixedParallelState
76
77
pid_t parallel_master_pid ;
77
78
BackendId parallel_master_backend_id ;
78
79
79
- /* Entrypoint for parallel workers. */
80
+ /* Entrypoint for parallel workers (deprecated)! */
80
81
parallel_worker_main_type entrypoint ;
81
82
82
83
/* Mutex protects remaining fields. */
@@ -106,16 +107,36 @@ static FixedParallelState *MyFixedParallelState;
106
107
/* List of active parallel contexts. */
107
108
static dlist_head pcxt_list = DLIST_STATIC_INIT (pcxt_list );
108
109
110
+ /*
111
+ * List of internal parallel worker entry points. We need this for
112
+ * reasons explained in LookupParallelWorkerFunction(), below.
113
+ */
114
+ static const struct
115
+ {
116
+ const char * fn_name ;
117
+ parallel_worker_main_type fn_addr ;
118
+ } InternalParallelWorkers [] =
119
+
120
+ {
121
+ {
122
+ "ParallelQueryMain" , ParallelQueryMain
123
+ }
124
+ };
125
+
109
126
/* Private functions. */
110
127
static void HandleParallelMessage (ParallelContext * pcxt , int i , StringInfo msg );
111
- static void ParallelExtensionTrampoline (dsm_segment * seg , shm_toc * toc );
112
128
static void WaitForParallelWorkersToExit (ParallelContext * pcxt );
129
+ static parallel_worker_main_type LookupParallelWorkerFunction (char * libraryname , char * funcname );
113
130
114
131
115
132
/*
116
133
* Establish a new parallel context. This should be done after entering
117
134
* parallel mode, and (unless there is an error) the context should be
118
135
* destroyed before exiting the current subtransaction.
136
+ *
137
+ * NB: specifying the entrypoint as a function address is unportable.
138
+ * This will go away in Postgres 10, in favor of the API provided by
139
+ * CreateParallelContextForExternalFunction; in the meantime use that.
119
140
*/
120
141
ParallelContext *
121
142
CreateParallelContext (parallel_worker_main_type entrypoint , int nworkers )
@@ -163,9 +184,9 @@ CreateParallelContext(parallel_worker_main_type entrypoint, int nworkers)
163
184
}
164
185
165
186
/*
166
- * Establish a new parallel context that calls a function provided by an
167
- * extension. This works around the fact that the library might get mapped
168
- * at a different address in each backend .
187
+ * Establish a new parallel context that calls a function specified by name.
188
+ * Unlike CreateParallelContext, this is robust against possible differences
189
+ * in address space layout between different processes .
169
190
*/
170
191
ParallelContext *
171
192
CreateParallelContextForExternalFunction (char * library_name ,
@@ -179,7 +200,7 @@ CreateParallelContextForExternalFunction(char *library_name,
179
200
oldcontext = MemoryContextSwitchTo (TopTransactionContext );
180
201
181
202
/* Create the context. */
182
- pcxt = CreateParallelContext (ParallelExtensionTrampoline , nworkers );
203
+ pcxt = CreateParallelContext (NULL , nworkers );
183
204
pcxt -> library_name = pstrdup (library_name );
184
205
pcxt -> function_name = pstrdup (function_name );
185
206
@@ -248,10 +269,9 @@ InitializeParallelDSM(ParallelContext *pcxt)
248
269
pcxt -> nworkers ));
249
270
shm_toc_estimate_keys (& pcxt -> estimator , 1 );
250
271
251
- /* Estimate how much we'll need for extension entrypoint info. */
272
+ /* Estimate how much we'll need for entrypoint info. */
252
273
if (pcxt -> library_name != NULL )
253
274
{
254
- Assert (pcxt -> entrypoint == ParallelExtensionTrampoline );
255
275
Assert (pcxt -> function_name != NULL );
256
276
shm_toc_estimate_chunk (& pcxt -> estimator , strlen (pcxt -> library_name )
257
277
+ strlen (pcxt -> function_name ) + 2 );
@@ -367,7 +387,7 @@ InitializeParallelDSM(ParallelContext *pcxt)
367
387
}
368
388
shm_toc_insert (pcxt -> toc , PARALLEL_KEY_ERROR_QUEUE , error_queue_space );
369
389
370
- /* Serialize extension entrypoint information. */
390
+ /* Serialize entrypoint information. */
371
391
if (pcxt -> library_name != NULL )
372
392
{
373
393
Size lnamelen = strlen (pcxt -> library_name );
@@ -377,7 +397,7 @@ InitializeParallelDSM(ParallelContext *pcxt)
377
397
+ strlen (pcxt -> function_name ) + 2 );
378
398
strcpy (extensionstate , pcxt -> library_name );
379
399
strcpy (extensionstate + lnamelen + 1 , pcxt -> function_name );
380
- shm_toc_insert (pcxt -> toc , PARALLEL_KEY_EXTENSION_TRAMPOLINE ,
400
+ shm_toc_insert (pcxt -> toc , PARALLEL_KEY_ENTRYPOINT ,
381
401
extensionstate );
382
402
}
383
403
}
@@ -669,6 +689,10 @@ DestroyParallelContext(ParallelContext *pcxt)
669
689
}
670
690
671
691
/* Free memory. */
692
+ if (pcxt -> library_name )
693
+ pfree (pcxt -> library_name );
694
+ if (pcxt -> function_name )
695
+ pfree (pcxt -> function_name );
672
696
pfree (pcxt );
673
697
}
674
698
@@ -939,6 +963,8 @@ ParallelWorkerMain(Datum main_arg)
939
963
shm_mq * mq ;
940
964
shm_mq_handle * mqh ;
941
965
char * libraryspace ;
966
+ char * entrypointstate ;
967
+ parallel_worker_main_type entrypt ;
942
968
char * gucspace ;
943
969
char * combocidspace ;
944
970
char * tsnapspace ;
@@ -1038,6 +1064,25 @@ ParallelWorkerMain(Datum main_arg)
1038
1064
Assert (libraryspace != NULL );
1039
1065
RestoreLibraryState (libraryspace );
1040
1066
1067
+ /*
1068
+ * Identify the entry point to be called. In theory this could result in
1069
+ * loading an additional library, though most likely the entry point is in
1070
+ * the core backend or in a library we just loaded.
1071
+ */
1072
+ entrypointstate = shm_toc_lookup (toc , PARALLEL_KEY_ENTRYPOINT );
1073
+ if (entrypointstate != NULL )
1074
+ {
1075
+ char * library_name ;
1076
+ char * function_name ;
1077
+
1078
+ library_name = entrypointstate ;
1079
+ function_name = entrypointstate + strlen (library_name ) + 1 ;
1080
+
1081
+ entrypt = LookupParallelWorkerFunction (library_name , function_name );
1082
+ }
1083
+ else
1084
+ entrypt = fps -> entrypoint ;
1085
+
1041
1086
/* Restore database connection. */
1042
1087
BackgroundWorkerInitializeConnectionByOid (fps -> database_id ,
1043
1088
fps -> authenticated_user_id );
@@ -1101,10 +1146,11 @@ ParallelWorkerMain(Datum main_arg)
1101
1146
/*
1102
1147
* Time to do the real work: invoke the caller-supplied code.
1103
1148
*
1104
- * If you get a crash at this line, see the comments for
1105
- * ParallelExtensionTrampoline.
1149
+ * If you get a crash at this line, try using
1150
+ * CreateParallelContextForExternalFunction instead of
1151
+ * CreateParallelContext.
1106
1152
*/
1107
- fps -> entrypoint (seg , toc );
1153
+ entrypt (seg , toc );
1108
1154
1109
1155
/* Must exit parallel mode to pop active snapshot. */
1110
1156
ExitParallelMode ();
@@ -1119,33 +1165,6 @@ ParallelWorkerMain(Datum main_arg)
1119
1165
pq_putmessage ('X' , NULL , 0 );
1120
1166
}
1121
1167
1122
- /*
1123
- * It's unsafe for the entrypoint invoked by ParallelWorkerMain to be a
1124
- * function living in a dynamically loaded module, because the module might
1125
- * not be loaded in every process, or might be loaded but not at the same
1126
- * address. To work around that problem, CreateParallelContextForExtension()
1127
- * arranges to call this function rather than calling the extension-provided
1128
- * function directly; and this function then looks up the real entrypoint and
1129
- * calls it.
1130
- */
1131
- static void
1132
- ParallelExtensionTrampoline (dsm_segment * seg , shm_toc * toc )
1133
- {
1134
- char * extensionstate ;
1135
- char * library_name ;
1136
- char * function_name ;
1137
- parallel_worker_main_type entrypt ;
1138
-
1139
- extensionstate = shm_toc_lookup (toc , PARALLEL_KEY_EXTENSION_TRAMPOLINE );
1140
- Assert (extensionstate != NULL );
1141
- library_name = extensionstate ;
1142
- function_name = extensionstate + strlen (library_name ) + 1 ;
1143
-
1144
- entrypt = (parallel_worker_main_type )
1145
- load_external_function (library_name , function_name , true, NULL );
1146
- entrypt (seg , toc );
1147
- }
1148
-
1149
1168
/*
1150
1169
* Update shared memory with the ending location of the last WAL record we
1151
1170
* wrote, if it's greater than the value already stored there.
@@ -1161,3 +1180,47 @@ ParallelWorkerReportLastRecEnd(XLogRecPtr last_xlog_end)
1161
1180
fps -> last_xlog_end = last_xlog_end ;
1162
1181
SpinLockRelease (& fps -> mutex );
1163
1182
}
1183
+
1184
+ /*
1185
+ * Look up (and possibly load) a parallel worker entry point function.
1186
+ *
1187
+ * For functions contained in the core code, we use library name "postgres"
1188
+ * and consult the InternalParallelWorkers array. External functions are
1189
+ * looked up, and loaded if necessary, using load_external_function().
1190
+ *
1191
+ * The point of this is to pass function names as strings across process
1192
+ * boundaries. We can't pass actual function addresses because of the
1193
+ * possibility that the function has been loaded at a different address
1194
+ * in a different process. This is obviously a hazard for functions in
1195
+ * loadable libraries, but it can happen even for functions in the core code
1196
+ * on platforms using EXEC_BACKEND (e.g., Windows).
1197
+ *
1198
+ * At some point it might be worthwhile to get rid of InternalParallelWorkers[]
1199
+ * in favor of applying load_external_function() for core functions too;
1200
+ * but that raises portability issues that are not worth addressing now.
1201
+ */
1202
+ static parallel_worker_main_type
1203
+ LookupParallelWorkerFunction (char * libraryname , char * funcname )
1204
+ {
1205
+ /*
1206
+ * If the function is to be loaded from postgres itself, search the
1207
+ * InternalParallelWorkers array.
1208
+ */
1209
+ if (strcmp (libraryname , "postgres" ) == 0 )
1210
+ {
1211
+ int i ;
1212
+
1213
+ for (i = 0 ; i < lengthof (InternalParallelWorkers ); i ++ )
1214
+ {
1215
+ if (strcmp (InternalParallelWorkers [i ].fn_name , funcname ) == 0 )
1216
+ return InternalParallelWorkers [i ].fn_addr ;
1217
+ }
1218
+
1219
+ /* We can only reach this by programming error. */
1220
+ elog (ERROR , "internal function \"%s\" not found" , funcname );
1221
+ }
1222
+
1223
+ /* Otherwise load from external library. */
1224
+ return (parallel_worker_main_type )
1225
+ load_external_function (libraryname , funcname , true, NULL );
1226
+ }
0 commit comments