<commit_before>/*
 * Copyright (c) 2014-2016 DeNA Co., Ltd., <NAME>, <NAME>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 * IN THE SOFTWARE.
 */
#include <assert.h>
#include <inttypes.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdarg.h>
#include "h2o.h"
#include "h2o/http1.h"
#include "h2o/http2.h"

struct st_h2o_accept_data_t {
    h2o_accept_ctx_t *ctx;
    h2o_socket_t *sock;
    h2o_timeout_entry_t timeout;
    struct timeval connected_at;
};

struct st_h2o_memcached_resumption_accept_data_t {
    struct st_h2o_accept_data_t super;
    h2o_memcached_req_t *get_req;
};

struct st_h2o_redis_resumption_accept_data_t {
    struct st_h2o_accept_data_t super;
    h2o_redis_command_t *get_command;
};

static void on_accept_timeout(h2o_timeout_entry_t *entry);
static void on_redis_accept_timeout(h2o_timeout_entry_t *entry);
static void on_memcached_accept_timeout(h2o_timeout_entry_t *entry);

static struct {
    struct {
        h2o_memcached_context_t *ctx;
    } memcached;
    struct {
        h2o_iovec_t host;
        uint16_t port;
        h2o_iovec_t prefix;
    } redis;
    unsigned expiration;
} async_resumption_context;


static struct st_h2o_accept_data_t *create_accept_data(h2o_accept_ctx_t *ctx, h2o_socket_t *sock, struct timeval connected_at, size_t sz)
{
    struct st_h2o_accept_data_t *data = h2o_mem_alloc(sz);
    data->ctx = ctx;
    data->sock = sock;
    data->timeout = (h2o_timeout_entry_t){0};
    data->timeout.cb = on_accept_timeout;
    h2o_timeout_link(ctx->ctx->loop, &ctx->ctx->handshake_timeout, &data->timeout);
    data->connected_at = connected_at;
    return data;
}

static struct st_h2o_accept_data_t *create_redis_accept_data(h2o_accept_ctx_t *ctx, h2o_socket_t *sock, struct timeval connected_at)
{
    struct st_h2o_redis_resumption_accept_data_t *data =
    (struct st_h2o_redis_resumption_accept_data_t *)create_accept_data(ctx, sock, connected_at, sizeof(struct st_h2o_redis_resumption_accept_data_t));
    data->get_command = NULL;
    data->super.timeout.cb = on_redis_accept_timeout;
    return &data->super;
}

static struct st_h2o_accept_data_t *create_memcached_accept_data(h2o_accept_ctx_t *ctx, h2o_socket_t *sock, struct timeval connected_at)
{
    struct st_h2o_memcached_resumption_accept_data_t *data =
    (struct st_h2o_memcached_resumption_accept_data_t *)create_accept_data(ctx, sock, connected_at, sizeof(struct st_h2o_memcached_resumption_accept_data_t));
    data->get_req = NULL;
    data->super.timeout.cb = on_memcached_accept_timeout;
    return &data->super;
}

static void setup_accept_data(h2o_accept_ctx_t *ctx, h2o_socket_t *sock, struct timeval connected_at)
{
    struct st_h2o_accept_data_t *data;

    if (async_resumption_context.memcached.ctx != NULL) {
        data = create_memcached_accept_data(ctx, sock, connected_at);
    } else if (async_resumption_context.redis.host.len != 0) {
        data = create_redis_accept_data(ctx, sock, connected_at);
    } else {
        data = create_accept_data(ctx, sock, connected_at, sizeof(struct st_h2o_accept_data_t));
    }

    sock->data = data;
}

static void free_accept_data(struct st_h2o_accept_data_t *data)
{
    h2o_timeout_unlink(&data->timeout);
    free(data);
}

static void memcached_resumption_on_get(h2o_iovec_t session_data, void *_accept_data)
{
    struct st_h2o_memcached_resumption_accept_data_t *accept_data = _accept_data;
    accept_data->get_req = NULL;
    h2o_socket_ssl_resume_server_handshake(accept_data->super.sock, session_data);
}

static void memcached_resumption_get(h2o_socket_t *sock, h2o_iovec_t session_id)
{
    struct st_h2o_memcached_resumption_accept_data_t *data = sock->data;

    data->get_req =
        h2o_memcached_get(async_resumption_context.memcached.ctx, data->super.ctx->libmemcached_receiver, session_id, memcached_resumption_on_get,
                          data, H2O_MEMCACHED_ENCODE_KEY | H2O_MEMCACHED_ENCODE_VALUE);
}

static void memcached_resumption_new(h2o_socket_t *sock, h2o_iovec_t session_id, h2o_iovec_t session_data)
{
    h2o_memcached_set(async_resumption_context.memcached.ctx, session_id, session_data,
                      (uint32_t)time(NULL) + async_resumption_context.expiration,
                      H2O_MEMCACHED_ENCODE_KEY | H2O_MEMCACHED_ENCODE_VALUE);
}

