* Finish parallel execution. We wait for parallel workers to finish, and * accumulate their buffer/WAL usage. */
| 1125 | * accumulate their buffer/WAL usage. |
| 1126 | */ |
| 1127 | void |
| 1128 | ExecParallelFinish(ParallelExecutorInfo *pei) |
| 1129 | { |
| 1130 | int nworkers = pei->pcxt->nworkers_launched; |
| 1131 | int i; |
| 1132 | |
| 1133 | /* Make this be a no-op if called twice in a row. */ |
| 1134 | if (pei->finished) |
| 1135 | return; |
| 1136 | |
| 1137 | /* |
| 1138 | * Detach from tuple queues ASAP, so that any still-active workers will |
| 1139 | * notice that no further results are wanted. |
| 1140 | */ |
| 1141 | if (pei->tqueue != NULL) |
| 1142 | { |
| 1143 | for (i = 0; i < nworkers; i++) |
| 1144 | shm_mq_detach(pei->tqueue[i]); |
| 1145 | pfree(pei->tqueue); |
| 1146 | pei->tqueue = NULL; |
| 1147 | } |
| 1148 | |
| 1149 | /* |
| 1150 | * While we're waiting for the workers to finish, let's get rid of the |
| 1151 | * tuple queue readers. (Any other local cleanup could be done here too.) |
| 1152 | */ |
| 1153 | if (pei->reader != NULL) |
| 1154 | { |
| 1155 | for (i = 0; i < nworkers; i++) |
| 1156 | DestroyTupleQueueReader(pei->reader[i]); |
| 1157 | pfree(pei->reader); |
| 1158 | pei->reader = NULL; |
| 1159 | } |
| 1160 | |
| 1161 | /* Now wait for the workers to finish. */ |
| 1162 | WaitForParallelWorkersToFinish(pei->pcxt); |
| 1163 | |
| 1164 | /* |
| 1165 | * Next, accumulate buffer/WAL usage. (This must wait for the workers to |
| 1166 | * finish, or we might get incomplete data.) |
| 1167 | */ |
| 1168 | for (i = 0; i < nworkers; i++) |
| 1169 | InstrAccumParallelQuery(&pei->buffer_usage[i], &pei->wal_usage[i]); |
| 1170 | |
| 1171 | pei->finished = true; |
| 1172 | } |
| 1173 | |
| 1174 | /* |
| 1175 | * Accumulate instrumentation, and then clean up whatever ParallelExecutorInfo |
no test coverage detected