Queue the result (if any) for the one worker thread. Become the worker if possible. NULL result is a wakeup or a topup. topup signifies that the working thread is the caller looking for additional events.
| 2256 | // NULL result is a wakeup or a topup. |
| 2257 | // topup signifies that the working thread is the caller looking for additional events. |
| 2258 | static pn_event_batch_t *pconnection_process(pconnection_t *pc, iocp_result_t *result, bool topup) { |
| 2259 | bool first = true; |
| 2260 | bool waking = false; |
| 2261 | bool tick_required = false; |
| 2262 | bool open = false; |
| 2263 | |
| 2264 | while (true) { |
| 2265 | { |
| 2266 | csguard g(&pc->context.cslock); |
| 2267 | if (first) { |
| 2268 | first = false; |
| 2269 | if (result) |
| 2270 | pc->completion_queue->push(result); |
| 2271 | else if (!topup) |
| 2272 | wake_complete(&pc->context); |
| 2273 | if (!topup) { |
| 2274 | if (pc->context.working) |
| 2275 | return NULL; |
| 2276 | pc->context.working = true; |
| 2277 | } |
| 2278 | open = pc->started && !pc->connecting && !pc->context.closing; |
| 2279 | } |
| 2280 | else { |
| 2281 | // Just re-acquired lock after processing IO and engine work |
| 2282 | if (pconnection_has_event(pc)) |
| 2283 | return &pc->batch; |
| 2284 | |
| 2285 | if (!pconnection_work_pending(pc)) { |
| 2286 | pc->context.working = false; |
| 2287 | if (pn_connection_driver_finished(&pc->driver)) { |
| 2288 | pconnection_begin_close(pc); |
| 2289 | } |
| 2290 | if (pc->context.closing && pconnection_can_free(pc)) { |
| 2291 | if (pconnection_cleanup(pc)) { |
| 2292 | g.release(); |
| 2293 | pconnection_final_free(pc); |
| 2294 | return NULL; |
| 2295 | } // else disconnect logic has the free obligation |
| 2296 | } |
| 2297 | return NULL; |
| 2298 | } |
| 2299 | } |
| 2300 | |
| 2301 | if (pc->queued_disconnect) { // From pn_proactor_disconnect() |
| 2302 | pc->queued_disconnect = false; |
| 2303 | if (!pc->context.closing) { |
| 2304 | if (pc->disconnect_condition) { |
| 2305 | pn_condition_copy(pn_transport_condition(pc->driver.transport), pc->disconnect_condition); |
| 2306 | } |
| 2307 | pn_connection_driver_close(&pc->driver); |
| 2308 | } |
| 2309 | } |
| 2310 | |
| 2311 | assert(pc->work_queue->empty()); |
| 2312 | if (pc->completion_queue->size()) |
| 2313 | std::swap(pc->work_queue, pc->completion_queue); |
| 2314 | |
| 2315 | if (pc->wake_count) { |
no test coverage detected