void h2o_accept_setup_memcached_ssl_resumption(h2o_memcached_context_t *memc, unsigned expiration)
{
    async_resumption_context.memcached.ctx = memc;
    async_resumption_context.expiration = expiration;
    h2o_socket_ssl_async_resumption_init(memcached_resumption_get, memcached_resumption_new, NULL);
}

static void on_redis_connect(void)
{
    fprintf(stderr, "connected to redis at %s:%" PRIu16 "\n", async_resumption_context.redis.host.base, async_resumption_context.redis.port);
}

static void on_redis_close(const char *errstr)
{
    if (errstr == NULL) {
        fprintf(stderr, "disconnected from redis at %s:%" PRIu16 "\n", async_resumption_context.redis.host.base, async_resumption_context.redis.port);
    } else {
        fprintf(stderr, "redis connection failure: %s\n", errstr);
    }
}

static void dispose_redis_connection(void *conn)
{
    h2o_redis_free((h2o_redis_conn_t *)conn);
}

static h2o_redis_conn_t *get_redis_connection(h2o_context_t *ctx)
{
    static size_t key = SIZE_MAX;
    h2o_redis_conn_t **conn = (h2o_redis_conn_t **)h2o_context_get_storage(ctx, &key, dispose_redis_connection);
    if (*conn == NULL) {
        *conn = h2o_redis_create_connection(ctx->loop, sizeof(h2o_redis_conn_t));
        (*conn)->on_connect = on_redis_connect;
        (*conn)->on_close = on_redis_close;
    }
    return *conn;
}

#define BASE64_LENGTH(len) (((len) + 2) / 3 * 4 + 1)

static h2o_iovec_t build_redis_key(h2o_iovec_t session_id, h2o_iovec_t prefix)
{
    h2o_iovec_t key;
    key.base =  h2o_mem_alloc(prefix.len + BASE64_LENGTH(session_id.len));
    if (prefix.len != 0) {
        memcpy(key.base, prefix.base, prefix.len);
    }
    key.len = prefix.len;
    key.len += h2o_base64_encode(key.base + key.len, session_id.base, session_id.len, 1);
    return key;
}

static h2o_iovec_t build_redis_value(h2o_iovec_t session_data)
{
    h2o_iovec_t value;
    value.base =  h2o_mem_alloc(BASE64_LENGTH(session_data.len));
    value.len = h2o_base64_encode(value.base, session_data.base, session_data.len, 1);
    return value;
}

#undef BASE64_LENGTH

static void redis_resumption_on_get(redisReply *reply, void *_accept_data)
{
    struct st_h2o_redis_resumption_accept_data_t *accept_data = _accept_data;
    accept_data->get_command = NULL;

    h2o_iovec_t session_data;
    if (reply != NULL && reply->type == REDIS_REPLY_STRING) {
        session_data = h2o_decode_base64url(NULL, reply->str, reply->len);
    } else {
        session_data = h2o_iovec_init(NULL, 0);
    }

    h2o_socket_ssl_resume_server_handshake(accept_data->super.sock, session_data);

    if (session_data.base != NULL)
        free(session_data.base);
}

static void on_redis_resumption_get_failed(h2o_timeout_entry_t *timeout_entry)
{
    struct st_h2o_redis_resumption_accept_data_t *accept_data = H2O_STRUCT_FROM_MEMBER(struct st_h2o_redis_resumption_accept_data_t, super.timeout, timeout_entry);
    accept_data->get_command = NULL;
    h2o_socket_ssl_resume_server_handshake(accept_data->super.sock, h2o_iovec_init(NULL, 0));
    h2o_timeout_unlink(timeout_entry);
}

static void redis_resumption_get(h2o_socket_t *sock, h2o_iovec_t session_id)
{
    struct st_h2o_redis_resumption_accept_data_t *accept_data = sock->data;
    h2o_redis_conn_t *conn = get_redis_connection(accept_data->super.ctx->ctx);

    if (conn->state == H2O_REDIS_CONNECTION_STATE_CONNECTED) {
        h2o_iovec_t key = build_redis_key(session_id, async_resumption_context.redis.prefix);
        accept_data->get_command = h2o_redis_command(conn, redis_resumption_on_get, accept_data, "GET %s", key.base);
        free(key.base);
    } else {
        if (conn->state == H2O_REDIS_CONNECTION_STATE_CLOSED) {
            // try to connect
            h2o_redis_connect(conn, async_resumption_context.redis.host.base, async_resumption_context.redis.port);
        }
        // abort resumption
        h2o_timeout_unlink(&accept_data->super.timeout);
        accept_data->super.timeout.cb = on_redis_resumption_get_failed;
        h2o_timeout_link(accept_data->super.ctx->ctx->loop, &accept_data->super.ctx->ctx->zero_timeout, &accept_data->super.timeout);
    }
}

