granicus.if.org Git - apache/blob

2 *

3 * Licensed under the Apache License, Version 2.0 (the "License");

4 * you may not use this file except in compliance with the License.

5 * You may obtain a copy of the License at

6 *

7 * http://www.apache.org/licenses/LICENSE-2.0

8

9 * Unless required by applicable law or agreed to in writing, software

10 * distributed under the License is distributed on an "AS IS" BASIS,

11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

12 * See the License for the specific language governing permissions and

13 * limitations under the License.

14 */

15

16 #include <stddef.h>

17 #include <apr_strings.h>

18 #include <nghttp2/nghttp2.h>

19

20 #include <mpm_common.h>

21 #include <httpd.h>

22 #include <mod_proxy.h>

23

24 #include "mod_http2.h"

25 #include "h2.h"

26 #include "h2_util.h"

27 #include "h2_proxy_session.h"

28

29 APLOG_USE_MODULE(proxy_http2);

30

31 typedef struct h2_proxy_stream {

32 int id;

33 apr_pool_t *pool;

34 h2_proxy_session *session;

35

36 const char *url;

37 request_rec *r;

38 h2_request *req;

39

40 h2_stream_state_t state;

41 unsigned int suspended : 1;

42 unsigned int data_received : 1;

43

44 apr_bucket_brigade *input;

45 apr_bucket_brigade *output;

46

47 apr_table_t *saves;

48 } h2_proxy_stream;

49

50

51 static void dispatch_event(h2_proxy_session *session, h2_proxys_event_t ev,

52 int arg, const char *msg);

53

54

55 static apr_status_t proxy_session_pre_close(void *theconn)

56 {

57 proxy_conn_rec *p_conn = (proxy_conn_rec *)theconn;

58 h2_proxy_session *session = p_conn->data;

59

60 if (session && session->ngh2) {

61 ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,

62 "proxy_session(%s): pool cleanup, state=%d, streams=%d",

63 session->id, session->state,

64 (int)h2_ihash_count(session->streams));

65 session->aborted = 1;

66 dispatch_event(session, H2_PROXYS_EV_PRE_CLOSE, 0, NULL);

67 nghttp2_session_del(session->ngh2);

68 session->ngh2 = NULL;

69 p_conn->data = NULL;

70 }

71 return APR_SUCCESS;

72 }

73

74 static int proxy_pass_brigade(apr_bucket_alloc_t *bucket_alloc,

75 proxy_conn_rec *p_conn,

76 conn_rec *origin, apr_bucket_brigade *bb,

77 int flush)

78 {

79 apr_status_t status;

80 apr_off_t transferred;

81

82 if (flush) {

83 apr_bucket *e = apr_bucket_flush_create(bucket_alloc);

84 APR_BRIGADE_INSERT_TAIL(bb, e);

85 }

86 apr_brigade_length(bb, 0, &transferred);

87 if (transferred != -1)

88 p_conn->worker->s->transferred += transferred;

89 status = ap_pass_brigade(origin->output_filters, bb);

90 /* Cleanup the brigade now to avoid buckets lifetime

91 * issues in case of error returned below. */

92 apr_brigade_cleanup(bb);

93 if (status != APR_SUCCESS) {

94 ap_log_cerror(APLOG_MARK, APLOG_ERR, status, origin, APLOGNO(03357)

95 "pass output failed to %pI (%s)",

96 p_conn->addr, p_conn->hostname);

97 }

98 return status;

99 }

100

101 static ssize_t raw_send(nghttp2_session *ngh2, const uint8_t *data,

102 size_t length, int flags, void *user_data)

103 {

104 h2_proxy_session *session = user_data;

105 apr_bucket *b;

106 apr_status_t status;

107 int flush = 1;

108

109 if (data) {

110 b = apr_bucket_transient_create((const char*)data, length,

111 session->c->bucket_alloc);

112 APR_BRIGADE_INSERT_TAIL(session->output, b);

113 }

114

115 status = proxy_pass_brigade(session->c->bucket_alloc,

116 session->p_conn, session->c,

117 session->output, flush);

118 ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, session->c,

119 "h2_proxy_sesssion(%s): raw_send %d bytes, flush=%d",

120 session->id, (int)length, flush);

121 if (status != APR_SUCCESS) {

122 return NGHTTP2_ERR_CALLBACK_FAILURE;

123 }

124 return length;

125 }

126

127 static int on_frame_recv(nghttp2_session *ngh2, const nghttp2_frame *frame,

128 void *user_data)

129 {

130 h2_proxy_session *session = user_data;

131 int n;

132

133 if (APLOGcdebug(session->c)) {

134 char buffer[256];

135

136 h2_util_frame_print(frame, buffer, sizeof(buffer)/sizeof(buffer[0]));

137 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03341)

138 "h2_proxy_session(%s): recv FRAME[%s]",

139 session->id, buffer);

140 }

141

142 switch (frame->hd.type) {

143 case NGHTTP2_HEADERS:

144 break;

145 case NGHTTP2_PUSH_PROMISE:

146 break;

147 case NGHTTP2_SETTINGS:

148 if (frame->settings.niv > 0) {

149 n = nghttp2_session_get_remote_settings(ngh2, NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS);

150 if (n > 0) {

151 session->remote_max_concurrent = n;

152 }

153 }

154 break;

155 case NGHTTP2_GOAWAY:

156 /* we expect the remote server to tell us the highest stream id

157 * that it has started processing. */

158 session->last_stream_id = frame->goaway.last_stream_id;

159 dispatch_event(session, H2_PROXYS_EV_REMOTE_GOAWAY, 0, NULL);

160 if (APLOGcinfo(session->c)) {

161 char buffer[256];

162

163 h2_util_frame_print(frame, buffer, sizeof(buffer)/sizeof(buffer[0]));

164 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03342)

165 "h2_proxy_session(%s): recv FRAME[%s]",

166 session->id, buffer);

167 }

168 break;

169 default:

170 break;

171 }

172 return 0;

173 }

174

175 static int before_frame_send(nghttp2_session *ngh2,

176 const nghttp2_frame *frame, void *user_data)

177 {

178 h2_proxy_session *session = user_data;

179 if (APLOGcdebug(session->c)) {

180 char buffer[256];

181

182 h2_util_frame_print(frame, buffer, sizeof(buffer)/sizeof(buffer[0]));

183 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03343)

184 "h2_proxy_session(%s): sent FRAME[%s]",

185 session->id, buffer);

186 }

187 return 0;

188 }

189

190 static int add_header(void *table, const char *n, const char *v)

