proxy.c

   1 #include <sys/queue.h>
   2 #include <event2/event.h>
   3 #include <event2/listener.h>
   4 #include "proxy.h"
   5 #include "httpconn.h"
   6
   7 enum server_state {
   8         SERVER_STATE_INITIAL,
   9         SERVER_STATE_CONNECTING,
  10         SERVER_STATE_CONNECTED,
  11         SERVER_STATE_REQUEST_SENT,
  12         SERVER_STATE_IDLE
  13 };
  14
  15 struct server {
  16         TAILQ_ENTRY(server) next;
  17         enum server_state state;
  18         size_t nserviced;
  19         char *host;
  20         int port;
  21         struct http_conn *conn;
  22         struct client *client;
  23 };
  24 TAILQ_HEAD(server_list, server);
  25
  26 enum client_state {
  27         CLIENT_STATE_ACTIVE,
  28         CLIENT_STATE_CLOSING
  29 };
  30
  31 struct client {
  32         enum client_state state;
  33         struct server_list requests;
  34         size_t nrequests;
  35         struct http_conn *conn;
  36         struct server *server;
  37 };
  38
  39 static void on_client_error(struct http_conn *, enum http_conn_error, void *);
  40 static void on_client_request(struct http_conn *, struct http_request *, void *);
  41 static void on_client_read_body(struct http_conn *, struct evbuffer *, void *);
  42 static void on_client_msg_complete(struct http_conn *, void *);
  43 static void on_client_write_more(struct http_conn *, void *);
  44
  45 static void on_server_connected(struct http_conn *, void *);
  46 static void on_server_error(struct http_conn *, enum http_conn_error, void *);
  47 static void on_server_response(struct http_conn *, struct http_response *, void *);
  48 static void on_server_read_body(struct http_conn *, struct evbuffer *, void *);
  49 static void on_server_msg_complete(struct http_conn *, void *);
  50 static void on_server_write_more(struct http_conn *, void *);
  51
  52 static const struct http_cbs client_methods = {
  53         0,
  54         on_client_error,
  55         on_client_request,
  56         0,
  57         on_client_read_body,
  58         on_client_msg_complete,
  59         on_client_write_more,
  60         on_client_flush
  61 };
  62
  63 static const struct http_cbs server_methods = {
  64         on_server_connected,
  65         on_server_error,
  66         0,
  67         on_server_response,
  68         on_server_read_body,
  69         on_server_msg_complete,
  70         on_server_write_more,
  71         on_server_flush
  72 };
  73
  74 static struct event_base *proxy_event_base;
  75 static struct evdns_base *proxy_evdns_base;
  76 static struct evconnlistener *listener = NULL;
  77 static struct server_list idle_servers = TAILQ_HEAD_INITIALIZER(&idle_servers);
  78 static size_t max_pending_requests = 8;
  79
  80 static struct server *
  81 server_new(const char *host, int port)
  82 {
  83         struct server *server;
  84
  85         server = mem_calloc(1, sizeof(*server));
  86         server->host = mem_strdup(host);
  87         server->port = port;
  88         server->conn = http_conn_new(proxy_event_base, -1, HTTP_SERVER,
  89                                 &server_methods, server);
  90
  91         return server;
  92 }
  93
  94 static void
  95 server_free(struct server *server)
  96 {
  97         if (!server)
  98                 return;
  99
 100         mem_free(server->host);
 101         http_conn_free(server->conn);
 102         mem_free(server);
 103 }
 104
 105 static int
 106 server_connect(struct server *server)
 107 {
 108         server->state = SERVER_STATE_CONNECTING;
 109         return http_conn_connect(server->conn, proxy_evdns_base, AF_UNSPEC,
 110                                  server->host, server->port);
 111 }
 112
 113 static int
 114 server_write_client_request(struct server *server)
 115 {
 116         struct http_request *req;
 117
 118         req = TAILQ_FIRST(&server->client->requests);
 119
 120         if (!req || server->state == SERVER_STATE_REQUEST_SENT ||
 121             server->state < SERVER_STATE_CONNECTED)
 122                 return 0;
 123
 124         /* it might be nice to support pipelining... */
 125         if (!evutil_ascii_strcasecmp(server->host, req->url->host) &&
 126             server->port == req->url->port) {
 127                 http_conn_write_request(server->conn, req);
 128                 server->state = SERVER_STATE_REQUEST_SENT;
 129                 return 1;
 130         }
 131
 132         return 0;
 133 }
 134
 135 static struct client *
 136 client_new(evutil_socket_t sock)
 137 {
 138         struct client *client;
 139
 140         client = mem_calloc(1, sizeof(*client));
 141         TAILQ_INIT(&client->requests);
 142         client->conn = http_conn_new(proxy_event_base, sock, HTTP_CLIENT,
 143                                 &client_methods, client);
 144
 145         return client;
 146 }
 147
 148 static void
 149 client_free(struct client *client)
 150 {
 151         struct http_request *req;
 152
 153         if (!client)
 154                 return;
 155
 156         while ((req = TAILQ_FIRST(&client->requests))) {
 157                 TAILQ_REMOVE(req, &client->requests, next);
 158                 http_request_free(req);
 159         }
 160
 161         server_free(client->server);
 162         http_conn_free(client->conn);
 163         mem_free(client);
 164 }
 165
 166 static int
 167 client_scrub_request(struct client *client, struct http_request *req)
 168 {
 169         // prune headers; verify that req contains a host to connect to
 170         // XXX remove proxy auth msgs?
 171
 172         if (!req->url->host) {
 173                 http_conn_send_error(client->conn, 401);
 174                 goto fail;
 175         }
 176         if (evutil_ascii_strcasecmp(req->url->scheme, "http")) {
 177                 http_conn_send_error(client->conn, 400);
 178                 goto fail;
 179         }
 180
 181         if (req->url->port < 0)
 182                 req->url->port = 80;
 183
 184         return 0;
 185
 186 fail:
 187         http_request_free(req);
 188         return -1;
 189 }
 190
 191 static int
 192 client_associate_server(struct client *client, const struct url *url)
 193 {
 194         struct server *it;
 195
 196         assert(client->server == NULL);
 197         assert(url->host != NULL && url->port > 0);
 198
 199         /* try to find an idle server */
 200         TAILQ_FOREACH(it, &idle_servers, next) {
 201                 if (!evutil_ascii_strcasecmp(it->host, url->host) &&
 202                     it->port == url->port) {
 203                         TAILQ_REMOVE(&idle_servers, it, next);
 204                         client->server = it;
 205                         return 0;
 206                 }
 207         }
 208
 209         /* we didn't find one. lets setup a new one. */
 210         client->server = server_new(url->host, url->port);
 211         server->client = client;
 212
 213         return server_connect(client->server);
 214 }
 215
 216 static void
 217 client_start_reading_request_body(struct client *client)
 218 {
 219         if (http_conn_current_message_has_body(client->conn) &&
 220             client->nrequests == 1)
 221                 http_conn_start_reading(client->conn);
 222 }
 223
 224 static void
 225 client_request_serviced(struct client *client)
 226 {
 227         struct http_request *req;
 228
 229         assert(client->nrequests > 0)
 230
 231         // - pop the first req on our req list
 232         // - if the server's connection isn't persistent, we should terminate the server connection.
 233         // - if the client's connection isn't persistent, we should terminate the client connection.
 234         // - if we expect more reqs on the client connection, start reading again in case we stopped
 235         //   earlier.
 236
 237         req = TAILQ_FIRST(&client->requests);
 238         TAILQ_REMOVE(&client->requests, req, next);
 239         http_request_free(req);
 240         client->nrequests--;
 241
 242         if (client->server) {
 243                 /* let's try to reuse this server connection */
 244                 if (http_conn_is_persistent(client->server->conn)) {
 245                         if (!server_write_client_request(client->server)) {
 246                                 client->server->state = SERVER_STATE_IDLE;
 247                                 TAILQ_INSERT_TAIL(&idle_servers,
 248                                                   client->server,
 249                                                   next);
 250                                 client->server->client = NULL;
 251                                 client->server = NULL;
 252                         }
 253                 } else {
 254                         server_free(client->server);
 255                         client->server = NULL;
 256                 }
 257         }
 258
 259         if (!http_conn_is_persistent(client->conn)) {
 260                 // XXX maybe shutdown the socket?
 261                 client->state = STATE_CLIENT_CLOSING;
 262                 http_conn_stop_reading(client->conn);
 263                 http_conn_flush(client->conn);
 264         } else if (!http_conn_current_message_has_body(client->conn) &&
 265                    client->nrequests < max_pending_requests) {
 266                 http_conn_start_reading(client->conn);
 267         }
 268 }
 269
 270 static void
 271 client_notice_server_failed(struct client *client)
 272 {
 273         struct http_request *req;
 274         struct server *server = client->server;
 275
 276         client->server = NULL;
 277
 278         while ((req = TAILQ_FIRST(&client->requests))) {
 279                 if (evutil_ascii_strcasecmp(req->url->host, server->host) ||
 280                     req->url->port != server->port)
 281                         break;
 282                 // XXX need a more descriptive error
 283                 http_conn_send_error(client->conn, 502);
 284                 client_request_serviced(client);
 285         }
 286 }
 287
 288 /* http event slots */
 289
 290 static void
 291 on_client_error(struct http_conn *conn, enum http_conn_error err, void *arg)
 292 {
 293         // what can we do here?
 294         log_warn("proxy: client connection failed.");
 295         client_free(arg);
 296 }
 297
 298 static void
 299 on_client_request(struct http_conn *conn, struct http_request *req, void *arg)
 300 {
 301         struct client *client = arg;
 302
 303         // - translate proxy request into a server request
 304         // - if the proxy request doesn't include a scheme and host,
 305         //   probably reject with 404 not found
 306         // - create a new transaction for this request
 307         // - if npending >= max_pending_requests OR the request has a msg body
 308         //   coming, stop reading from the conn for now
 309         // - if client has no active server, connect or reuse an idle one
 310         // - if the active server is for the same address and port as the request,
 311         //   and the server is known to support pipelining, write the request to
 312         //   the server
 313
 314         assert(client->state == CLIENT_STATE_ACTIVE);
 315
 316         if (client_scrub_request(client, req) < 0)
 317                 return;
 318
 319         TAILQ_INSERT_TAIL(&client->requests, req, next);
 320         if (++client->nrequests > max_pending_requests ||
 321             http_conn_current_message_has_body(conn))
 322                 http_conn_stop_reading(conn);
 323
 324         if (!client->server && client_associate_server(client, req->url) < 0)
 325                 return;
 326
 327         server_write_client_request(client->server);
 328 }
 329
 330 static void
 331 on_client_read_body(struct http_conn *conn, struct evbuffer *buf, void *arg)
 332 {
 333         struct client *client = arg;
 334
 335         if (!http_conn_write_buf(client->server->conn, buf))
 336                 http_conn_stop_reading(conn);
 337 }
 338
 339 static void
 340 on_client_msg_complete(struct http_conn *conn, void *arg)
 341 {
 342         struct client *client = arg;
 343
 344         if (http_conn_current_message_has_body(conn))
 345                 http_conn_write_finished(client->server->conn);
 346 }
 347
 348 static void
 349 on_client_write_more(struct http_conn *conn, void *arg)
 350 {
 351         struct client *client = arg;
 352
 353         http_conn_start_reading(client->server->conn);
 354 }
 355
 356 static void
 357 on_client_flush(struct http_conn *conn, void *arg)
 358 {
 359         struct client *client = arg;
 360
 361         // XXX perhaps delay before closing?
 362         if (client->state == CLIENT_STATE_CLOSING)
 363                 client_free(client);
 364 }
 365
 366 static void
 367 on_server_connected(struct http_conn *conn, void *arg)
 368 {
 369         struct server *server = arg;
 370
 371         assert(server->state == SERVER_STATE_CONNECTING);
 372         server->state = SERVER_STATE_CONNECTED;
 373         server_write_client_request(server);
 374 }
 375
 376 static void
 377 on_server_error(struct http_conn *conn, enum http_conn_error err, void *arg)
 378 {
 379         struct server *server = arg;
 380
 381         switch (server->state) {
 382         case SERVER_STATE_CONNECTING:
 383         case SERVER_STATE_CONNECTED:
 384         case SERVER_STATE_REQUEST_SENT:
 385                 // XXX if we haven't serviced any reqs on this server yet,
 386                 //     we should try resending the first request
 387                 assert(server->client != NULL);
 388                 client_notice_server_failed(server->client);
 389                 break;
 390         case SERVER_STATE_IDLE:
 391                 assert(server->client == NULL);
 392                 TAILQ_REMOVE(&idle_servers, server, next);
 393                 break;
 394         default:
 395                 log_fatal("server: error cb called in invalid state");
 396         }
 397
 398         server_free(server);
 399 }
 400
 401 static void
 402 on_server_response(struct http_conn *conn, struct http_response *resp, void *arg)
 403 {
 404         struct server *server = arg;
 405
 406         http_conn_write_response(server->client->conn, resp);
 407         // XXX maybe not read body on error?
 408         // XXX handle expect 100-continue, etc
 409
 410         client_start_reading_body(server->client);
 411 }
 412
 413 static void
 414 on_server_read_body(struct http_conn *conn, struct evbuffer *buf, void *arg)
 415 {
 416         struct server *server = arg;
 417
 418         if (!http_conn_write_buf(server->client->conn, buf))
 419                 http_conn_stop_reading(conn);
 420 }
 421
 422 static void
 423 on_server_msg_complete(struct http_conn *conn, void *arg)
 424 {
 425         struct server *server = arg;
 426
 427         if (http_conn_current_message_has_body(conn))
 428                 http_conn_write_finished(server->client->conn);
 429         client_request_serviced(server->client);
 430 }
 431
 432 static void
 433 on_server_write_more(struct http_conn *conn, void *arg)
 434 {
 435         struct server *server = arg;
 436
 437         http_conn_start_reading(server->client->conn);
 438 }
 439
 440 static void
 441 on_server_flush(struct http_conn *conn, void *arg)
 442 {
 443 }
 444
 445 static void
 446 client_accept(struct evconnlistener *ecs, evutil_socket_t s,
 447               struct sockaddr *addr, int len, void *arg)
 448 {
 449         struct client *client;
 450
 451         client = client_new(s);
 452         // XXX do we want to keep track of the client obj somehow?
 453 }
 454
 455 /* public API */
 456
 457 void
 458 proxy_client_set_max_pending_requests(size_t nreqs)
 459 {
 460         max_pending_requests = nreqs;
 461 }
 462
 463 size_t
 464 proxy_client_get_max_pending_requests(void)
 465 {
 466         return max_pending_requests;
 467 }
 468
 469 int
 470 proxy_init(const struct sockaddr *listen_here, int socklen)
 471 {
 472         struct evconlistener *lcs = NULL;
 473         struct event_base *base = NULL;
 474         struct evdns_base *dns = NULL;
 475
 476         base = event_base_new();
 477         if (!base)
 478                 goto fail;
 479
 480         dns = evdns_base_new(base, 1);
 481         if (!dns)
 482                 goto fail;
 483
 484         lcs = evconnlistener_new_bind(base, client_accept, NULL, CLOSE_ON_FREE,
 485                                         -1, listen_here, len);
 486
 487         if (!lcs) {
 488                 log_error("proxy: couldn't listen on %s: %s",
 489                           format_addr(listen_here),
 490                           socket_error_string(-1));
 491                 goto fail;
 492         }
 493
 494         listener = lcs;
 495         proxy_event_base = base;
 496         proxy_evdns_base = dns;
 497
 498         return 0;
 499
 500 fail:
 501         if (ecs)
 502                 evconlistener_free(ecs);
 503         if (dns)
 504                 evdns_base_free(dns, 0);
 505         if (base)
 506                 event_base_free(base);
 507
 508         return -1;
 509 }
 510
 511 void
 512 proxy_cleanup(void)
 513 {
 514         // TODO
 515 }