| 1433 | } |
| 1434 | |
| 1435 | void |
| 1436 | HttpTunnel::consumer_reenable(HttpTunnelConsumer *c) |
| 1437 | { |
| 1438 | HttpTunnelProducer *p = c->producer; |
| 1439 | |
| 1440 | if (p && p->alive) { |
| 1441 | // Only do flow control if enabled and the producer is an external |
| 1442 | // source. Otherwise disable by making the backlog zero. Because |
| 1443 | // the backlog short cuts quit when the value is equal (or |
| 1444 | // greater) to the target, we use strict comparison only for |
| 1445 | // checking low water, otherwise the flow control can stall out. |
| 1446 | uint64_t backlog = (flow_state.enabled_p && p->is_source()) ? p->backlog(flow_state.high_water) : 0; |
| 1447 | HttpTunnelProducer *srcp = p->flow_control_source; |
| 1448 | |
| 1449 | if (backlog >= flow_state.high_water) { |
| 1450 | if (dbg_ctl_http_tunnel.on()) { |
| 1451 | Dbg(dbg_ctl_http_tunnel, "[%" PRId64 "] Throttle %p %" PRId64 " / %" PRId64, sm->sm_id, p, backlog, p->backlog()); |
| 1452 | } |
| 1453 | p->throttle(); // p becomes srcp for future calls to this method |
| 1454 | } else { |
| 1455 | if (srcp && srcp->alive && c->is_sink()) { |
| 1456 | // Check if backlog is below low water - note we need to check |
| 1457 | // against the source producer, not necessarily the producer |
| 1458 | // for this consumer. We don't have to recompute the backlog |
| 1459 | // if they are the same because we know low water <= high |
| 1460 | // water so the value is sufficiently accurate. |
| 1461 | if (srcp != p) { |
| 1462 | backlog = srcp->backlog(flow_state.low_water); |
| 1463 | } |
| 1464 | if (backlog < flow_state.low_water) { |
| 1465 | if (dbg_ctl_http_tunnel.on()) { |
| 1466 | Dbg(dbg_ctl_http_tunnel, "[%" PRId64 "] Unthrottle %p %" PRId64 " / %" PRId64, sm->sm_id, p, backlog, p->backlog()); |
| 1467 | } |
| 1468 | srcp->unthrottle(); |
| 1469 | if (srcp->read_vio) { |
| 1470 | srcp->read_vio->reenable(); |
| 1471 | } |
| 1472 | // Kick source producer to get flow ... well, flowing. |
| 1473 | this->producer_handler(VC_EVENT_READ_READY, srcp); |
| 1474 | } else { |
| 1475 | // We can stall for small thresholds on network sinks because this event happens |
| 1476 | // before the actual socket write. So we trap for the buffer becoming empty to |
| 1477 | // make sure we get an event to unthrottle after the write. |
| 1478 | if (HT_HTTP_CLIENT == c->vc_type) { |
| 1479 | NetVConnection *netvc = dynamic_cast<NetVConnection *>(c->write_vio->vc_server); |
| 1480 | if (netvc) { // really, this should always be true. |
| 1481 | netvc->trapWriteBufferEmpty(); |
| 1482 | } |
| 1483 | } |
| 1484 | } |
| 1485 | } |
| 1486 | if (p->read_vio) { |
| 1487 | p->read_vio->reenable(); |
| 1488 | } |
| 1489 | } |
| 1490 | } |
| 1491 | } |
| 1492 |
no test coverage detected