191 {

192 apr_table_addn(table, n, v);

193 return 1;

194 }

195

196 static void process_proxy_header(request_rec *r, const char *n, const char *v)

197 {

198 static const struct {

199 const char *name;

200 ap_proxy_header_reverse_map_fn func;

201 } transform_hdrs[] = {

202 { "Location", ap_proxy_location_reverse_map },

203 { "Content-Location", ap_proxy_location_reverse_map },

204 { "URI", ap_proxy_location_reverse_map },

205 { "Destination", ap_proxy_location_reverse_map },

206 { "Set-Cookie", ap_proxy_cookie_reverse_map },

207 { NULL, NULL }

208 };

209 proxy_dir_conf *dconf;

210 int i;

211

212 for (i = 0; transform_hdrs[i].name; ++i) {

213 if (!ap_casecmpstr(transform_hdrs[i].name, n)) {

214 dconf = ap_get_module_config(r->per_dir_config, &proxy_module);

215 apr_table_add(r->headers_out, n,

216 (*transform_hdrs[i].func)(r, dconf, v));

217 return;

218 }

219 }

220 apr_table_add(r->headers_out, n, v);

221 }

222

223 static apr_status_t h2_proxy_stream_add_header_out(h2_proxy_stream *stream,

224 const char *n, apr_size_t nlen,

225 const char *v, apr_size_t vlen)

226 {

227 if (n[0] == ':') {

228 if (!stream->data_received && !strncmp(":status", n, nlen)) {

229 char *s = apr_pstrndup(stream->r->pool, v, vlen);

230

231 apr_table_setn(stream->r->notes, "proxy-status", s);

232 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c,

233 "h2_proxy_stream(%s-%d): got status %s",

234 stream->session->id, stream->id, s);

235 stream->r->status = (int)apr_atoi64(s);

236 if (stream->r->status <= 0) {

237 stream->r->status = 500;

238 return APR_EGENERAL;

239 }

240 }

241 return APR_SUCCESS;

242 }

243

244 if (!h2_proxy_res_ignore_header(n, nlen)) {

245 char *hname, *hvalue;

246

247 hname = apr_pstrndup(stream->pool, n, nlen);

248 h2_util_camel_case_header(hname, nlen);

249 hvalue = apr_pstrndup(stream->pool, v, vlen);

250

251 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, stream->session->c,

252 "h2_proxy_stream(%s-%d): got header %s: %s",

253 stream->session->id, stream->id, hname, hvalue);

254 process_proxy_header(stream->r, hname, hvalue);

255 }

256 return APR_SUCCESS;

257 }

258

259 static int log_header(void *ctx, const char *key, const char *value)

260 {

261 h2_proxy_stream *stream = ctx;

262 ap_log_rerror(APLOG_MARK, APLOG_TRACE2, 0, stream->r,

263 "h2_proxy_stream(%s-%d), header_out %s: %s",

264 stream->session->id, stream->id, key, value);

265 return 1;

266 }

267

268 static void h2_proxy_stream_end_headers_out(h2_proxy_stream *stream)

269 {

270 h2_proxy_session *session = stream->session;

271 request_rec *r = stream->r;

272 apr_pool_t *p = r->pool;

273

274 /* Now, add in the cookies from the response to the ones already saved */

275 apr_table_do(add_header, stream->saves, r->headers_out, "Set-Cookie", NULL);

276

277 /* and now load 'em all in */

278 if (!apr_is_empty_table(stream->saves)) {

279 apr_table_unset(r->headers_out, "Set-Cookie");

280 r->headers_out = apr_table_overlay(p, r->headers_out, stream->saves);

281 }

282

283 /* handle Via header in response */

284 if (session->conf->viaopt != via_off

285 && session->conf->viaopt != via_block) {

286 const char *server_name = ap_get_server_name(stream->r);

287 apr_port_t port = ap_get_server_port(stream->r);

288 char portstr[32];

289

290 /* If USE_CANONICAL_NAME_OFF was configured for the proxy virtual host,

291 * then the server name returned by ap_get_server_name() is the

292 * origin server name (which does make too much sense with Via: headers)

293 * so we use the proxy vhost's name instead.

294 */

295 if (server_name == stream->r->hostname) {

296 server_name = stream->r->server->server_hostname;

297 }

298 if (ap_is_default_port(port, stream->r)) {

299 portstr[0] = '\0';

300 }

301 else {

302 apr_snprintf(portstr, sizeof(portstr), ":%d", port);

303 }

304

305 /* create a "Via:" response header entry and merge it */

306 apr_table_addn(r->headers_out, "Via",

307 (session->conf->viaopt == via_full)

308 ? apr_psprintf(p, "%d.%d %s%s (%s)",

309 HTTP_VERSION_MAJOR(r->proto_num),

310 HTTP_VERSION_MINOR(r->proto_num),

311 server_name, portstr,

312 AP_SERVER_BASEVERSION)

313 : apr_psprintf(p, "%d.%d %s%s",

314 HTTP_VERSION_MAJOR(r->proto_num),

315 HTTP_VERSION_MINOR(r->proto_num),

316 server_name, portstr)

317 );

318 }

319

320 if (APLOGrtrace2(stream->r)) {

321 ap_log_rerror(APLOG_MARK, APLOG_TRACE2, 0, stream->r,

322 "h2_proxy_stream(%s-%d), header_out after merging",

323 stream->session->id, stream->id);

324 apr_table_do(log_header, stream, stream->r->headers_out, NULL);

325 }

326 }

327

328 static int on_data_chunk_recv(nghttp2_session *ngh2, uint8_t flags,

329 int32_t stream_id, const uint8_t *data,

330 size_t len, void *user_data)

331 {

332 h2_proxy_session *session = user_data;

333 h2_proxy_stream *stream;

334 apr_bucket *b;

335 apr_status_t status;

336

337 stream = nghttp2_session_get_stream_user_data(ngh2, stream_id);

338 if (!stream) {

339 ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, ap_server_conf, APLOGNO(03358)

340 "h2_proxy_session(%s): recv data chunk for "

341 "unknown stream %d, ignored",

342 session->id, stream_id);

343 return 0;

344 }

345

346 if (!stream->data_received) {

347 /* last chance to manipulate response headers.

348 * after this, only trailers */

349 h2_proxy_stream_end_headers_out(stream);

350 stream->data_received = 1;

351 }

352

353 b = apr_bucket_transient_create((const char*)data, len,

354 stream->r->connection->bucket_alloc);

355 APR_BRIGADE_INSERT_TAIL(stream->output, b);

356 /* always flush after a DATA frame, as we have no other indication

357 * of buffer use */

