PostgreSQL Source Code git master
Functions | Variables
parallel.c File Reference
#include "postgres_fe.h"
#include <sys/wait.h>
#include "pg_upgrade.h"
Include dependency graph for parallel.c:

Go to the source code of this file.

Functions

void  parallel_exec_prog (const char *log_file, const char *opt_log_file, const char *fmt,...)
 
void  parallel_transfer_all_new_dbs (DbInfoArr *old_db_arr, DbInfoArr *new_db_arr, char *old_pgdata, char *new_pgdata, char *old_tablespace, char *new_tablespace)
 
bool  reap_child (bool wait_for_child)
 

Variables

static int  parallel_jobs
 

Function Documentation

◆ parallel_exec_prog()

void parallel_exec_prog ( const char *  log_file,
const char *  opt_log_file,
const char *  fmt,
  ... 
)

Definition at line 63 of file parallel.c.

65{
66 va_list args;
67 char cmd[MAX_STRING];
68
69#ifndef WIN32
70 pid_t child;
71#else
72 HANDLE child;
73 exec_thread_arg *new_arg;
74#endif
75
76 va_start(args, fmt);
77 vsnprintf(cmd, sizeof(cmd), fmt, args);
78 va_end(args);
79
80 if (user_opts.jobs <= 1)
81 /* exit_on_error must be true to allow jobs */
82 exec_prog(log_file, opt_log_file, true, true, "%s", cmd);
83 else
84 {
85 /* parallel */
86#ifdef WIN32
87 if (thread_handles == NULL)
88 thread_handles = pg_malloc(user_opts.jobs * sizeof(HANDLE));
89
90 if (exec_thread_args == NULL)
91 {
92 int i;
93
94 exec_thread_args = pg_malloc(user_opts.jobs * sizeof(exec_thread_arg *));
95
96 /*
97 * For safety and performance, we keep the args allocated during
98 * the entire life of the process, and we don't free the args in a
99 * thread different from the one that allocated it.
100 */
101 for (i = 0; i < user_opts.jobs; i++)
102 exec_thread_args[i] = pg_malloc0(sizeof(exec_thread_arg));
103 }
104
105 cur_thread_args = (void **) exec_thread_args;
106#endif
107 /* harvest any dead children */
108 while (reap_child(false) == true)
109 ;
110
111 /* must we wait for a dead child? */
113 reap_child(true);
114
115 /* set this before we start the job */
117
118 /* Ensure stdio state is quiesced before forking */
119 fflush(NULL);
120
121#ifndef WIN32
122 child = fork();
123 if (child == 0)
124 /* use _exit to skip atexit() functions */
125 _exit(!exec_prog(log_file, opt_log_file, true, true, "%s", cmd));
126 else if (child < 0)
127 /* fork failed */
128 pg_fatal("could not create worker process: %m");
129#else
130 /* empty array element are always at the end */
131 new_arg = exec_thread_args[parallel_jobs - 1];
132
133 /* Can only pass one pointer into the function, so use a struct */
134 pg_free(new_arg->log_file);
135 new_arg->log_file = pg_strdup(log_file);
136 pg_free(new_arg->opt_log_file);
137 new_arg->opt_log_file = opt_log_file ? pg_strdup(opt_log_file) : NULL;
138 pg_free(new_arg->cmd);
139 new_arg->cmd = pg_strdup(cmd);
140
141 child = (HANDLE) _beginthreadex(NULL, 0, (void *) win32_exec_prog,
142 new_arg, 0, NULL);
143 if (child == 0)
144 pg_fatal("could not create worker thread: %m");
145
146 thread_handles[parallel_jobs - 1] = child;
147#endif
148 }
149}
bool exec_prog(const char *log_filename, const char *opt_log_file, bool report_error, bool exit_on_error, const char *fmt,...)
Definition: exec.c:85
static int parallel_jobs
Definition: parallel.c:19
bool reap_child(bool wait_for_child)
Definition: parallel.c:281
void * pg_malloc(size_t size)
Definition: fe_memutils.c:47
char * pg_strdup(const char *in)
Definition: fe_memutils.c:85
void * pg_malloc0(size_t size)
Definition: fe_memutils.c:53
void pg_free(void *ptr)
Definition: fe_memutils.c:105
i
int i
Definition: isn.c:77
args
#define pg_fatal(...)
static char * log_file
Definition: pg_ctl.c:87
#define MAX_STRING
Definition: pg_upgrade.h:22
#define vsnprintf
Definition: port.h:238
UserOpts user_opts
Definition: option.c:30
int jobs
Definition: pg_upgrade.h:338

References generate_unaccent_rules::args, exec_prog(), i, UserOpts::jobs, log_file, MAX_STRING, parallel_jobs, pg_fatal, pg_free(), pg_malloc(), pg_malloc0(), pg_strdup(), reap_child(), user_opts, and vsnprintf.

Referenced by create_new_objects(), and generate_old_dump().

◆ parallel_transfer_all_new_dbs()

void parallel_transfer_all_new_dbs ( DbInfoArr *  old_db_arr,
DbInfoArr *  new_db_arr,
char *  old_pgdata,
char *  new_pgdata,
char *  old_tablespace,
char *  new_tablespace 
)

Definition at line 173 of file parallel.c.