static void redis_resumption_new(h2o_socket_t *sock, h2o_iovec_t session_id, h2o_iovec_t session_data)
{
    struct st_h2o_redis_resumption_accept_data_t *accept_data = sock->data;
    h2o_redis_conn_t *conn = get_redis_connection(accept_data->super.ctx->ctx);

    if (conn->state == H2O_REDIS_CONNECTION_STATE_CLOSED) {
        // try to connect
        h2o_redis_connect(conn, async_resumption_context.redis.host.base, async_resumption_context.redis.port);
    }

    h2o_iovec_t key = build_redis_key(session_id, async_resumption_context.redis.prefix);
    h2o_iovec_t value = build_redis_value(session_data);
    h2o_redis_command(conn, NULL, NULL, "SETEX %s %d %s", key.base, async_resumption_context.expiration * 10, value.base);
    free(key.base);
    free(value.base);
}

void h2o_accept_setup_redis_ssl_resumption(const char *host, uint16_t port, unsigned expiration, const char *prefix)
{
    async_resumption_context.redis.host = h2o_strdup(NULL, host, SIZE_MAX);
    async_resumption_context.redis.port = port;
    async_resumption_context.redis.prefix = h2o_strdup(NULL, prefix, SIZE_MAX);
    async_resumption_context.expiration = expiration;

    h2o_socket_ssl_async_resumption_init(redis_resumption_get, redis_resumption_new, NULL);
}

static void accept_timeout(struct st_h2o_accept_data_t *data)
{
    /* TODO log */
    h2o_socket_t *sock = data->sock;
    free_accept_data(data);
    h2o_socket_close(sock);
}

static void on_accept_timeout(h2o_timeout_entry_t *entry)
{
    struct st_h2o_accept_data_t *data = H2O_STRUCT_FROM_MEMBER(struct st_h2o_accept_data_t, timeout, entry);
    accept_timeout(data);
}

static void on_redis_accept_timeout(h2o_timeout_entry_t *entry)
{
    struct st_h2o_redis_resumption_accept_data_t *data = H2O_STRUCT_FROM_MEMBER(struct st_h2o_redis_resumption_accept_data_t, super.timeout, entry);
    if (data->get_command != NULL) {
        data->get_command->cb = NULL;
        data->get_command = NULL;
    }
    accept_timeout(&data->super);
}

static void on_memcached_accept_timeout(h2o_timeout_entry_t *entry)
{
    struct st_h2o_memcached_resumption_accept_data_t *data = H2O_STRUCT_FROM_MEMBER(struct st_h2o_memcached_resumption_accept_data_t, super.timeout, entry);
    if (data->get_req != NULL) {
        h2o_memcached_cancel_get(async_resumption_context.memcached.ctx, data->get_req);
        data->get_req = NULL;
    }
    accept_timeout(&data->super);
}

static void on_ssl_handshake_complete(h2o_socket_t *sock, const char *err)
{
    struct st_h2o_accept_data_t *data = sock->data;
    sock->data = NULL;

    if (err != NULL) {
        h2o_socket_close(sock);
        goto Exit;
    }

    h2o_iovec_t proto = h2o_socket_ssl_get_selected_protocol(sock);
    const h2o_iovec_t *ident;
    for (ident = h2o_http2_alpn_protocols; ident->len != 0; ++ident) {
        if (proto.len == ident->len && memcmp(proto.base, ident->base, proto.len) == 0) {
            /* connect as http2 */
            h2o_http2_accept(data->ctx, sock, data->connected_at);
            goto Exit;
        }
    }
    /* connect as http1 */
    h2o_http1_accept(data->ctx, sock, data->connected_at);

Exit:
    free_accept_data(data);
}