358 b = apr_bucket_flush_create(stream->r->connection->bucket_alloc);

359 APR_BRIGADE_INSERT_TAIL(stream->output, b);

360

361 ap_log_rerror(APLOG_MARK, APLOG_DEBUG, 0, stream->r, APLOGNO(03359)

362 "h2_proxy_session(%s): pass response data for "

363 "stream %d, %d bytes", session->id, stream_id, (int)len);

364 status = ap_pass_brigade(stream->r->output_filters, stream->output);

365 if (status != APR_SUCCESS) {

366 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, session->c, APLOGNO(03344)

367 "h2_proxy_session(%s): passing output on stream %d",

368 session->id, stream->id);

369 nghttp2_submit_rst_stream(ngh2, NGHTTP2_FLAG_NONE,

370 stream_id, NGHTTP2_STREAM_CLOSED);

371 return NGHTTP2_ERR_STREAM_CLOSING;

372 }

373 return 0;

374 }

375

376 static int on_stream_close(nghttp2_session *ngh2, int32_t stream_id,

377 uint32_t error_code, void *user_data)

378 {

379 h2_proxy_session *session = user_data;

380 if (!session->aborted) {

381 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03360)

382 "h2_proxy_session(%s): stream=%d, closed, err=%d",

383 session->id, stream_id, error_code);

384 dispatch_event(session, H2_PROXYS_EV_STREAM_DONE, stream_id, NULL);

385 }

386 return 0;

387 }

388

389 static int on_header(nghttp2_session *ngh2, const nghttp2_frame *frame,

390 const uint8_t *namearg, size_t nlen,

391 const uint8_t *valuearg, size_t vlen, uint8_t flags,

392 void *user_data)

393 {

394 h2_proxy_session *session = user_data;

395 h2_proxy_stream *stream;

396 const char *n = (const char*)namearg;

397 const char *v = (const char*)valuearg;

398

399 (void)session;

400 if (frame->hd.type == NGHTTP2_HEADERS && nlen) {

401 stream = nghttp2_session_get_stream_user_data(ngh2, frame->hd.stream_id);

402 if (stream) {

403 if (h2_proxy_stream_add_header_out(stream, n, nlen, v, vlen)) {

404 return NGHTTP2_ERR_CALLBACK_FAILURE;

405 }

406 }

407 }

408 else if (frame->hd.type == NGHTTP2_PUSH_PROMISE) {

409 }

410

411 return 0;

412 }

413

414 static ssize_t stream_data_read(nghttp2_session *ngh2, int32_t stream_id,

415 uint8_t *buf, size_t length,

416 uint32_t *data_flags,

417 nghttp2_data_source *source, void *user_data)

418 {

419 h2_proxy_stream *stream;

420 apr_status_t status = APR_SUCCESS;

421

422 *data_flags = 0;

423 stream = nghttp2_session_get_stream_user_data(ngh2, stream_id);

424 if (!stream) {

425 ap_log_error(APLOG_MARK, APLOG_DEBUG, 0, ap_server_conf, APLOGNO(03361)

426 "h2_proxy_stream(%s): data_read, stream %d not found",

427 stream->session->id, stream_id);

428 return NGHTTP2_ERR_CALLBACK_FAILURE;

429 }

430

431 if (APR_BRIGADE_EMPTY(stream->input)) {

432 status = ap_get_brigade(stream->r->input_filters, stream->input,

433 AP_MODE_READBYTES, APR_NONBLOCK_READ,

434 H2MAX(APR_BUCKET_BUFF_SIZE, length));

435 ap_log_rerror(APLOG_MARK, APLOG_TRACE2, status, stream->r,

436 "h2_proxy_stream(%s-%d): request body read",

437 stream->session->id, stream->id);

438 }

439

440 if (status == APR_SUCCESS) {

441 ssize_t readlen = 0;

442 while (status == APR_SUCCESS

443 && (readlen < length)

444 && !APR_BRIGADE_EMPTY(stream->input)) {

445 apr_bucket* b = APR_BRIGADE_FIRST(stream->input);

446 if (APR_BUCKET_IS_METADATA(b)) {

447 if (APR_BUCKET_IS_EOS(b)) {

448 *data_flags |= NGHTTP2_DATA_FLAG_EOF;

449 }

450 else {

451 /* we do nothing more regarding any meta here */

452 }

453 }

454 else {

455 const char *bdata = NULL;

456 apr_size_t blen = 0;

457 status = apr_bucket_read(b, &bdata, &blen, APR_BLOCK_READ);

458

459 if (status == APR_SUCCESS && blen > 0) {

460 ssize_t copylen = H2MIN(length - readlen, blen);

461 memcpy(buf, bdata, copylen);

462 buf += copylen;

463 readlen += copylen;

464 if (copylen < blen) {

465 /* We have data left in the bucket. Split it. */

466 status = apr_bucket_split(b, copylen);

467 }

468 }

469 }

470 apr_bucket_delete(b);

471 }

472

473 ap_log_rerror(APLOG_MARK, APLOG_TRACE2, status, stream->r,

474 "h2_proxy_stream(%d): request body read %ld bytes, flags=%d",

475 stream->id, (long)readlen, (int)*data_flags);

476 return readlen;

477 }

478 else if (APR_STATUS_IS_EAGAIN(status)) {

479 /* suspended stream, needs to be re-awakened */

480 ap_log_rerror(APLOG_MARK, APLOG_TRACE2, status, stream->r,

481 "h2_proxy_stream(%s-%d): suspending",

482 stream->session->id, stream_id);

483 stream->suspended = 1;

484 h2_iq_add(stream->session->suspended, stream->id, NULL, NULL);

485 return NGHTTP2_ERR_DEFERRED;

486 }

487 else {

488 nghttp2_submit_rst_stream(ngh2, NGHTTP2_FLAG_NONE,

489 stream_id, NGHTTP2_STREAM_CLOSED);

490 return NGHTTP2_ERR_STREAM_CLOSING;

491 }

492 }

493

494 h2_proxy_session *h2_proxy_session_setup(const char *id, proxy_conn_rec *p_conn,

495 proxy_server_conf *conf,

496 unsigned char window_bits_connection,

497 unsigned char window_bits_stream,

498 h2_proxy_request_done *done)

