<commit_before>/*
 * Copyright 2016 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <assert.h>
#include <math.h>
#include <netinet/in.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <sys/resource.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include "common.h"
#include "flow.h"
#include "interval.h"
#include "lib.h"
#include "numlist.h"
#include "percentiles.h"
#include "sample.h"
#include "thread.h"
#include "workload.h"

static inline void track_write_time(struct options *opts, struct flow *flow)
{
        if (flow->bytes_to_write == opts->request_size)
                clock_gettime(CLOCK_MONOTONIC, &flow->write_time);
}

static inline void track_finish_time(struct flow *flow)
{
        struct timespec finish_time;

        clock_gettime(CLOCK_MONOTONIC, &finish_time);
        numlist_add(flow->latency, seconds_between(&flow->write_time,
                                                   &finish_time));
}

static void client_events(struct thread *t, int epfd,
                          struct epoll_event *events, int nfds, char *buf)
{
        struct script_slave *ss = t->script_slave;
        struct options *opts = t->opts;
        struct callbacks *cb = t->cb;
        struct flow *flow;
        ssize_t num_bytes;
        int i;

        for (i = 0; i < nfds; i++) {
                flow = events[i].data.ptr;
                if (flow->fd == t->stop_efd) {
                        t->stop = 1;
                        break;
                }
                if (events[i].events & EPOLLRDHUP) {
                        delflow(t->index, epfd, flow, cb);
                        continue;
                }
                if (events[i].events & EPOLLOUT) {
                        ssize_t to_write = flow->bytes_to_write;
                        int flags = 0;

                        if (to_write > opts->buffer_size) {
                                to_write = opts->buffer_size;
                                flags |= MSG_MORE;
                        }
                        track_write_time(opts, flow);
                        num_bytes = do_write(ss, flow->fd, buf, to_write, flags);
                        if (num_bytes == -1) {
                                PLOG_ERROR(cb, "write");
                                continue;
                        }
                        flow->bytes_to_write -= num_bytes;
                        if (flow->bytes_to_write > 0)
                                continue;
                        /* Successfully sent request, now wait for response */
                        events[i].events = EPOLLRDHUP | EPOLLIN;
                        epoll_ctl_or_die(epfd, EPOLL_CTL_MOD, flow->fd,
                                         &events[i], cb);
                        flow->bytes_to_read = opts->response_size;
                } else if (events[i].events & EPOLLIN) {
                        ssize_t to_read = flow->bytes_to_read;

                        if (to_read > opts->buffer_size)
                                to_read = opts->buffer_size;
                        num_bytes = do_read(ss, flow->fd, buf, to_read, 0);
                        if (num_bytes == -1) {
                                PLOG_ERROR(cb, "read");
                                continue;
                        }
                        if (num_bytes == 0) {
                                delflow(t->index, epfd, flow, cb);
                                continue;
                        }
                        flow->bytes_read += num_bytes;
                        flow->bytes_to_read -= num_bytes;
                        if (flow->bytes_to_read > 0)
                                continue;
                        t->transactions++;
                        flow->transactions++;
                        track_finish_time(flow);
                        interval_collect(flow, t);
                        /* Successfully read resp., now wait to send request */
                        events[i].events = EPOLLRDHUP | EPOLLOUT;
                        epoll_ctl_or_die(epfd, EPOLL_CTL_MOD, flow->fd,
                                         &events[i], cb);
                        flow->bytes_to_write = opts->request_size;
                }
        }
}