static ssize_t parse_proxy_line(char *src, size_t len, struct sockaddr *sa, socklen_t *salen)
{
#define CHECK_EOF()                                                                                                                \
    if (p == end)                                                                                                                  \
    return -2
#define EXPECT_CHAR(ch)                                                                                                            \
    do {                                                                                                                           \
        CHECK_EOF();                                                                                                               \
        if (*p++ != ch)                                                                                                            \
            return -1;                                                                                                             \
    } while (0)
#define SKIP_TO_WS()                                                                                                               \
    do {                                                                                                                           \
        do {                                                                                                                       \
            CHECK_EOF();                                                                                                           \
        } while (*p++ != ' ');                                                                                                     \
        --p;                                                                                                                       \
    } while (0)

    char *p = src, *end = p + len;
    void *addr;
    in_port_t *port;

    /* "PROXY "*/
    EXPECT_CHAR('P');
    EXPECT_CHAR('R');
    EXPECT_CHAR('O');
    EXPECT_CHAR('X');
    EXPECT_CHAR('Y');
    EXPECT_CHAR(' ');

    /* "TCP[46] " */
    CHECK_EOF();
    if (*p++ != 'T') {
        *salen = 0; /* indicate that no data has been obtained */
        goto SkipToEOL;
    }
    EXPECT_CHAR('C');
    EXPECT_CHAR('P');
    CHECK_EOF();
    switch (*p++) {
    case '4':
        *salen = sizeof(struct sockaddr_in);
        memset(sa, 0, sizeof(struct sockaddr_in));
        sa->sa_family = AF_INET;
        addr = &((struct sockaddr_in *)sa)->sin_addr;
        port = &((struct sockaddr_in *)sa)->sin_port;
        break;
    case '6':
        *salen = sizeof(struct sockaddr_in6);
        memset(sa, 0, sizeof(struct sockaddr_in6));
        sa->sa_family = AF_INET6;
        addr = &((struct sockaddr_in6 *)sa)->sin6_addr;
        port = &((struct sockaddr_in6 *)sa)->sin6_port;
        break;
    default:
        return -1;
    }
    EXPECT_CHAR(' ');

    /* parse peer address */
    char *addr_start = p;
    SKIP_TO_WS();
    *p = '\0';
    if (inet_pton(sa->sa_family, addr_start, addr) != 1)
        return -1;
    *p++ = ' ';

    /* skip local address */
    SKIP_TO_WS();
    ++p;

    /* parse peer port */
    char *port_start = p;
    SKIP_TO_WS();
    *p = '\0';
    unsigned short usval;
    if (sscanf(port_start, "%hu", &usval) != 1)
        return -1;
    *port = htons(usval);
    *p++ = ' ';

SkipToEOL:
    do {
        CHECK_EOF();
    } while (*p++ != '\r');
    CHECK_EOF();
    if (*p++ != '\n')
        return -2;
    return p - src;

#undef CHECK_EOF
#undef EXPECT_CHAR
#undef SKIP_TO_WS
}

static void on_read_proxy_line(h2o_socket_t *sock, const char *err)
{
    struct st_h2o_accept_data_t *data = sock->data;

    if (err != NULL) {
        free_accept_data(data);
        h2o_socket_close(sock);
        return;
    }

    struct sockaddr_storage addr;
    socklen_t addrlen;
    ssize_t r = parse_proxy_line(sock->input->bytes, sock->input->size, (void *)&addr, &addrlen);
    switch (r) {
    case -1: /* error, just pass the input to the next handler */
        break;
    case -2: /* incomplete */
        return;
    default:
        h2o_buffer_consume(&sock->input, r);
        if (addrlen != 0)
            h2o_socket_setpeername(sock, (void *)&addr, addrlen);
        break;
    }

    if (data->ctx->ssl_ctx != NULL) {
        h2o_socket_ssl_handshake(sock, data->ctx->ssl_ctx, NULL, on_ssl_handshake_complete);
    } else {
        struct st_h2o_accept_data_t *data = sock->data;
        sock->data = NULL;
        h2o_http1_accept(data->ctx, sock, data->connected_at);
        free_accept_data(data);
    }
}

void h2o_accept(h2o_accept_ctx_t *ctx, h2o_socket_t *sock)
{
    struct timeval connected_at = *h2o_get_timestamp(ctx->ctx, NULL, NULL);

    if (ctx->expect_proxy_line || ctx->ssl_ctx != NULL) {
        setup_accept_data(ctx, sock, connected_at);
        if (ctx->expect_proxy_line) {
            h2o_socket_read_start(sock, on_read_proxy_line);
        } else {
            h2o_socket_ssl_handshake(sock, ctx->ssl_ctx, NULL, on_ssl_handshake_complete);
        }
    } else {
        h2o_http1_accept(ctx, sock, connected_at);
    }
}<commit_msg>tweak<commit_after>/*
 * Copyright (c) 2014-2016 DeNA Co., Ltd., <NAME>, <NAME>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 * IN THE SOFTWARE.
 */
#include <assert.h>
#include <inttypes.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdarg.h>
#include "h2o.h"
#include "h2o/http1.h"
#include "h2o/http2.h"

struct st_h2o_accept_data_t {
    h2o_accept_ctx_t *ctx;
    h2o_socket_t *sock;
    h2o_timeout_entry_t timeout;
    struct timeval connected_at;
};

struct st_h2o_memcached_resumption_accept_data_t {
    struct st_h2o_accept_data_t super;
    h2o_memcached_req_t *get_req;
};