499 {

500 if (!p_conn->data) {

501 apr_pool_t *pool = p_conn->scpool;

502 h2_proxy_session *session;

503 nghttp2_session_callbacks *cbs;

504 nghttp2_option *option;

505

506 session = apr_pcalloc(pool, sizeof(*session));

507 apr_pool_pre_cleanup_register(pool, p_conn, proxy_session_pre_close);

508 p_conn->data = session;

509

510 session->id = apr_pstrdup(p_conn->scpool, id);

511 session->c = p_conn->connection;

512 session->p_conn = p_conn;

513 session->conf = conf;

514 session->pool = p_conn->scpool;

515 session->state = H2_PROXYS_ST_INIT;

516 session->window_bits_stream = window_bits_stream;

517 session->window_bits_connection = window_bits_connection;

518 session->streams = h2_ihash_create(pool, offsetof(h2_proxy_stream, id));

519 session->suspended = h2_iq_create(pool, 5);

520 session->done = done;

521

522 session->input = apr_brigade_create(session->pool, session->c->bucket_alloc);

523 session->output = apr_brigade_create(session->pool, session->c->bucket_alloc);

524

525 nghttp2_session_callbacks_new(&cbs);

526 nghttp2_session_callbacks_set_on_frame_recv_callback(cbs, on_frame_recv);

527 nghttp2_session_callbacks_set_on_data_chunk_recv_callback(cbs, on_data_chunk_recv);

528 nghttp2_session_callbacks_set_on_stream_close_callback(cbs, on_stream_close);

529 nghttp2_session_callbacks_set_on_header_callback(cbs, on_header);

530 nghttp2_session_callbacks_set_before_frame_send_callback(cbs, before_frame_send);

531 nghttp2_session_callbacks_set_send_callback(cbs, raw_send);

532

533 nghttp2_option_new(&option);

534 nghttp2_option_set_peer_max_concurrent_streams(option, 100);

535 nghttp2_option_set_no_auto_window_update(option, 1);

536

537 nghttp2_session_client_new2(&session->ngh2, cbs, session, option);

538

539 nghttp2_option_del(option);

540 nghttp2_session_callbacks_del(cbs);

541

542 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03362)

543 "setup session for %s", p_conn->hostname);

544 }

545 return p_conn->data;

546 }

547

548 static apr_status_t session_start(h2_proxy_session *session)

549 {

550 nghttp2_settings_entry settings[2];

551 int rv, add_conn_window;

552 apr_socket_t *s;

553

554 s = ap_get_conn_socket(session->c);

555 #if (!defined(WIN32) && !defined(NETWARE)) || defined(DOXYGEN)

556 if (s) {

557 ap_sock_disable_nagle(s);

558 }

559 #endif

560

561 settings[0].settings_id = NGHTTP2_SETTINGS_ENABLE_PUSH;

562 settings[0].value = 0;

563 settings[1].settings_id = NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE;

564 settings[1].value = (1 << session->window_bits_stream) - 1;

565

566 rv = nghttp2_submit_settings(session->ngh2, NGHTTP2_FLAG_NONE, settings,

567 H2_ALEN(settings));

568

569 /* If the connection window is larger than our default, trigger a WINDOW_UPDATE */

570 add_conn_window = ((1 << session->window_bits_connection) - 1 -

571 NGHTTP2_INITIAL_CONNECTION_WINDOW_SIZE);

572 if (!rv && add_conn_window != 0) {

573 rv = nghttp2_submit_window_update(session->ngh2, NGHTTP2_FLAG_NONE, 0, add_conn_window);

574 }

575 return rv? APR_EGENERAL : APR_SUCCESS;

576 }

577

578 static apr_status_t open_stream(h2_proxy_session *session, const char *url,

579 request_rec *r, h2_proxy_stream **pstream)

580 {

581 h2_proxy_stream *stream;

582 apr_uri_t puri;

583 const char *authority, *scheme, *path;

584 apr_status_t status;

585

586 stream = apr_pcalloc(r->pool, sizeof(*stream));

587

588 stream->pool = r->pool;

589 stream->url = url;

590 stream->r = r;

591 stream->session = session;

592 stream->state = H2_STREAM_ST_IDLE;

593

594 stream->input = apr_brigade_create(stream->pool, session->c->bucket_alloc);

595 stream->output = apr_brigade_create(stream->pool, session->c->bucket_alloc);

596

597 stream->req = h2_req_create(1, stream->pool, 0);

598

599 status = apr_uri_parse(stream->pool, url, &puri);

600 if (status != APR_SUCCESS)

601 return status;

602

603 scheme = (strcmp(puri.scheme, "h2")? "http" : "https");

604 authority = puri.hostname;

605 if (!ap_strchr_c(authority, ':') && puri.port

606 && apr_uri_port_of_scheme(scheme) != puri.port) {

607 /* port info missing and port is not default for scheme: append */

608 authority = apr_psprintf(stream->pool, "%s:%d", authority, puri.port);

609 }

610 path = apr_uri_unparse(stream->pool, &puri, APR_URI_UNP_OMITSITEPART);

611 h2_req_make(stream->req, stream->pool, r->method, scheme,

612 authority, path, r->headers_in);

613

614 /* Tuck away all already existing cookies */

615 stream->saves = apr_table_make(r->pool, 2);

616 apr_table_do(add_header, stream->saves, r->headers_out,"Set-Cookie", NULL);

617

618 *pstream = stream;

619

620 return APR_SUCCESS;

621 }

622

623 static apr_status_t submit_stream(h2_proxy_session *session, h2_proxy_stream *stream)

624 {

625 h2_ngheader *hd;

626 nghttp2_data_provider *pp = NULL;

627 nghttp2_data_provider provider;

628 int rv;

629 apr_status_t status;

630

631 hd = h2_util_ngheader_make_req(stream->pool, stream->req);

632

633 status = ap_get_brigade(stream->r->input_filters, stream->input,

634 AP_MODE_READBYTES, APR_NONBLOCK_READ,

635 APR_BUCKET_BUFF_SIZE);

636 if ((status == APR_SUCCESS && !APR_BUCKET_IS_EOS(APR_BRIGADE_FIRST(stream->input)))

637 || APR_STATUS_IS_EAGAIN(status)) {

638 /* there might be data coming */

639 provider.source.fd = 0;

640 provider.source.ptr = NULL;

641 provider.read_callback = stream_data_read;

642 pp = &provider;

643 }

644

645 rv = nghttp2_submit_request(session->ngh2, NULL,

646 hd->nv, hd->nvlen, pp, stream);

647

648 if (APLOGcdebug(session->c)) {

649 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03363)

650 "h2_proxy_session(%s): submit %s%s -> %d",

651 session->id, stream->req->authority, stream->req->path,

652 rv);

653 }

654