176{
177#ifndef WIN32
178 pid_t child;
179#else
180 HANDLE child;
181 transfer_thread_arg *new_arg;
182#endif
183
184 if (user_opts.jobs <= 1)
185 transfer_all_new_dbs(old_db_arr, new_db_arr, old_pgdata, new_pgdata, NULL, NULL);
186 else
187 {
188 /* parallel */
189#ifdef WIN32
190 if (thread_handles == NULL)
191 thread_handles = pg_malloc(user_opts.jobs * sizeof(HANDLE));
192
193 if (transfer_thread_args == NULL)
194 {
195 int i;
196
197 transfer_thread_args = pg_malloc(user_opts.jobs * sizeof(transfer_thread_arg *));
198
199 /*
200 * For safety and performance, we keep the args allocated during
201 * the entire life of the process, and we don't free the args in a
202 * thread different from the one that allocated it.
203 */
204 for (i = 0; i < user_opts.jobs; i++)
205 transfer_thread_args[i] = pg_malloc0(sizeof(transfer_thread_arg));
206 }
207
208 cur_thread_args = (void **) transfer_thread_args;
209#endif
210 /* harvest any dead children */
211 while (reap_child(false) == true)
212 ;
213
214 /* must we wait for a dead child? */
216 reap_child(true);
217
218 /* set this before we start the job */
220
221 /* Ensure stdio state is quiesced before forking */
222 fflush(NULL);
223
224#ifndef WIN32
225 child = fork();
226 if (child == 0)
227 {
228 transfer_all_new_dbs(old_db_arr, new_db_arr, old_pgdata, new_pgdata,
229 old_tablespace, new_tablespace);
230 /* if we take another exit path, it will be non-zero */
231 /* use _exit to skip atexit() functions */
232 _exit(0);
233 }
234 else if (child < 0)
235 /* fork failed */
236 pg_fatal("could not create worker process: %m");
237#else
238 /* empty array element are always at the end */
239 new_arg = transfer_thread_args[parallel_jobs - 1];
240
241 /* Can only pass one pointer into the function, so use a struct */
242 new_arg->old_db_arr = old_db_arr;
243 new_arg->new_db_arr = new_db_arr;
244 pg_free(new_arg->old_pgdata);
245 new_arg->old_pgdata = pg_strdup(old_pgdata);
246 pg_free(new_arg->new_pgdata);
247 new_arg->new_pgdata = pg_strdup(new_pgdata);
248 pg_free(new_arg->old_tablespace);
249 new_arg->old_tablespace = old_tablespace ? pg_strdup(old_tablespace) : NULL;
250 new_arg->new_tablespace = new_tablespace ? pg_strdup(new_tablespace) : NULL;
251
252 child = (HANDLE) _beginthreadex(NULL, 0, (void *) win32_transfer_all_new_dbs,
253 new_arg, 0, NULL);
254 if (child == 0)
255 pg_fatal("could not create worker thread: %m");
256
257 thread_handles[parallel_jobs - 1] = child;
258#endif
259 }
260}
void transfer_all_new_dbs(DbInfoArr *old_db_arr, DbInfoArr *new_db_arr, char *old_pgdata, char *new_pgdata, char *old_tablespace, char *new_tablespace)
Definition: relfilenumber.c:172

References i, UserOpts::jobs, parallel_jobs, pg_fatal, pg_free(), pg_malloc(), pg_malloc0(), pg_strdup(), reap_child(), transfer_all_new_dbs(), and user_opts.

Referenced by transfer_all_new_tablespaces().

◆ reap_child()

bool reap_child ( bool  wait_for_child )

Definition at line 281 of file parallel.c.

282{
283#ifndef WIN32
284 int work_status;
285 pid_t child;
286#else
287 int thread_num;
288 DWORD res;
289#endif
290
291 if (user_opts.jobs <= 1 || parallel_jobs == 0)
292 return false;
293
294#ifndef WIN32
295 child = waitpid(-1, &work_status, wait_for_child ? 0 : WNOHANG);
296 if (child == (pid_t) -1)
297 pg_fatal("%s() failed: %m", "waitpid");
298 if (child == 0)
299 return false; /* no children, or no dead children */
300 if (work_status != 0)
301 pg_fatal("child process exited abnormally: status %d", work_status);
302#else
303 /* wait for one to finish */
304 thread_num = WaitForMultipleObjects(parallel_jobs, thread_handles,
305 false, wait_for_child ? INFINITE : 0);
306
307 if (thread_num == WAIT_TIMEOUT || thread_num == WAIT_FAILED)
308 return false;
309
310 /* compute thread index in active_threads */
311 thread_num -= WAIT_OBJECT_0;
312
313 /* get the result */
314 GetExitCodeThread(thread_handles[thread_num], &res);
315 if (res != 0)
316 pg_fatal("child worker exited abnormally: %m");
317
318 /* dispose of handle to stop leaks */
319 CloseHandle(thread_handles[thread_num]);
320
321 /* Move last slot into dead child's position */
322 if (thread_num != parallel_jobs - 1)
323 {
324 void *tmp_args;
325
326 thread_handles[thread_num] = thread_handles[parallel_jobs - 1];
327
328 /*
329 * Move last active thread arg struct into the now-dead slot, and the
330 * now-dead slot to the end for reuse by the next thread. Though the
331 * thread struct is in use by another thread, we can safely swap the
332 * struct pointers within the array.
333 */
334 tmp_args = cur_thread_args[thread_num];
335 cur_thread_args[thread_num] = cur_thread_args[parallel_jobs - 1];
336 cur_thread_args[parallel_jobs - 1] = tmp_args;
337 }
338#endif
339
340 /* do this after job has been removed */
342
343 return true;
344}

References UserOpts::jobs, parallel_jobs, pg_fatal, and user_opts.

Referenced by create_new_objects(), generate_old_dump(), parallel_exec_prog(), parallel_transfer_all_new_dbs(), and transfer_all_new_tablespaces().

Variable Documentation

◆ parallel_jobs

int parallel_jobs
static

Definition at line 19 of file parallel.c.

Referenced by parallel_exec_prog(), parallel_transfer_all_new_dbs(), and reap_child().

AltStyle によって変換されたページ (->オリジナル) /