struct st_h2o_redis_resumption_accept_data_t {
    struct st_h2o_accept_data_t super;
    h2o_redis_command_t *get_command;
};

static void on_accept_timeout(h2o_timeout_entry_t *entry);
static void on_redis_accept_timeout(h2o_timeout_entry_t *entry);
static void on_memcached_accept_timeout(h2o_timeout_entry_t *entry);

static struct {
    struct {
        h2o_memcached_context_t *ctx;
    } memcached;
    struct {
        h2o_iovec_t host;
        uint16_t port;
        h2o_iovec_t prefix;
    } redis;
    unsigned expiration;
} async_resumption_context;

static struct st_h2o_accept_data_t *create_accept_data(h2o_accept_ctx_t *ctx, h2o_socket_t *sock, struct timeval connected_at, size_t sz)
{
    struct st_h2o_accept_data_t *data = h2o_mem_alloc(sz);
    data->ctx = ctx;
    data->sock = sock;
    data->timeout = (h2o_timeout_entry_t){0};
    h2o_timeout_link(ctx->ctx->loop, &ctx->ctx->handshake_timeout, &data->timeout);
    data->connected_at = connected_at;
    return data;
}

static struct st_h2o_accept_data_t *create_default_accept_data(h2o_accept_ctx_t *ctx, h2o_socket_t *sock, struct timeval connected_at)
{
    struct st_h2o_accept_data_t *data = create_accept_data(ctx, sock, connected_at, sizeof(struct st_h2o_accept_data_t));
    data->timeout.cb = on_accept_timeout;
    return data;
}

static struct st_h2o_accept_data_t *create_redis_accept_data(h2o_accept_ctx_t *ctx, h2o_socket_t *sock, struct timeval connected_at)
{
    struct st_h2o_redis_resumption_accept_data_t *data =
    (struct st_h2o_redis_resumption_accept_data_t *)create_accept_data(ctx, sock, connected_at, sizeof(struct st_h2o_redis_resumption_accept_data_t));
    data->get_command = NULL;
    data->super.timeout.cb = on_redis_accept_timeout;
    return &data->super;
}

static struct st_h2o_accept_data_t *create_memcached_accept_data(h2o_accept_ctx_t *ctx, h2o_socket_t *sock, struct timeval connected_at)
{
    struct st_h2o_memcached_resumption_accept_data_t *data =
    (struct st_h2o_memcached_resumption_accept_data_t *)create_accept_data(ctx, sock, connected_at, sizeof(struct st_h2o_memcached_resumption_accept_data_t));
    data->get_req = NULL;
    data->super.timeout.cb = on_memcached_accept_timeout;
    return &data->super;
}

static void free_accept_data(struct st_h2o_accept_data_t *data)
{
    h2o_timeout_unlink(&data->timeout);
    free(data);
}

static void free_default_accept_data(struct st_h2o_accept_data_t *_accept_data)
{
    free_accept_data(_accept_data);
}

static void free_redis_accept_data(struct st_h2o_accept_data_t *_accept_data)
{
    struct st_h2o_redis_resumption_accept_data_t *accept_data = (struct st_h2o_redis_resumption_accept_data_t *)_accept_data;
    assert(accept_data->get_command == NULL);
    free_accept_data(&accept_data->super);
}

static void free_memcached_accept_data(struct st_h2o_accept_data_t *_accept_data)
{
    struct st_h2o_memcached_resumption_accept_data_t *accept_data = (struct st_h2o_memcached_resumption_accept_data_t *)_accept_data;
    assert(accept_data->get_req == NULL);
    free_accept_data(&accept_data->super);
}

static struct {
    struct st_h2o_accept_data_t *(*create)(h2o_accept_ctx_t *ctx, h2o_socket_t *sock, struct timeval connected_at);
    void (*free)(struct st_h2o_accept_data_t *accept_data);
} accept_data_callbacks = {
    create_default_accept_data,
    free_default_accept_data,
};

static void memcached_resumption_on_get(h2o_iovec_t session_data, void *_accept_data)
{
    struct st_h2o_memcached_resumption_accept_data_t *accept_data = _accept_data;
    accept_data->get_req = NULL;
    h2o_socket_ssl_resume_server_handshake(accept_data->super.sock, session_data);
}

static void memcached_resumption_get(h2o_socket_t *sock, h2o_iovec_t session_id)
{
    struct st_h2o_memcached_resumption_accept_data_t *data = sock->data;

    data->get_req =
        h2o_memcached_get(async_resumption_context.memcached.ctx, data->super.ctx->libmemcached_receiver, session_id, memcached_resumption_on_get,
                          data, H2O_MEMCACHED_ENCODE_KEY | H2O_MEMCACHED_ENCODE_VALUE);
}

