<commit_before>#include <unistd.h>
#include <stdlib.h>
#include <time.h>
#include "corvus.h"
#include "client.h"
#include "mbuf.h"
#include "socket.h"
#include "logging.h"
#include "event.h"

int client_trigger_event(struct connection *client, struct mbuf *buf)
{
    if (buf->pos < buf->last && !client->event_triggered) {
        if (socket_trigger_event(client->ev->fd) == CORVUS_ERR) {
            LOG(ERROR, "%s: fail to trigger readable event", __func__);
            return CORVUS_ERR;
        }
        client->event_triggered = true;
    }
    return CORVUS_OK;
}

int client_read(struct connection *client, int read_socket)
{
    struct command *cmd;
    struct mbuf *buf;
    int status = CORVUS_OK, limit = 16;

    if (!STAILQ_EMPTY(&client->info->cmd_queue)
            && STAILQ_FIRST(&client->info->cmd_queue)->parse_done)
    {
        event_reregister(&client->ctx->loop, client, E_WRITABLE);
        return CORVUS_OK;
    }

    do {
        buf = conn_get_buf(client);
        if (mbuf_read_size(buf) <= 0) {
            if (!read_socket) break;
            buf = conn_get_buf(client);
            status = conn_read(client, buf);
            if (status != CORVUS_OK) return status;
        }

        cmd = conn_get_cmd(client);
        cmd->client = client;

        status = cmd_parse_req(cmd, buf);
        if (cmd->parse_done && (--limit) < 0) {
            return client_trigger_event(client, buf);
        }
    } while (status == CORVUS_OK);

    return status;
}

void client_make_iov(struct conn_info *info)
{
    struct command *cmd;
    int64_t t = get_time();

    while (!STAILQ_EMPTY(&info->cmd_queue)) {
        cmd = STAILQ_FIRST(&info->cmd_queue);
        LOG(DEBUG, "client make iov %d %d", cmd->cmd_count, cmd->cmd_done_count);
        if (cmd->cmd_count != cmd->cmd_done_count) {
            break;
        }
        STAILQ_REMOVE_HEAD(&info->cmd_queue, cmd_next);
        STAILQ_NEXT(cmd, cmd_next) = NULL;

        /* before write */
        cmd->req_time[1] = t;

        cmd_make_iovec(cmd, &info->iov);

        cmd_stats(cmd);
        cmd_free(cmd);
    }
    LOG(DEBUG, "client make iov %d", info->iov.len);
}

int client_write(struct connection *client)
{
    struct context *ctx = client->ctx;
    struct conn_info *info = client->info;

    if (!STAILQ_EMPTY(&info->cmd_queue)) {
        client_make_iov(info);
    }

    if (info->iov.len <= 0) {
        cmd_iov_reset(&info->iov);
        return CORVUS_OK;
    }

    int status = conn_write(client, 1);

    if (status == CORVUS_ERR) {
        LOG(ERROR, "client_write: client %d fail to write iov", client->fd);
        return CORVUS_ERR;
    }
    if (status == CORVUS_AGAIN) return CORVUS_OK;

    if (info->iov.cursor >= info->iov.len) {
        cmd_iov_reset(&info->iov);
    }

    if (event_reregister(&ctx->loop, client, E_READABLE) == CORVUS_ERR) {
        LOG(ERROR, "client_write: fail to reregister client %d", client->fd);
        return CORVUS_ERR;
    }
    if (client_trigger_event(client, conn_get_buf(client)) == CORVUS_ERR) {
        LOG(ERROR, "client_write: fail to trigger event %d %d",
                client->fd, client->ev->fd);
        return CORVUS_ERR;
    }

    if (client->ctx->state == CTX_BEFORE_QUIT
            || client->ctx->state == CTX_QUITTING)
    {
        return CORVUS_ERR;
    }

    LOG(DEBUG, "client write ok");
    return CORVUS_OK;
}