655 if (rv > 0) {

656 stream->id = rv;

657 stream->state = H2_STREAM_ST_OPEN;

658 h2_ihash_add(session->streams, stream);

659 dispatch_event(session, H2_PROXYS_EV_STREAM_SUBMITTED, rv, NULL);

660

661 return APR_SUCCESS;

662 }

663 return APR_EGENERAL;

664 }

665

666 static apr_status_t feed_brigade(h2_proxy_session *session, apr_bucket_brigade *bb)

667 {

668 apr_status_t status = APR_SUCCESS;

669 apr_size_t readlen = 0;

670 ssize_t n;

671

672 while (status == APR_SUCCESS && !APR_BRIGADE_EMPTY(bb)) {

673 apr_bucket* b = APR_BRIGADE_FIRST(bb);

674

675 if (APR_BUCKET_IS_METADATA(b)) {

676 /* nop */

677 }

678 else {

679 const char *bdata = NULL;

680 apr_size_t blen = 0;

681

682 status = apr_bucket_read(b, &bdata, &blen, APR_BLOCK_READ);

683 if (status == APR_SUCCESS && blen > 0) {

684 n = nghttp2_session_mem_recv(session->ngh2, (const uint8_t *)bdata, blen);

685 ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,

686 "h2_proxy_session(%s): feeding %ld bytes -> %ld",

687 session->id, (long)blen, (long)n);

688 if (n < 0) {

689 if (nghttp2_is_fatal((int)n)) {

690 status = APR_EGENERAL;

691 }

692 }

693 else {

694 readlen += n;

695 if (n < blen) {

696 apr_bucket_split(b, n);

697 }

698 }

699 }

700 }

701 apr_bucket_delete(b);

702 }

703

704 ap_log_cerror(APLOG_MARK, APLOG_TRACE1, status, session->c,

705 "h2_proxy_session(%s): fed %ld bytes of input to session",

706 session->id, (long)readlen);

707 if (readlen == 0 && status == APR_SUCCESS) {

708 return APR_EAGAIN;

709 }

710 return status;

711 }

712

713 static apr_status_t h2_proxy_session_read(h2_proxy_session *session, int block,

714 apr_interval_time_t timeout)

715 {

716 apr_status_t status = APR_SUCCESS;

717

718 if (APR_BRIGADE_EMPTY(session->input)) {

719 apr_socket_t *socket = NULL;

720 apr_time_t save_timeout = -1;

721

722 if (block) {

723 socket = ap_get_conn_socket(session->c);

724 if (socket) {

725 apr_socket_timeout_get(socket, &save_timeout);

726 apr_socket_timeout_set(socket, timeout);

727 }

728 else {

729 /* cannot block on timeout */

730 ap_log_cerror(APLOG_MARK, APLOG_WARNING, 0, session->c, APLOGNO(03379)

731 "h2_proxy_session(%s): unable to get conn socket",

732 session->id);

733 return APR_ENOTIMPL;

734 }

735 }

736

737 status = ap_get_brigade(session->c->input_filters, session->input,

738 AP_MODE_READBYTES,

739 block? APR_BLOCK_READ : APR_NONBLOCK_READ,

740 64 * 1024);

741 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, session->c,

742 "h2_proxy_session(%s): read from conn", session->id);

743 if (socket && save_timeout != -1) {

744 apr_socket_timeout_set(socket, save_timeout);

745 }

746 }

747

748 if (status == APR_SUCCESS) {

749 status = feed_brigade(session, session->input);

750 }

751 else if (APR_STATUS_IS_TIMEUP(status)) {

752 /* nop */

753 }

754 else if (!APR_STATUS_IS_EAGAIN(status)) {

755 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, session->c, APLOGNO(03380)

756 "h2_proxy_session(%s): read error", session->id);

757 dispatch_event(session, H2_PROXYS_EV_CONN_ERROR, status, NULL);

758 }

759

760 return status;

761 }

762

763 apr_status_t h2_proxy_session_submit(h2_proxy_session *session,

764 const char *url, request_rec *r)

765 {

766 h2_proxy_stream *stream;

767 apr_status_t status;

768

769 status = open_stream(session, url, r, &stream);

770 if (status == APR_SUCCESS) {

771 ap_log_rerror(APLOG_MARK, APLOG_DEBUG, 0, r, APLOGNO(03381)

772 "process stream(%d): %s %s%s, original: %s",

773 stream->id, stream->req->method,

774 stream->req->authority, stream->req->path,

775 r->the_request);

776 status = submit_stream(session, stream);

777 }

778 return status;

779 }

780

781 static apr_status_t check_suspended(h2_proxy_session *session)

782 {

783 h2_proxy_stream *stream;

784 int i, stream_id;

785 apr_status_t status;

786

787 for (i = 0; i < session->suspended->nelts; ++i) {

788 stream_id = session->suspended->elts[i];

789 stream = nghttp2_session_get_stream_user_data(session->ngh2, stream_id);

790 if (stream) {

791 status = ap_get_brigade(stream->r->input_filters, stream->input,

792 AP_MODE_READBYTES, APR_NONBLOCK_READ,

793 APR_BUCKET_BUFF_SIZE);

794 if (status == APR_SUCCESS && !APR_BRIGADE_EMPTY(stream->input)) {

795 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, status, session->c,

796 "h2_proxy_stream(%s-%d): resuming",

797 session->id, stream_id);

798 stream->suspended = 0;

799 h2_iq_remove(session->suspended, stream_id);

800 nghttp2_session_resume_data(session->ngh2, stream_id);

801 dispatch_event(session, H2_PROXYS_EV_STREAM_RESUMED, 0, NULL);

802 check_suspended(session);

803 return APR_SUCCESS;

804 }

805 else if (status != APR_SUCCESS && !APR_STATUS_IS_EAGAIN(status)) {

806 ap_log_cerror(APLOG_MARK, APLOG_WARNING, status, session->c,

807 APLOGNO(03382) "h2_proxy_stream(%s-%d): check input",

808 session->id, stream_id);

809 h2_iq_remove(session->suspended, stream_id);

810 dispatch_event(session, H2_PROXYS_EV_STREAM_RESUMED, 0, NULL);

811 check_suspended(session);

812 return APR_SUCCESS;

813 }

814 }

815 else {

816 /* gone? */

817 h2_iq_remove(session->suspended, stream_id);

818 check_suspended(session);

819 return APR_SUCCESS;

820 }

821 }

822 return APR_EAGAIN;

823 }

824

825 static apr_status_t session_shutdown(h2_proxy_session *session, int reason,

826 const char *msg)