static void memcached_resumption_new(h2o_socket_t *sock, h2o_iovec_t session_id, h2o_iovec_t session_data)
{
    h2o_memcached_set(async_resumption_context.memcached.ctx, session_id, session_data,
                      (uint32_t)time(NULL) + async_resumption_context.expiration,
                      H2O_MEMCACHED_ENCODE_KEY | H2O_MEMCACHED_ENCODE_VALUE);
}

void h2o_accept_setup_memcached_ssl_resumption(h2o_memcached_context_t *memc, unsigned expiration)
{
    async_resumption_context.memcached.ctx = memc;
    async_resumption_context.expiration = expiration;
    h2o_socket_ssl_async_resumption_init(memcached_resumption_get, memcached_resumption_new, NULL);
    accept_data_callbacks.create = create_memcached_accept_data;
    accept_data_callbacks.free = free_memcached_accept_data;
}

static void on_redis_connect(void)
{
    fprintf(stderr, "connected to redis at %s:%" PRIu16 "\n", async_resumption_context.redis.host.base, async_resumption_context.redis.port);
}

static void on_redis_close(const char *errstr)
{
    if (errstr == NULL) {
        fprintf(stderr, "disconnected from redis at %s:%" PRIu16 "\n", async_resumption_context.redis.host.base, async_resumption_context.redis.port);
    } else {
        fprintf(stderr, "redis connection failure: %s\n", errstr);
    }
}

static void dispose_redis_connection(void *conn)
{
    h2o_redis_free((h2o_redis_conn_t *)conn);
}

static h2o_redis_conn_t *get_redis_connection(h2o_context_t *ctx)
{
    static size_t key = SIZE_MAX;
    h2o_redis_conn_t **conn = (h2o_redis_conn_t **)h2o_context_get_storage(ctx, &key, dispose_redis_connection);
    if (*conn == NULL) {
        *conn = h2o_redis_create_connection(ctx->loop, sizeof(h2o_redis_conn_t));
        (*conn)->on_connect = on_redis_connect;
        (*conn)->on_close = on_redis_close;
    }
    return *conn;
}

#define BASE64_LENGTH(len) (((len) + 2) / 3 * 4 + 1)

static h2o_iovec_t build_redis_key(h2o_iovec_t session_id, h2o_iovec_t prefix)
{
    h2o_iovec_t key;
    key.base =  h2o_mem_alloc(prefix.len + BASE64_LENGTH(session_id.len));
    if (prefix.len != 0) {
        memcpy(key.base, prefix.base, prefix.len);
    }
    key.len = prefix.len;
    key.len += h2o_base64_encode(key.base + key.len, session_id.base, session_id.len, 1);
    return key;
}

static h2o_iovec_t build_redis_value(h2o_iovec_t session_data)
{
    h2o_iovec_t value;
    value.base =  h2o_mem_alloc(BASE64_LENGTH(session_data.len));
    value.len = h2o_base64_encode(value.base, session_data.base, session_data.len, 1);
    return value;
}

#undef BASE64_LENGTH

static void redis_resumption_on_get(redisReply *reply, void *_accept_data)
{
    struct st_h2o_redis_resumption_accept_data_t *accept_data = _accept_data;
    accept_data->get_command = NULL;

    h2o_iovec_t session_data;
    if (reply != NULL && reply->type == REDIS_REPLY_STRING) {
        session_data = h2o_decode_base64url(NULL, reply->str, reply->len);
    } else {
        session_data = h2o_iovec_init(NULL, 0);
    }

    h2o_socket_ssl_resume_server_handshake(accept_data->super.sock, session_data);

    if (session_data.base != NULL)
        free(session_data.base);
}

static void on_redis_resumption_get_failed(h2o_timeout_entry_t *timeout_entry)
{
    struct st_h2o_redis_resumption_accept_data_t *accept_data = H2O_STRUCT_FROM_MEMBER(struct st_h2o_redis_resumption_accept_data_t, super.timeout, timeout_entry);
    accept_data->get_command = NULL;
    h2o_socket_ssl_resume_server_handshake(accept_data->super.sock, h2o_iovec_init(NULL, 0));
    h2o_timeout_unlink(timeout_entry);
}

static void redis_resumption_get(h2o_socket_t *sock, h2o_iovec_t session_id)
{
    struct st_h2o_redis_resumption_accept_data_t *accept_data = sock->data;
    h2o_redis_conn_t *conn = get_redis_connection(accept_data->super.ctx->ctx);

    if (conn->state == H2O_REDIS_CONNECTION_STATE_CONNECTED) {
        h2o_iovec_t key = build_redis_key(session_id, async_resumption_context.redis.prefix);
        accept_data->get_command = h2o_redis_command(conn, redis_resumption_on_get, accept_data, "GET %s", key.base);
        free(key.base);
    } else {
        if (conn->state == H2O_REDIS_CONNECTION_STATE_CLOSED) {
            // try to connect
            h2o_redis_connect(conn, async_resumption_context.redis.host.base, async_resumption_context.redis.port);
        }
        // abort resumption
        h2o_timeout_unlink(&accept_data->super.timeout);
        accept_data->super.timeout.cb = on_redis_resumption_get_failed;
        h2o_timeout_link(accept_data->super.ctx->ctx->loop, &accept_data->super.ctx->ctx->zero_timeout, &accept_data->super.timeout);
    }
}