void client_ready(struct connection *self, uint32_t mask)
{
    if (self->eof) {
        if (self->info->refcount <= 0) {
            conn_free(self);
            conn_recycle(self->ctx, self);
        }
        return;
    }

    self->info->last_active = time(NULL);

    if (mask & E_ERROR) {
        LOG(DEBUG, "error");<commit_msg>set writable if writing not finished<commit_after>#include <unistd.h>
#include <stdlib.h>
#include <time.h>
#include "corvus.h"
#include "client.h"
#include "mbuf.h"
#include "socket.h"
#include "logging.h"
#include "event.h"

int client_trigger_event(struct connection *client, struct mbuf *buf)
{
    if (buf->pos < buf->last && !client->event_triggered) {
        if (socket_trigger_event(client->ev->fd) == CORVUS_ERR) {
            LOG(ERROR, "%s: fail to trigger readable event", __func__);
            return CORVUS_ERR;
        }
        client->event_triggered = true;
    }
    return CORVUS_OK;
}

int client_read(struct connection *client, int read_socket)
{
    struct command *cmd;
    struct mbuf *buf;
    int status = CORVUS_OK, limit = 16;

    if (!STAILQ_EMPTY(&client->info->cmd_queue)
            && STAILQ_FIRST(&client->info->cmd_queue)->parse_done)
    {
        event_reregister(&client->ctx->loop, client, E_WRITABLE);
        return CORVUS_OK;
    }

    do {
        buf = conn_get_buf(client);
        if (mbuf_read_size(buf) <= 0) {
            if (!read_socket) break;
            buf = conn_get_buf(client);
            status = conn_read(client, buf);
            if (status != CORVUS_OK) return status;
        }

        cmd = conn_get_cmd(client);
        cmd->client = client;

        status = cmd_parse_req(cmd, buf);
        if (cmd->parse_done && (--limit) < 0) {
            return client_trigger_event(client, buf);
        }
    } while (status == CORVUS_OK);

    return status;
}

void client_make_iov(struct conn_info *info)
{
    struct command *cmd;
    int64_t t = get_time();

    while (!STAILQ_EMPTY(&info->cmd_queue)) {
        cmd = STAILQ_FIRST(&info->cmd_queue);
        LOG(DEBUG, "client make iov %d %d", cmd->cmd_count, cmd->cmd_done_count);
        if (cmd->cmd_count != cmd->cmd_done_count) {
            break;
        }
        STAILQ_REMOVE_HEAD(&info->cmd_queue, cmd_next);
        STAILQ_NEXT(cmd, cmd_next) = NULL;

        /* before write */
        cmd->req_time[1] = t;

        cmd_make_iovec(cmd, &info->iov);

        cmd_stats(cmd);
        cmd_free(cmd);
    }
    LOG(DEBUG, "client make iov %d", info->iov.len);
}

int client_write(struct connection *client)
{
    struct context *ctx = client->ctx;
    struct conn_info *info = client->info;

    if (!STAILQ_EMPTY(&info->cmd_queue)) {
        client_make_iov(info);
    }

    if (info->iov.len <= 0) {
        cmd_iov_reset(&info->iov);
        return CORVUS_OK;
    }

    int status = conn_write(client, 1);

    if (status == CORVUS_ERR) {
        LOG(ERROR, "client_write: client %d fail to write iov", client->fd);
        return CORVUS_ERR;
    }
    if (status == CORVUS_AGAIN) return CORVUS_OK;

    if (info->iov.cursor >= info->iov.len) {
        cmd_iov_reset(&info->iov);
        if (event_reregister(&ctx->loop, client, E_READABLE) == CORVUS_ERR) {
            LOG(ERROR, "client_write: fail to reregister client %d", client->fd);
            return CORVUS_ERR;
        }
        if (client_trigger_event(client, conn_get_buf(client)) == CORVUS_ERR) {
            LOG(ERROR, "client_write: fail to trigger event %d %d",
                    client->fd, client->ev->fd);
            return CORVUS_ERR;
        }
    } else if (event_reregister(&ctx->loop, client, E_WRITABLE) == CORVUS_ERR) {
        LOG(ERROR, "client_write: fail to reregister client %d", client->fd);
        return CORVUS_ERR;
    }

    if (client->ctx->state == CTX_BEFORE_QUIT
            || client->ctx->state == CTX_QUITTING)
    {
        return CORVUS_ERR;
    }

    LOG(DEBUG, "client write ok");
    return CORVUS_OK;
}

void client_ready(struct connection *self, uint32_t mask)
{
    if (self->eof) {
        if (self->info->refcount <= 0) {
            conn_free(self);
            conn_recycle(self->ctx, self);
        }
        return;
    }

    self->info->last_active = time(NULL);

    if (mask & E_ERROR) {
        LOG(DEBUG, "error");
{'fraction_non_alphanumeric': 0.13867949051208733, 'fraction_numerical': 0.0031193137509747857, 'mean_word_length': 2.429144385026738, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 111, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4365152', 'n_tokens_mistral': 2735, 'n_tokens_neox': 2655, 'n_words': 556}