isolationtester.c Source File (original) (raw)

19#define PREP_WAITING "isolationtester_waiting"

26{

60#define STEP_NONBLOCK 0x1

61#define STEP_RETRY 0x2

64 int nwaiting, int flags);

66 int flags);

76static void

78{

79 int i;

84}

86int

87 main(int argc, char **argv)

88{

89 const char *conninfo;

90 const char *env_wait;

94 int opt;

95 int i;

97 while ((opt = getopt(argc, argv, "V")) != -1)

98 {

99 switch (opt)

100 {

101 case 'V':

102 puts("isolationtester (PostgreSQL) " PG_VERSION);

103 exit(0);

104 default:

105 fprintf(stderr, "Usage: isolationtester [CONNINFO]\n");

107 }

108 }

109

110

111

112

113

114 setbuf(stdout, NULL);

115 setbuf(stderr, NULL);

116

117

118

119

120

121

122

124 conninfo = argv[optind];

125 else

126 conninfo = "dbname = postgres";

127

128

129

130

131

132 env_wait = getenv("PG_TEST_TIMEOUT_DEFAULT");

133 if (env_wait != NULL)

135

136

139

140

142

143 printf("Parsed test spec with %d sessions\n", testspec->nsessions);

144

145

146

147

148

152

154 {

155 const char *sessionname;

156

157 if (i == 0)

158 sessionname = "control connection";

159 else

161

163

166 {

167 fprintf(stderr, "Connection %d failed: %s",

169 exit(1);

170 }

171

172

173

174

175

176

177

178 if (i != 0)

182 else

185 NULL);

186

187

188

189

190

191

192

194 "SELECT set_config('application_name',\n"

195 " current_setting('application_name') || '/' || $1,\n"

196 " false)",

197 1, NULL,

198 &sessionname,

199 NULL, NULL, 0);

201 {

202 fprintf(stderr, "setting of application name failed: %s",

204 exit(1);

205 }

206

207

210 }

211

212

213

214

215

216

217

218

219

222 "SELECT pg_catalog.pg_isolation_test_session_is_blocked($1, '{");

223

228

231 {

232 fprintf(stderr, "prepare of lock wait query failed: %s",

234 exit(1);

235 }

238

239

240

241

242

244

245 return 0;

246}

247

248

249

250

251static void