static void redis_resumption_new(h2o_socket_t *sock, h2o_iovec_t session_id, h2o_iovec_t session_data)
{
    struct st_h2o_redis_resumption_accept_data_t *accept_data = sock->data;
    h2o_redis_conn_t *conn = get_redis_connection(accept_data->super.ctx->ctx);

    if (conn->state == H2O_REDIS_CONNECTION_STATE_CLOSED) {
        // try to connect
        h2o_redis_connect(conn, async_resumption_context.redis.host.base, async_resumption_context.redis.port);
    }

    h2o_iovec_t key = build_redis_key(session_id, async_resumption_context.redis.prefix);
    h2o_iovec_t value = build_redis_value(session_data);
    h2o_redis_command(conn, NULL, NULL, "SETEX %s %d %s", key.base, async_resumption_context.expiration * 10, value.base);
    free(key.base);
    free(value.base);
}

void h2o_accept_setup_redis_ssl_resumption(const char *host, uint16_t port, unsigned expiration, const char *prefix)
{
    async_resumption_context.redis.host = h2o_strdup(NULL, host, SIZE_MAX);
    async_resumption_context.redis.port = port;
    async_resumption_context.redis.prefix = h2o_strdup(NULL, prefix, SIZE_MAX);
    async_resumption_context.expiration = expiration;

    h2o_socket_ssl_async_resumption_init(redis_resumption_get, redis_resumption_new, NULL);

    accept_data_callbacks.create = create_redis_accept_data;
    accept_data_callbacks.free = free_redis_accept_data;
}

static void accept_timeout(struct st_h2o_accept_data_t *data)
{
    /* TODO log */
    h2o_socket_t *sock = data->sock;
    accept_data_callbacks.free(data);
    h2o_socket_close(sock);
}

static void on_accept_timeout(h2o_timeout_entry_t *entry)
{
    struct st_h2o_accept_data_t *data = H2O_STRUCT_FROM_MEMBER(struct st_h2o_accept_data_t, timeout, entry);
    accept_timeout(data);
}

static void on_redis_accept_timeout(h2o_timeout_entry_t *entry)
{
    struct st_h2o_redis_resumption_accept_data_t *data = H2O_STRUCT_FROM_MEMBER(struct st_h2o_redis_resumption_accept_data_t, super.timeout, entry);
    if (data->get_command != NULL) {
        data->get_command->cb = NULL;
        data->get_command = NULL;
    }
    accept_timeout(&data->super);
}

static void on_memcached_accept_timeout(h2o_timeout_entry_t *entry)
{
    struct st_h2o_memcached_resumption_accept_data_t *data = H2O_STRUCT_FROM_MEMBER(struct st_h2o_memcached_resumption_accept_data_t, super.timeout, entry);
    if (data->get_req != NULL) {
        h2o_memcached_cancel_get(async_resumption_context.memcached.ctx, data->get_req);
        data->get_req = NULL;
    }
    accept_timeout(&data->super);
}

static void on_ssl_handshake_complete(h2o_socket_t *sock, const char *err)
{
    struct st_h2o_accept_data_t *data = sock->data;
    sock->data = NULL;

    if (err != NULL) {
        h2o_socket_close(sock);
        goto Exit;
    }

    h2o_iovec_t proto = h2o_socket_ssl_get_selected_protocol(sock);
    const h2o_iovec_t *ident;
    for (ident = h2o_http2_alpn_protocols; ident->len != 0; ++ident) {
        if (proto.len == ident->len && memcmp(proto.base, ident->base, proto.len) == 0) {
            /* connect as http2 */
            h2o_http2_accept(data->ctx, sock, data->connected_at);
            goto Exit;
        }
    }
    /* connect as http1 */
    h2o_http1_accept(data->ctx, sock, data->connected_at);

Exit:
    accept_data_callbacks.free(data);
}