827 {

828 apr_status_t status = APR_SUCCESS;

829 const char *err = msg;

830

831 AP_DEBUG_ASSERT(session);

832 if (!err && reason) {

833 err = nghttp2_strerror(reason);

834 }

835 nghttp2_submit_goaway(session->ngh2, NGHTTP2_FLAG_NONE, 0,

836 reason, (uint8_t*)err, err? strlen(err):0);

837 status = nghttp2_session_send(session->ngh2);

838 dispatch_event(session, H2_PROXYS_EV_LOCAL_GOAWAY, reason, err);

839 return status;

840 }

841

842

843 static const char *StateNames[] = {

844 "INIT", /* H2_PROXYS_ST_INIT */

845 "DONE", /* H2_PROXYS_ST_DONE */

846 "IDLE", /* H2_PROXYS_ST_IDLE */

847 "BUSY", /* H2_PROXYS_ST_BUSY */

848 "WAIT", /* H2_PROXYS_ST_WAIT */

849 "LSHUTDOWN", /* H2_PROXYS_ST_LOCAL_SHUTDOWN */

850 "RSHUTDOWN", /* H2_PROXYS_ST_REMOTE_SHUTDOWN */

851 };

852

853 static const char *state_name(h2_proxys_state state)

854 {

855 if (state >= (sizeof(StateNames)/sizeof(StateNames[0]))) {

856 return "unknown";

857 }

858 return StateNames[state];

859 }

860

861 static int is_accepting_streams(h2_proxy_session *session)

862 {

863 switch (session->state) {

864 case H2_PROXYS_ST_IDLE:

865 case H2_PROXYS_ST_BUSY:

866 case H2_PROXYS_ST_WAIT:

867 return 1;

868 default:

869 return 0;

870 }

871 }

872

873 static void transit(h2_proxy_session *session, const char *action,

874 h2_proxys_state nstate)

875 {

876 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03345)

877 "h2_proxy_session(%s): transit [%s] -- %s --> [%s]", session->id,

878 state_name(session->state), action, state_name(nstate));

879 session->state = nstate;

880 }

881

882 static void ev_init(h2_proxy_session *session, int arg, const char *msg)

883 {

884 switch (session->state) {

885 case H2_PROXYS_ST_INIT:

886 if (h2_ihash_empty(session->streams)) {

887 transit(session, "init", H2_PROXYS_ST_IDLE);

888 }

889 else {

890 transit(session, "init", H2_PROXYS_ST_BUSY);

891 }

892 break;

893

894 default:

895 /* nop */

896 break;

897 }

898 }

899

900 static void ev_local_goaway(h2_proxy_session *session, int arg, const char *msg)

901 {

902 switch (session->state) {

903 case H2_PROXYS_ST_LOCAL_SHUTDOWN:

904 /* already did that? */

905 break;

906 case H2_PROXYS_ST_IDLE:

907 case H2_PROXYS_ST_REMOTE_SHUTDOWN:

908 /* all done */

909 transit(session, "local goaway", H2_PROXYS_ST_DONE);

910 break;

911 default:

912 transit(session, "local goaway", H2_PROXYS_ST_LOCAL_SHUTDOWN);

913 break;

914 }

915 }

916

917 static void ev_remote_goaway(h2_proxy_session *session, int arg, const char *msg)

918 {

919 switch (session->state) {

920 case H2_PROXYS_ST_REMOTE_SHUTDOWN:

921 /* already received that? */

922 break;

923 case H2_PROXYS_ST_IDLE:

924 case H2_PROXYS_ST_LOCAL_SHUTDOWN:

925 /* all done */

926 transit(session, "remote goaway", H2_PROXYS_ST_DONE);

927 break;

928 default:

929 transit(session, "remote goaway", H2_PROXYS_ST_REMOTE_SHUTDOWN);

930 break;

931 }

932 }

933

934 static void ev_conn_error(h2_proxy_session *session, int arg, const char *msg)

935 {

936 switch (session->state) {

937 case H2_PROXYS_ST_INIT:

938 case H2_PROXYS_ST_DONE:

939 case H2_PROXYS_ST_LOCAL_SHUTDOWN:

940 /* just leave */

941 transit(session, "conn error", H2_PROXYS_ST_DONE);

942 break;

943

944 default:

945 ap_log_cerror(APLOG_MARK, APLOG_TRACE1, arg, session->c,

946 "h2_proxy_session(%s): conn error -> shutdown", session->id);

947 session_shutdown(session, arg, msg);

948 break;

949 }

950 }

951

952 static void ev_proto_error(h2_proxy_session *session, int arg, const char *msg)

953 {

954 switch (session->state) {

955 case H2_PROXYS_ST_DONE:

956 case H2_PROXYS_ST_LOCAL_SHUTDOWN:

957 /* just leave */

958 transit(session, "proto error", H2_PROXYS_ST_DONE);

959 break;

960

961 default:

962 ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,

963 "h2_proxy_session(%s): proto error -> shutdown", session->id);

964 session_shutdown(session, arg, msg);

965 break;

966 }

967 }

968

969 static void ev_conn_timeout(h2_proxy_session *session, int arg, const char *msg)

970 {

971 switch (session->state) {

972 case H2_PROXYS_ST_LOCAL_SHUTDOWN:

973 transit(session, "conn timeout", H2_PROXYS_ST_DONE);

974 break;

975 default:

976 session_shutdown(session, arg, msg);

977 transit(session, "conn timeout", H2_PROXYS_ST_DONE);

978 break;

979 }

980 }

981

982 static void ev_no_io(h2_proxy_session *session, int arg, const char *msg)

983 {

984 switch (session->state) {

985 case H2_PROXYS_ST_BUSY:

986 case H2_PROXYS_ST_LOCAL_SHUTDOWN:

987 case H2_PROXYS_ST_REMOTE_SHUTDOWN:

988 /* nothing for input and output to do. If we remain

989 * in this state, we go into a tight loop and suck up

990 * CPU cycles. Ideally, we'd like to do a blocking read, but that

991 * is not possible if we have scheduled tasks and wait

992 * for them to produce something. */

993 if (h2_ihash_empty(session->streams)) {

994 if (!is_accepting_streams(session)) {

995 /* We are no longer accepting new streams and have

996 * finished processing existing ones. Time to leave. */

997 session_shutdown(session, arg, msg);

998 transit(session, "no io", H2_PROXYS_ST_DONE);

999 }

1000 else {

1001 /* When we have no streams, no task events are possible,

1002 * switch to blocking reads */

1003 transit(session, "no io", H2_PROXYS_ST_IDLE);

1004 }

1005 }

1006 else {

1007 /* Unable to do blocking reads, as we wait on events from

1008 * task processing in other threads. Do a busy wait with

1009 * backoff timer. */

1010 transit(session, "no io", H2_PROXYS_ST_WAIT);

1011 }

1012 break;

1013 default:

1014 /* nop */

1015 break;

1016 }

1017 }