static int client_connect(int i, int epfd, struct thread *t)
{
        struct script_slave *ss = t->script_slave;
        struct options *opts = t->opts;
        struct callbacks *cb = t->cb;
        struct addrinfo *ai = t->ai;
        struct flow *flow;
        int fd;

        fd = do_socket_open(ss, ai);
        if (fd == -1) {
                PLOG_FATAL(cb, "socket");
                return fd;
        }

        if (opts->min_rto)
                set_min_rto(fd, opts->min_rto, cb);
        if (opts->debug)
                set_debug(fd, 1, cb);
        if (opts->local_host)
                set_local_host(fd, opts, cb);
        if (do_connect(fd, ai->ai_addr, ai->ai_addrlen))
                PLOG_FATAL(cb, "do_connect");

        flow = addflow(t->index, epfd, fd, i, EPOLLOUT, opts, cb);
        flow->bytes_to_write = opts->request_size;
        flow->itv = interval_create(opts->interval, t);

        return fd;
}

/**
 * The function expects @fd_listen is in a "ready" state in the @epfd
 * epoll set, and directly calls accept() on @fd_listen. The readiness
 * should guarantee that the accept() doesn't block.
 *
 * After a client socket fd is obtained, a new flow is created as part
 * of the thread @t.  The state of the flow is set to "waiting for a
 * request".
 */
static void server_accept(int fd_listen, int epfd, struct thread *t)
{
        struct options *opts = t->opts;
        struct callbacks *cb = t->cb;
        struct sockaddr_storage cli_addr;
        struct flow *flow;
        socklen_t cli_len;
        int client;

        cli_len = sizeof(cli_addr);
        client = accept(fd_listen, (struct sockaddr *)&cli_addr, &cli_len);
        if (client == -1) {
                if (errno == EINTR || errno == ECONNABORTED)
                        return;
                PLOG_ERROR(cb, "accept");
                return;
        }
        flow = addflow(t->index, epfd, client, t->next_flow_id++,
                       EPOLLIN, opts, cb);
        flow->bytes_to_read = opts->request_size;
        flow->itv = interval_create(opts->interval, t);
}