253{

254 int nallsteps;

255 Step **allsteps;

256 int i,

257 j,

258 k;

259

260

261 nallsteps = 0;

264

265 allsteps = pg_malloc(nallsteps * sizeof(Step *));

266

267 k = 0;

269 {

272 }

273

275

276

277 for (i = 1; i < nallsteps; i++)

278 {

279 if (strcmp(allsteps[i - 1]->name,

280 allsteps[i]->name) == 0)

281 {

282 fprintf(stderr, "duplicate step name: %s\n",

283 allsteps[i]->name);

284 exit(1);

285 }

286 }

287

288

290 {

292

293 for (j = 0; j < session->nsteps; j++)

295 }

296

297

298

299

300

302 {

304

306 {

309 allsteps,

310 nallsteps,

311 sizeof(Step *),

313

314 if (this == NULL)

315 {

316 fprintf(stderr, "undefined step \"%s\" specified in permutation\n",

317 pstep->name);

318 exit(1);

319 }

320 pstep->step = *this;

321

322

324 }

325

326

327

328

329

330

331

333 {

335

336 for (k = 0; k < pstep->nblockers; k++)

337 {

339 int n;

340

342 continue;

343

344 blocker->step = NULL;

345 for (n = 0; n < p->nsteps; n++)

346 {

348

349 if (strcmp(otherp->name, blocker->stepname) == 0)

350 {

351 blocker->step = otherp->step;

352 break;

353 }

354 }

355 if (blocker->step == NULL)

356 {

357 fprintf(stderr, "undefined blocking step \"%s\" referenced in permutation step \"%s\"\n",

359 exit(1);

360 }

361

363 {

364 fprintf(stderr, "permutation step \"%s\" cannot block on its own session\n",

365 pstep->name);

366 exit(1);

367 }

368 }

369 }

370 }

371

372

373

374

375

376

378 {

379 for (i = 0; i < nallsteps; i++)

380 {

381 if (!allsteps[i]->used)

382 fprintf(stderr, "unused step name: %s\n", allsteps[i]->name);

383 }

384 }

385

386 free(allsteps);

387}

388

389

390

391

392

393static void

395{

398 else

400}

401

402

403

404

405static void

407{

408 int nsteps;

409 int i;

412 int *piles;

413

414

415 nsteps = 0;

418

419

422 for (i = 0; i < nsteps; i++)

423 stepptrs[i] = steps + i;

424

425

426

427

428

429

430

431

432

433

436 piles[i] = 0;

437

439

441 free(stepptrs);

443}

444

445static void

448{

449 int i;

450 bool found = false;

451

453 {

454

456 {

458

459

460

461

462

463

464 steps[nsteps]->name = newstep->name;

465 steps[nsteps]->step = newstep;

466

467 piles[i]++;

468

470

471 piles[i]--;

472

473 found = true;

474 }

475 }

476

477

478 if (!found)

480}

481

482

483

484

485static void

487{

488 int i;

489

491 {

493

495 }

496}

497

498static int

500{

503

504 return strcmp(stepa->name, stepb->name);

505}

506

507static int

509{

510 char *stepname = (char *) a;

512

513 return strcmp(stepname, step->name);

514}

515

516

517

518

519static void

521{

523 int i;

524 int nwaiting = 0;

526

528

529 printf("\nstarting permutation:");

530 for (i = 0; i < nsteps; i++)

533

534

536 {

539 {

541 }

543 {

545 exit(1);

546 }

548 }

549

550

552 {

554 {

557 {

559 }

561 {

562 fprintf(stderr, "setup of session %s failed: %s",

563 conns[i + 1].sessionname,

565 exit(1);

566 }

568 }

569 }

570

571

572 for (i = 0; i < nsteps; i++)

573 {

578 bool mustwait;

579 int j;

580

581

582

583

584

586 {

588

590

592 {

594

595

596

597

598

599

601 {

602

603 int w;

604

605 for (w = 0; w < nwaiting; w++)

606 {

607 if (oldstep == waiting[w])

608 break;

609 }

610 if (w >= nwaiting)

611 abort();

612 if (w + 1 < nwaiting)

615 nwaiting--;

616 }

617

618

619

620

621

622

623

624

627

628

629

630

631

632

633

634

635

637 {

638 struct timeval current_time;

640

646 {

647 fprintf(stderr, "step %s timed out after %d seconds\n",

650 fprintf(stderr, "active steps are:");

652 {

654

658 }

660 exit(1);

661 }

662 }

663 }

664 }

665

666

668 {

669 fprintf(stdout, "failed to send query for step %s: %s\n",

671 exit(1);

672 }

673

674

676

677

679 {

681

685 }

686

687

689

690

693

694

695 if (mustwait)

696 waiting[nwaiting++] = pstep;

697 }

698

699

701 if (nwaiting != 0)

702 {

703 fprintf(stderr, "failed to complete permutation due to mutually-blocking steps\n");

704 exit(1);

705 }

706

707

709 {

711 {

714 {

716 }

718 {

719 fprintf(stderr, "teardown of session %s failed: %s",

720 conns[i + 1].sessionname,

722

723 }

725 }

726 }

727

728

730 {

733 {

735 }

737 {

738 fprintf(stderr, "teardown failed: %s",

740

741 }

743 }

744

746}

747

748

749

750

751

752

753

754static int

756 int nwaiting, int flags)

757{

758 int old_nwaiting;

759 bool have_blocker;

760

761 do

762 {

763 int w = 0;

764

765

767

768

769 old_nwaiting = nwaiting;

770 have_blocker = false;

771

772

773 while (w < nwaiting)

774 {

776 {

777

778 if (waiting[w]->nblockers > 0)

779 have_blocker = true;

780 w++;

781 }

782 else

783 {

784

785 if (w + 1 < nwaiting)

788 nwaiting--;

789 }

790 }

791

792

793

794

795

796

797

798

799

800 } while (have_blocker && (nwaiting < old_nwaiting || any_new_notice));

801 return nwaiting;

802}

803

804

805

806

807

808

809

810

811

812

813

814

815

816

817static bool

819{

823 fd_set read_set;

825 struct timeval timeout;

827 int ret;

830 bool canceled = false;

831

832

833

834

835

836

838 {

839 int i;

840

842 {

844

846 {

847 printf("step %s: %s <waiting ...>\n",

849 return true;

850 }

851 }

852 }

853

854 if (sock < 0)

855 {

857 exit(1);

858 }

859

861 FD_ZERO(&read_set);

862

864 {

865 FD_SET(sock, &read_set);

866 timeout.tv_sec = 0;

867 timeout.tv_usec = 10000;

868

869 ret = select(sock + 1, &read_set, NULL, NULL, &timeout);

870 if (ret < 0)

871 {

872 if (errno == EINTR)

873 continue;

874 fprintf(stderr, "select failed: %m\n");

875 exit(1);

876 }

877 else if (ret == 0)

878 {

879 struct timeval current_time;

881

882

884 {

886

889 NULL, NULL, 0);

892 {

893 fprintf(stderr, "lock wait query failed: %s",

895 exit(1);

896 }

899

900 if (waiting)

901 {

902

903

904

905

906

907

908

909

910

912 {

913 fprintf(stderr, "PQconsumeInput failed: %s\n",

915 exit(1);

916 }

918 break;

919

920

921

922

923

925 printf("step %s: %s <waiting ...>\n",

927 return true;

928 }

929

930 }

931

932

937

938

939

940

941

942

943

944

945

946

948 {

950

952 {

953

954

955

956

957 printf("isolationtester: canceling step %s after %d seconds\n",

959 canceled = true;

960 }

961 else

964 }

965

966

967

968

969

970

971

972

974 {

975 fprintf(stderr, "step %s timed out after %d seconds\n",

977 exit(1);

978 }

979 }

981 {

982 fprintf(stderr, "PQconsumeInput failed: %s\n",

984 exit(1);

985 }

986 }

987

988

989

990

991

993 {

995 printf("step %s: %s <waiting ...>\n",

997 return true;

998 }

999

1000

1002 printf("step %s: <... completed>\n", step->name);

1003 else

1005

1007 {

1009 {

1012 break;

1015 break;

1017

1018

1019

1020

1021

1022

1023 {

1028

1029 if (sev && msg)

1030 printf("%s: %s\n", sev, msg);

1031 else

1033 }

1034 break;

1035 default:

1036 printf("unexpected result status: %s\n",

1038 }

1040 }

1041

1042

1045 {

1046

1047 const char *sendername = NULL;

1048 char pidstring[32];

1049 int i;

1050

1052 {

1054 {

1056 break;

1057 }

1058 }

1059 if (sendername == NULL)

1060 {

1061

1062 snprintf(pidstring, sizeof(pidstring), "PID %d", notify->be_pid);

1063 sendername = pidstring;

1064 }

1065 printf("%s: NOTIFY \"%s\" with payload \"%s\" from %s\n",

1067 notify->relname, notify->extra, sendername);

1070 }

1071

1072

1074

1075 return false;

1076}

1077

1078

1079static bool

1081{

1082 int i;

1083

1085 {

1088

1090 {

1092

1093 break;

1095

1099 return true;

1100 break;

1102

1105 return true;

1106 break;

1107 }

1108 }

1109 return false;

1110}

1111

1112static void

1114{

1116

1117 memset(&popt, 0, sizeof(popt));

1119 popt.align = true;

1122}

1123

1124

1125static void

1127{

1129

1130

1132

1135}

1136

1137

1138static void

1140{

1141

1142}

#define fprintf(file, fmt, msg)

PGcancelConn * PQcancelCreate(PGconn *conn)

int PQcancelBlocking(PGcancelConn *cancelConn)

void PQcancelFinish(PGcancelConn *cancelConn)

char * PQcancelErrorMessage(const PGcancelConn *cancelConn)

PGconn * PQconnectdb(const char *conninfo)

ConnStatusType PQstatus(const PGconn *conn)

void PQfinish(PGconn *conn)

int PQbackendPID(const PGconn *conn)

PQnoticeProcessor PQsetNoticeProcessor(PGconn *conn, PQnoticeProcessor proc, void *arg)

char * PQerrorMessage(const PGconn *conn)

int PQsocket(const PGconn *conn)

void PQfreemem(void *ptr)

PGresult * PQexecParams(PGconn *conn, const char *command, int nParams, const Oid *paramTypes, const char *const *paramValues, const int *paramLengths, const int *paramFormats, int resultFormat)

PGresult * PQexecPrepared(PGconn *conn, const char *stmtName, int nParams, const char *const *paramValues, const int *paramLengths, const int *paramFormats, int resultFormat)

char * PQgetvalue(const PGresult *res, int tup_num, int field_num)

PGresult * PQgetResult(PGconn *conn)

ExecStatusType PQresultStatus(const PGresult *res)

void PQclear(PGresult *res)

int PQntuples(const PGresult *res)

PGresult * PQprepare(PGconn *conn, const char *stmtName, const char *query, int nParams, const Oid *paramTypes)

char * PQresultErrorMessage(const PGresult *res)

int PQconsumeInput(PGconn *conn)

char * PQresultErrorField(const PGresult *res, int fieldcode)

int PQsendQuery(PGconn *conn, const char *query)

char * PQresStatus(ExecStatusType status)

int PQisBusy(PGconn *conn)

PGresult * PQexec(PGconn *conn, const char *query)

PGnotify * PQnotifies(PGconn *conn)

void PQprint(FILE *fout, const PGresult *res, const PQprintOpt *po)

void * pg_malloc(size_t size)

void * pg_malloc0(size_t size)

static void printResultSet(PGresult *res)

static int step_bsearch_cmp(const void *a, const void *b)

static IsoConnInfo * conns

static int step_qsort_cmp(const void *a, const void *b)

static void run_testspec(TestSpec *testspec)

static void run_named_permutations(TestSpec *testspec)

static int try_complete_steps(TestSpec *testspec, PermutationStep **waiting, int nwaiting, int flags)

int main(int argc, char **argv)

static void run_all_permutations_recurse(TestSpec *testspec, int *piles, int nsteps, PermutationStep **steps)

static bool try_complete_step(TestSpec *testspec, PermutationStep *pstep, int flags)

static void run_all_permutations(TestSpec *testspec)

static void blackholeNoticeProcessor(void *arg, const char *message)

struct IsoConnInfo IsoConnInfo

static void check_testspec(TestSpec *testspec)

static void run_permutation(TestSpec *testspec, int nsteps, PermutationStep **steps)

static void isotesterNoticeProcessor(void *arg, const char *message)

static int64 max_step_wait

static bool step_has_blocker(PermutationStep *pstep)

static void disconnect_atexit(void)

static bool any_new_notice

int getopt(int nargc, char *const *nargv, const char *ostr)

#define qsort(a, b, c, d)

#define PG_DIAG_MESSAGE_PRIMARY

void initPQExpBuffer(PQExpBuffer str)

void appendPQExpBuffer(PQExpBuffer str, const char *fmt,...)

void appendPQExpBufferStr(PQExpBuffer str, const char *data)

void termPQExpBuffer(PQExpBuffer str)

char * psprintf(const char *fmt,...)

PermutationStep * active_step

const char * backend_pid_str

PermutationStepBlockerType blocktype

PermutationStepBlocker ** blockers

Permutation ** permutations

static volatile sig_atomic_t waiting

#define select(n, r, w, e, timeout)

int gettimeofday(struct timeval *tp, void *tzp)