1018

1019 static void ev_stream_submitted(h2_proxy_session *session, int stream_id,

1020 const char *msg)

1021 {

1022 switch (session->state) {

1023 case H2_PROXYS_ST_IDLE:

1024 case H2_PROXYS_ST_WAIT:

1025 transit(session, "stream submitted", H2_PROXYS_ST_BUSY);

1026 break;

1027 default:

1028 /* nop */

1029 break;

1030 }

1031 }

1032

1033 static void ev_stream_done(h2_proxy_session *session, int stream_id,

1034 const char *msg)

1035 {

1036 h2_proxy_stream *stream;

1037

1038 stream = nghttp2_session_get_stream_user_data(session->ngh2, stream_id);

1039 if (stream) {

1040 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03364)

1041 "h2_proxy_sesssion(%s): stream(%d) closed",

1042 session->id, stream_id);

1043

1044 if (!stream->data_received) {

1045 apr_bucket *b;

1046 /* if the response had no body, this is the time to flush

1047 * an empty brigade which will also "write" the resonse

1048 * headers */

1049 h2_proxy_stream_end_headers_out(stream);

1050 stream->data_received = 1;

1051 b = apr_bucket_flush_create(stream->r->connection->bucket_alloc);

1052 APR_BRIGADE_INSERT_TAIL(stream->output, b);

1053 b = apr_bucket_eos_create(stream->r->connection->bucket_alloc);

1054 APR_BRIGADE_INSERT_TAIL(stream->output, b);

1055 ap_pass_brigade(stream->r->output_filters, stream->output);

1056 }

1057

1058 stream->state = H2_STREAM_ST_CLOSED;

1059 h2_ihash_remove(session->streams, stream_id);

1060 h2_iq_remove(session->suspended, stream_id);

1061 if (session->done) {

1062 session->done(session, stream->r, 1, 1);

1063 }

1064 }

1065

1066 switch (session->state) {

1067 default:

1068 /* nop */

1069 break;

1070 }

1071 }

1072

1073 static void ev_stream_resumed(h2_proxy_session *session, int arg, const char *msg)

1074 {

1075 switch (session->state) {

1076 case H2_PROXYS_ST_WAIT:

1077 transit(session, "stream resumed", H2_PROXYS_ST_BUSY);

1078 break;

1079 default:

1080 /* nop */

1081 break;

1082 }

1083 }

1084

1085 static void ev_data_read(h2_proxy_session *session, int arg, const char *msg)

1086 {

1087 switch (session->state) {

1088 case H2_PROXYS_ST_IDLE:

1089 case H2_PROXYS_ST_WAIT:

1090 transit(session, "data read", H2_PROXYS_ST_BUSY);

1091 break;

1092 default:

1093 /* nop */

1094 break;

1095 }

1096 }

1097

1098 static void ev_ngh2_done(h2_proxy_session *session, int arg, const char *msg)

1099 {

1100 switch (session->state) {

1101 case H2_PROXYS_ST_DONE:

1102 /* nop */

1103 break;

1104 default:

1105 transit(session, "nghttp2 done", H2_PROXYS_ST_DONE);

1106 break;

1107 }

1108 }

1109

1110 static void ev_pre_close(h2_proxy_session *session, int arg, const char *msg)

1111 {

1112 switch (session->state) {

1113 case H2_PROXYS_ST_DONE:

1114 case H2_PROXYS_ST_LOCAL_SHUTDOWN:

1115 /* nop */

1116 break;

1117 default:

1118 session_shutdown(session, arg, msg);

1119 break;

1120 }

1121 }

1122

1123 static void dispatch_event(h2_proxy_session *session, h2_proxys_event_t ev,

1124 int arg, const char *msg)

1125 {

1126 switch (ev) {

1127 case H2_PROXYS_EV_INIT:

1128 ev_init(session, arg, msg);

1129 break;

1130 case H2_PROXYS_EV_LOCAL_GOAWAY:

1131 ev_local_goaway(session, arg, msg);

1132 break;

1133 case H2_PROXYS_EV_REMOTE_GOAWAY:

1134 ev_remote_goaway(session, arg, msg);

1135 break;

1136 case H2_PROXYS_EV_CONN_ERROR:

1137 ev_conn_error(session, arg, msg);

1138 break;

1139 case H2_PROXYS_EV_PROTO_ERROR:

1140 ev_proto_error(session, arg, msg);

1141 break;

1142 case H2_PROXYS_EV_CONN_TIMEOUT:

1143 ev_conn_timeout(session, arg, msg);

1144 break;

1145 case H2_PROXYS_EV_NO_IO:

1146 ev_no_io(session, arg, msg);

1147 break;

1148 case H2_PROXYS_EV_STREAM_SUBMITTED:

1149 ev_stream_submitted(session, arg, msg);

1150 break;

1151 case H2_PROXYS_EV_STREAM_DONE:

1152 ev_stream_done(session, arg, msg);

1153 break;

1154 case H2_PROXYS_EV_STREAM_RESUMED:

1155 ev_stream_resumed(session, arg, msg);

1156 break;

1157 case H2_PROXYS_EV_DATA_READ:

1158 ev_data_read(session, arg, msg);

1159 break;

1160 case H2_PROXYS_EV_NGH2_DONE:

1161 ev_ngh2_done(session, arg, msg);

1162 break;

1163 case H2_PROXYS_EV_PRE_CLOSE:

1164 ev_pre_close(session, arg, msg);

1165 break;

1166 default:

1167 ap_log_cerror(APLOG_MARK, APLOG_TRACE1, 0, session->c,

1168 "h2_proxy_session(%s): unknown event %d",

1169 session->id, ev);

1170 break;

1171 }

1172 }

1173

1174 apr_status_t h2_proxy_session_process(h2_proxy_session *session)

1175 {

1176 apr_status_t status;

1177 int have_written = 0, have_read = 0;

1178

1179 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,

1180 "h2_proxy_session(%s): process", session->id);

1181

1182 run_loop:

1183 switch (session->state) {

1184 case H2_PROXYS_ST_INIT:

1185 status = session_start(session);

1186 if (status == APR_SUCCESS) {

1187 dispatch_event(session, H2_PROXYS_EV_INIT, 0, NULL);

1188 goto run_loop;

1189 }

1190 else {

1191 dispatch_event(session, H2_PROXYS_EV_CONN_ERROR, status, NULL);

1192 }

1193 break;

1194

1195 case H2_PROXYS_ST_BUSY:

1196 case H2_PROXYS_ST_LOCAL_SHUTDOWN:

1197 case H2_PROXYS_ST_REMOTE_SHUTDOWN:

1198 while (nghttp2_session_want_write(session->ngh2)) {

1199 int rv = nghttp2_session_send(session->ngh2);

1200 if (rv < 0 && nghttp2_is_fatal(rv)) {

1201 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,

1202 "h2_proxy_session(%s): write, rv=%d", session->id, rv);

1203 dispatch_event(session, H2_PROXYS_EV_CONN_ERROR, rv, NULL);

1204 break;

1205 }

1206 have_written = 1;

1207 }

1208

1209 if (nghttp2_session_want_read(session->ngh2)) {

1210 status = h2_proxy_session_read(session, 0, 0);

1211 if (status == APR_SUCCESS) {

1212 have_read = 1;

1213 }

1214 }

1215

1216 if (!have_written && !have_read

1217 && !nghttp2_session_want_write(session->ngh2)) {

1218 dispatch_event(session, H2_PROXYS_EV_NO_IO, 0, NULL);

1219 goto run_loop;

1220 }

1221 break;

1222

1223 case H2_PROXYS_ST_WAIT:

1224 if (check_suspended(session) == APR_EAGAIN) {

1225 /* no stream has become resumed. Do a blocking read with

1226 * ever increasing timeouts... */

1227 if (session->wait_timeout < 25) {

1228 session->wait_timeout = 25;

1229 }

1230 else {

1231 session->wait_timeout = H2MIN(apr_time_from_msec(100),

1232 2*session->wait_timeout);

1233 }

1234

1235 status = h2_proxy_session_read(session, 1, session->wait_timeout);

1236 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, status, session->c,

1237 APLOGNO(03365)

1238 "h2_proxy_session(%s): WAIT read, timeout=%fms",

1239 session->id, (float)session->wait_timeout/1000.0);

1240 if (status == APR_SUCCESS) {

1241 have_read = 1;

1242 dispatch_event(session, H2_PROXYS_EV_DATA_READ, 0, NULL);

1243 }

1244 else if (APR_STATUS_IS_TIMEUP(status)

1245 || APR_STATUS_IS_EAGAIN(status)) {

1246 /* go back to checking all inputs again */

1247 transit(session, "wait cycle", H2_PROXYS_ST_BUSY);

1248 }

1249 }

1250 break;

1251

1252 case H2_PROXYS_ST_IDLE:

1253 break;

1254

1255 case H2_PROXYS_ST_DONE: /* done, session terminated */

1256 return APR_EOF;

1257

1258 default:

1259 ap_log_cerror(APLOG_MARK, APLOG_ERR, APR_EGENERAL, session->c,

1260 APLOGNO(03346)"h2_proxy_session(%s): unknown state %d",

1261 session->id, session->state);

1262 dispatch_event(session, H2_PROXYS_EV_PROTO_ERROR, 0, NULL);

1263 break;

1264 }

1265

1266

1267 if (have_read || have_written) {

1268 session->wait_timeout = 0;

1269 }

1270

1271 if (!nghttp2_session_want_read(session->ngh2)

1272 && !nghttp2_session_want_write(session->ngh2)) {

1273 dispatch_event(session, H2_PROXYS_EV_NGH2_DONE, 0, NULL);

1274 }

1275

1276 return APR_SUCCESS; /* needs to be called again */

1277 }

1278

1279 typedef struct {

1280 h2_proxy_session *session;

1281 h2_proxy_request_done *done;

1282 } cleanup_iter_ctx;

1283

1284 static int done_iter(void *udata, void *val)

1285 {

1286 cleanup_iter_ctx *ctx = udata;

1287 h2_proxy_stream *stream = val;

1288 int touched = (!ctx->session->last_stream_id ||

1289 stream->id <= ctx->session->last_stream_id);

1290 ctx->done(ctx->session, stream->r, 0, touched);

1291 return 1;

1292 }

1293

1294 void h2_proxy_session_cleanup(h2_proxy_session *session,

1295 h2_proxy_request_done *done)

1296 {

1297 if (session->streams && !h2_ihash_empty(session->streams)) {

1298 cleanup_iter_ctx ctx;

1299 ctx.session = session;

1300 ctx.done = done;

1301 ap_log_cerror(APLOG_MARK, APLOG_DEBUG, 0, session->c, APLOGNO(03366)

1302 "h2_proxy_session(%s): terminated, %d streams unfinished",

1303 session->id, (int)h2_ihash_count(session->streams));

1304 h2_ihash_iter(session->streams, done_iter, &ctx);

1305 h2_ihash_clear(session->streams);

1306 }

1307 }

1308

1309 typedef struct {

1310 h2_proxy_session *session;

1311 conn_rec *c;

1312 apr_off_t bytes;

1313 int updated;

1314 } win_update_ctx;

1315

1316 static int win_update_iter(void *udata, void *val)

1317 {

1318 win_update_ctx *ctx = udata;

1319 h2_proxy_stream *stream = val;

1320

1321 if (stream->r && stream->r->connection == ctx->c) {

1322 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, ctx->session->c,

1323 "h2_proxy_session(%s-%d): win_update %ld bytes",

1324 ctx->session->id, (int)stream->id, (long)ctx->bytes);

1325 nghttp2_session_consume(ctx->session->ngh2, stream->id, ctx->bytes);

1326 ctx->updated = 1;

1327 return 0;

1328 }

1329 return 1;

1330 }

1331

1332

1333 void h2_proxy_session_update_window(h2_proxy_session *session,

1334 conn_rec *c, apr_off_t bytes)

1335 {

1336 if (session->streams && !h2_ihash_empty(session->streams)) {

1337 win_update_ctx ctx;

1338 ctx.session = session;

1339 ctx.c = c;

1340 ctx.bytes = bytes;

1341 ctx.updated = 0;

1342 h2_ihash_iter(session->streams, win_update_iter, &ctx);

1343

1344 if (!ctx.updated) {

1345 /* could not find the stream any more, possibly closed, update

1346 * the connection window at least */

1347 ap_log_cerror(APLOG_MARK, APLOG_TRACE2, 0, session->c,

1348 "h2_proxy_session(%s): win_update conn %ld bytes",

1349 session->id, (long)bytes);

1350 nghttp2_session_consume_connection(session->ngh2, (size_t)bytes);

1351 }

1352 }

1353 }

1354