static void run_client(struct thread *t)
{
        struct script_slave *ss = t->script_slave;
        struct options *opts = t->opts;
        const int flows_in_this_thread = flows_in_thread(opts->num_flows,
                                                         opts->num_threads,
                                                         t->index);
        struct callbacks *cb = t->cb;
        struct addrinfo *ai = t->ai;
        struct epoll_event *events;
        struct flow *stop_fl;
        int epfd, i;
        char *buf;
        CLEANUP(free) int *client_fds = NULL;

        client_fds = calloc(flows_in_this_thread, sizeof(int));
        if (!client_fds)
                PLOG_FATAL(cb, "alloc client_fds array");

        LOG_INFO(cb, "flows_in_this_thread=%d", flows_in_this_thread);
        epfd = epoll_create1(0);
        if (epfd == -1)
                PLOG_FATAL(cb, "epoll_create1");
        stop_fl = addflow_lite(epfd, t->stop_efd, EPOLLIN, cb);
        for (i = 0; i < flows_in_this_thread; i++)
                client_fds[i] = client_connect(i, epfd, t);
        events = calloc(opts->maxevents, sizeof(struct epoll_event));
        buf = buf_alloc(opts);
        if (!buf)
                PLOG_FATAL(cb, "buf_alloc");
        pthread_barrier_wait(t->ready);
        while (!t->stop) {
                int ms = opts->nonblocking ? 10 /* milliseconds */ : -1;
                int nfds = epoll_wait(epfd, events, opts->maxevents, ms);
                if (nfds == -1) {
                        if (errno == EINTR)
                                continue;
                        PLOG_FATAL(cb, "epoll_wait");
                }
                client_events(t, epfd, events, nfds, buf);
        }

        for (i = 0; i < flows_in_this_thread; i++) {
                if (do_socket_close(ss, client_fds[i], ai) < 0)
                        /* PLOG_FATAL(cb, "close"); */
                        /* XXX: ignore errors */ ;
        }

        free(buf);
        free(events);
        free(stop_fl);
        do_close(epfd);<commit_msg>tcp_rr: Pull up client flow initialization<commit_after>/*
 * Copyright 2016 Google Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <assert.h>
#include <math.h>
#include <netinet/in.h>
#include <pthread.h>
#include <stdbool.h>
#include <stdio.h>
#include <sys/resource.h>
#include <sys/time.h>
#include <time.h>
#include <unistd.h>
#include "common.h"
#include "flow.h"
#include "interval.h"
#include "lib.h"
#include "numlist.h"
#include "percentiles.h"
#include "sample.h"
#include "thread.h"
#include "workload.h"

static inline void track_write_time(struct options *opts, struct flow *flow)
{
        if (flow->bytes_to_write == opts->request_size)
                clock_gettime(CLOCK_MONOTONIC, &flow->write_time);
}

static inline void track_finish_time(struct flow *flow)
{
        struct timespec finish_time;

        clock_gettime(CLOCK_MONOTONIC, &finish_time);
        numlist_add(flow->latency, seconds_between(&flow->write_time,
                                                   &finish_time));
}

static void client_events(struct thread *t, int epfd,
                          struct epoll_event *events, int nfds, char *buf)
{
        struct script_slave *ss = t->script_slave;
        struct options *opts = t->opts;
        struct callbacks *cb = t->cb;
        struct flow *flow;
        ssize_t num_bytes;
        int i;

        for (i = 0; i < nfds; i++) {
                flow = events[i].data.ptr;
                if (flow->fd == t->stop_efd) {
                        t->stop = 1;
                        break;
                }
                if (events[i].events & EPOLLRDHUP) {
                        delflow(t->index, epfd, flow, cb);
                        continue;
                }
                if (events[i].events & EPOLLOUT) {
                        ssize_t to_write = flow->bytes_to_write;
                        int flags = 0;

                        if (to_write > opts->buffer_size) {
                                to_write = opts->buffer_size;
                                flags |= MSG_MORE;
                        }
                        track_write_time(opts, flow);
                        num_bytes = do_write(ss, flow->fd, buf, to_write, flags);
                        if (num_bytes == -1) {
                                PLOG_ERROR(cb, "write");
                                continue;
                        }
                        flow->bytes_to_write -= num_bytes;
                        if (flow->bytes_to_write > 0)
                                continue;
                        /* Successfully sent request, now wait for response */
                        events[i].events = EPOLLRDHUP | EPOLLIN;
                        epoll_ctl_or_die(epfd, EPOLL_CTL_MOD, flow->fd,
                                         &events[i], cb);
                        flow->bytes_to_read = opts->response_size;
                } else if (events[i].events & EPOLLIN) {
                        ssize_t to_read = flow->bytes_to_read;

                        if (to_read > opts->buffer_size)
                                to_read = opts->buffer_size;
                        num_bytes = do_read(ss, flow->fd, buf, to_read, 0);
                        if (num_bytes == -1) {
                                PLOG_ERROR(cb, "read");
                                continue;
                        }
                        if (num_bytes == 0) {
                                delflow(t->index, epfd, flow, cb);
                                continue;
                        }
                        flow->bytes_read += num_bytes;
                        flow->bytes_to_read -= num_bytes;
                        if (flow->bytes_to_read > 0)
                                continue;
                        t->transactions++;
                        flow->transactions++;
                        track_finish_time(flow);
                        interval_collect(flow, t);
                        /* Successfully read resp., now wait to send request */
                        events[i].events = EPOLLRDHUP | EPOLLOUT;
                        epoll_ctl_or_die(epfd, EPOLL_CTL_MOD, flow->fd,
                                         &events[i], cb);
                        flow->bytes_to_write = opts->request_size;
                }
        }
}

static int client_connect(struct thread *t)
{
        struct script_slave *ss = t->script_slave;
        struct options *opts = t->opts;
        struct callbacks *cb = t->cb;
        struct addrinfo *ai = t->ai;
        int fd;

        fd = do_socket_open(ss, ai);
        if (fd == -1) {
                PLOG_FATAL(cb, "socket");
                return fd;
        }

        if (opts->min_rto)
                set_min_rto(fd, opts->min_rto, cb);
        if (opts->debug)
                set_debug(fd, 1, cb);
        if (opts->local_host)
                set_local_host(fd, opts, cb);
        if (do_connect(fd, ai->ai_addr, ai->ai_addrlen))
                PLOG_FATAL(cb, "do_connect");

        return fd;
}

/**
 * The function expects @fd_listen is in a "ready" state in the @epfd
 * epoll set, and directly calls accept() on @fd_listen. The readiness
 * should guarantee that the accept() doesn't block.
 *
 * After a client socket fd is obtained, a new flow is created as part
 * of the thread @t.  The state of the flow is set to "waiting for a
 * request".
 */
static void server_accept(int fd_listen, int epfd, struct thread *t)
{
        struct options *opts = t->opts;
        struct callbacks *cb = t->cb;
        struct sockaddr_storage cli_addr;
        struct flow *flow;
        socklen_t cli_len;
        int client;

        cli_len = sizeof(cli_addr);
        client = accept(fd_listen, (struct sockaddr *)&cli_addr, &cli_len);
        if (client == -1) {
                if (errno == EINTR || errno == ECONNABORTED)
                        return;
                PLOG_ERROR(cb, "accept");
                return;
        }
        flow = addflow(t->index, epfd, client, t->next_flow_id++,
                       EPOLLIN, opts, cb);
        flow->bytes_to_read = opts->request_size;
        flow->itv = interval_create(opts->interval, t);
}

static void run_client(struct thread *t)
{
        struct script_slave *ss = t->script_slave;
        struct options *opts = t->opts;
        const int flows_in_this_thread = flows_in_thread(opts->num_flows,
                                                         opts->num_threads,
                                                         t->index);
        struct callbacks *cb = t->cb;
        struct addrinfo *ai = t->ai;
        struct epoll_event *events;
        struct flow *flow, *stop_fl;
        int epfd, fd, i;
        char *buf;
        CLEANUP(free) int *client_fds = NULL;

        client_fds = calloc(flows_in_this_thread, sizeof(int));
        if (!client_fds)
                PLOG_FATAL(cb, "alloc client_fds array");

        LOG_INFO(cb, "flows_in_this_thread=%d", flows_in_this_thread);
        epfd = epoll_create1(0);
        if (epfd == -1)
                PLOG_FATAL(cb, "epoll_create1");
        stop_fl = addflow_lite(epfd, t->stop_efd, EPOLLIN, cb);
        for (i = 0; i < flows_in_this_thread; i++) {
                fd = client_connect(t);

                flow = addflow(t->index, epfd, fd, i, EPOLLOUT, opts, cb);
                flow->bytes_to_write = opts->request_size;
                flow->itv = interval_create(opts->interval, t);

                client_fds[i] = fd;
                /* flow will be deleted by client_events() */
        }

        events = calloc(opts->maxevents, sizeof(struct epoll_event));
        buf = buf_alloc(opts);
        if (!buf)
                PLOG_FATAL(cb, "buf_alloc");
        pthread_barrier_wait(t->ready);
        while (!t->stop) {
                int ms = opts->nonblocking ? 10 /* milliseconds */ : -1;
                int nfds = epoll_wait(epfd, events, opts->maxevents, ms);
                if (nfds == -1) {
                        if (errno == EINTR)
                                continue;
                        PLOG_FATAL(cb, "epoll_wait");
                }
                client_events(t, epfd, events, nfds, buf);
        }

        for (i = 0; i < flows_in_this_thread; i++) {
                if (do_socket_close(ss, client_fds[i], ai) < 0)
                        /* PLOG_FATAL(cb, "close"); */
                        /* XXX: ignore errors */ ;
        }

        free(buf);
        free(events);
        free(stop_fl);
        do_close(epfd);
{'fraction_non_alphanumeric': 0.10612290254472719, 'fraction_numerical': 0.003444827203022558, 'mean_word_length': 1.42835941716136, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 171, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5460429', 'n_tokens_mistral': 5333, 'n_tokens_neox': 5081, 'n_words': 1305}