static ssize_t parse_proxy_line(char *src, size_t len, struct sockaddr *sa, socklen_t *salen)
{
#define CHECK_EOF()                                                                                                                \
    if (p == end)                                                                                                                  \
    return -2
#define EXPECT_CHAR(ch)                                                                                                            \
    do {                                                                                                                           \
        CHECK_EOF();                                                                                                               \
        if (*p++ != ch)                                                                                                            \
            return -1;                                                                                                             \
    } while (0)
#define SKIP_TO_WS()                                                                                                               \
    do {                                                                                                                           \
        do {                                                                                                                       \
            CHECK_EOF();                                                                                                           \
        } while (*p++ != ' ');                                                                                                     \
        --p;                                                                                                                       \
    } while (0)

    char *p = src, *end = p + len;
    void *addr;
    in_port_t *port;

    /* "PROXY "*/
    EXPECT_CHAR('P');
    EXPECT_CHAR('R');
    EXPECT_CHAR('O');
    EXPECT_CHAR('X');
    EXPECT_CHAR('Y');
    EXPECT_CHAR(' ');

    /* "TCP[46] " */
    CHECK_EOF();
    if (*p++ != 'T') {
        *salen = 0; /* indicate that no data has been obtained */
        goto SkipToEOL;
    }
    EXPECT_CHAR('C');
    EXPECT_CHAR('P');
    CHECK_EOF();
    switch (*p++) {
    case '4':
        *salen = sizeof(struct sockaddr_in);
        memset(sa, 0, sizeof(struct sockaddr_in));
        sa->sa_family = AF_INET;
        addr = &((struct sockaddr_in *)sa)->sin_addr;
        port = &((struct sockaddr_in *)sa)->sin_port;
        break;
    case '6':
        *salen = sizeof(struct sockaddr_in6);
        memset(sa, 0, sizeof(struct sockaddr_in6));
        sa->sa_family = AF_INET6;
        addr = &((struct sockaddr_in6 *)sa)->sin6_addr;
        port = &((struct sockaddr_in6 *)sa)->sin6_port;
        break;
    default:
        return -1;
    }
    EXPECT_CHAR(' ');

    /* parse peer address */
    char *addr_start = p;
    SKIP_TO_WS();
    *p = '\0';
    if (inet_pton(sa->sa_family, addr_start, addr) != 1)
        return -1;
    *p++ = ' ';

    /* skip local address */
    SKIP_TO_WS();
    ++p;

    /* parse peer port */
    char *port_start = p;
    SKIP_TO_WS();
    *p = '\0';
    unsigned short usval;
    if (sscanf(port_start, "%hu", &usval) != 1)
        return -1;
    *port = htons(usval);
    *p++ = ' ';

SkipToEOL:
    do {
        CHECK_EOF();
    } while (*p++ != '\r');
    CHECK_EOF();
    if (*p++ != '\n')
        return -2;
    return p - src;

#undef CHECK_EOF
#undef EXPECT_CHAR
#undef SKIP_TO_WS
}

static void on_read_proxy_line(h2o_socket_t *sock, const char *err)
{
    struct st_h2o_accept_data_t *data = sock->data;

    if (err != NULL) {
        accept_data_callbacks.free(data);
        h2o_socket_close(sock);
        return;
    }

    struct sockaddr_storage addr;
    socklen_t addrlen;
    ssize_t r = parse_proxy_line(sock->input->bytes, sock->input->size, (void *)&addr, &addrlen);
    switch (r) {
    case -1: /* error, just pass the input to the next handler */
        break;
    case -2: /* incomplete */
        return;
    default:
        h2o_buffer_consume(&sock->input, r);
        if (addrlen != 0)
            h2o_socket_setpeername(sock, (void *)&addr, addrlen);
        break;
    }

    if (data->ctx->ssl_ctx != NULL) {
        h2o_socket_ssl_handshake(sock, data->ctx->ssl_ctx, NULL, on_ssl_handshake_complete);
    } else {
        struct st_h2o_accept_data_t *data = sock->data;
        sock->data = NULL;
        h2o_http1_accept(data->ctx, sock, data->connected_at);
        accept_data_callbacks.free(data);
    }
}

void h2o_accept(h2o_accept_ctx_t *ctx, h2o_socket_t *sock)
{
    struct timeval connected_at = *h2o_get_timestamp(ctx->ctx, NULL, NULL);

    if (ctx->expect_proxy_line || ctx->ssl_ctx != NULL) {
        sock->data = accept_data_callbacks.create(ctx, sock, connected_at);
        if (ctx->expect_proxy_line) {
            h2o_socket_read_start(sock, on_read_proxy_line);
        } else {
            h2o_socket_ssl_handshake(sock, ctx->ssl_ctx, NULL, on_ssl_handshake_complete);
        }
    } else {
        h2o_http1_accept(ctx, sock, connected_at);
    }
}
{'fraction_non_alphanumeric': 0.0984266971528118, 'fraction_numerical': 0.013582666448908487, 'mean_word_length': 2.750791220010209, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 221, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 128, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '456348', 'n_tokens_mistral': 12714, 'n_tokens_neox': 12488, 'n_words': 2297}