Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit79cc076

Browse files
committed
PGPRO-2065: add async mode to pgut_execute_parallel()
1 parent55b5656 commit79cc076

File tree

3 files changed

+59
-16
lines changed

3 files changed

+59
-16
lines changed

‎src/backup.c

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1021,8 +1021,15 @@ do_block_validation(void)
10211021
/* TODO write better info message */
10221022
if (check_isok)
10231023
elog(INFO,"Data files are valid");
1024-
else
1025-
elog(ERROR,"Data files are corrupted");
1024+
1025+
if (!check_isok)
1026+
{
1027+
if (thread_interrupted||interrupted)
1028+
elog(ERROR,"Checkdb failed");
1029+
else
1030+
elog(ERROR,"Data files are corrupted");
1031+
}
1032+
10261033

10271034
if (backup_files_list)
10281035
{
@@ -1131,7 +1138,7 @@ do_amcheck(void)
11311138
if (check_isok&& !db_skipped)
11321139
elog(INFO,"Indexes are valid");
11331140

1134-
if (!check_isok)
1141+
if (!check_isok&& !interrupted)
11351142
elog(ERROR,"Some indexes are corrupted");
11361143

11371144
if (db_skipped)
@@ -3398,7 +3405,7 @@ pg_ptrack_get_block(backup_files_arg *arguments,
33983405
res=pgut_execute_parallel(arguments->backup_conn,
33993406
arguments->cancel_conn,
34003407
"SELECT pg_catalog.pg_ptrack_get_block_2($1, $2, $3, $4)",
3401-
4, (constchar**)params, true, false);
3408+
4, (constchar**)params, true, false, false);
34023409

34033410
if (PQnfields(res)!=1)
34043411
{
@@ -3598,7 +3605,7 @@ amcheck_one_index(backup_files_arg *arguments,
35983605

35993606
res=pgut_execute_parallel(arguments->backup_conn,
36003607
arguments->cancel_conn,
3601-
query,2, (constchar**)params, true, true);
3608+
query,2, (constchar**)params, true, true, true);
36023609
}
36033610
else
36043611
{
@@ -3607,7 +3614,7 @@ amcheck_one_index(backup_files_arg *arguments,
36073614

36083615
res=pgut_execute_parallel(arguments->backup_conn,
36093616
arguments->cancel_conn,
3610-
query,1, (constchar**)params, true, true);
3617+
query,1, (constchar**)params, true, true, true);
36113618
}
36123619

36133620
if (PQresultStatus(res)!=PGRES_TUPLES_OK)

‎src/utils/pgut.c

Lines changed: 45 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -359,9 +359,10 @@ PGresult *
359359
pgut_execute_parallel(PGconn*conn,
360360
PGcancel*thread_cancel_conn,constchar*query,
361361
intnParams,constchar**params,
362-
booltext_result,boolok_error)
362+
booltext_result,boolok_error,boolasync)
363363
{
364364
PGresult*res;
365+
intret=0;
365366

366367
if (interrupted&& !in_cleanup)
367368
elog(ERROR,"interrupted");
@@ -387,15 +388,50 @@ pgut_execute_parallel(PGconn* conn,
387388
}
388389

389390
//on_before_exec(conn, thread_cancel_conn);
390-
if (nParams==0)
391-
res=PQexec(conn,query);
391+
if (async)
392+
{
393+
if (nParams==0)
394+
ret=PQsendQuery(conn,query);
395+
else
396+
ret=PQsendQueryParams(conn,query,nParams,NULL,params,NULL,NULL,
397+
/*
398+
* Specify zero to obtain results in text format,
399+
* or one to obtain results in binary format.
400+
*/
401+
(text_result) ?0 :1);
402+
403+
/* wait for processing */
404+
while (true)
405+
{
406+
407+
if (interrupted)
408+
elog(ERROR,"interrupted");
409+
410+
if (!PQconsumeInput(conn))
411+
elog(ERROR,"query failed: %squery was: %s",
412+
PQerrorMessage(conn),query);
413+
414+
/* query is no done */
415+
if (!PQisBusy(conn))
416+
break;
417+
418+
usleep(1000);
419+
}
420+
421+
res=PQgetResult(conn);
422+
}
392423
else
393-
res=PQexecParams(conn,query,nParams,NULL,params,NULL,NULL,
394-
/*
395-
* Specify zero to obtain results in text format,
396-
* or one to obtain results in binary format.
397-
*/
398-
(text_result) ?0 :1);
424+
{
425+
if (nParams==0)
426+
res=PQexec(conn,query);
427+
else
428+
res=PQexecParams(conn,query,nParams,NULL,params,NULL,NULL,
429+
/*
430+
* Specify zero to obtain results in text format,
431+
* or one to obtain results in binary format.
432+
*/
433+
(text_result) ?0 :1);
434+
}
399435
//on_after_exec(thread_cancel_conn);
400436

401437
switch (PQresultStatus(res))

‎src/utils/pgut.h

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -57,7 +57,7 @@ extern PGresult *pgut_execute_extended(PGconn* conn, const char *query, int nPar
5757
constchar**params,booltext_result,boolok_error);
5858
externPGresult*pgut_execute_parallel(PGconn*conn,PGcancel*thread_cancel_conn,
5959
constchar*query,intnParams,
60-
constchar**params,booltext_result,boolok_error);
60+
constchar**params,booltext_result,boolok_error,boolasync);
6161
externboolpgut_send(PGconn*conn,constchar*query,intnParams,constchar**params,intelevel);
6262
externvoidpgut_cancel(PGconn*conn);
6363
externintpgut_wait(intnum,PGconn*connections[],structtimeval*timeout);

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp