<commit_before>/*
* kinetic-c-client
* Copyright (C) 2014 Seagate Technology.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
*/
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <poll.h>
#include <errno.h>
#include <assert.h>
#include <limits.h>
#include <sys/resource.h>

#include "bus.h"
#include "sender.h"
#include "listener.h"
#include "threadpool.h"
#include "bus_internal_types.h"
#include "bus_ssl.h"
#include "util.h"
#include "yacht.h"

/* Function pointers for pthreads. */
void *listener_mainloop(void *arg);
void *sender_mainloop(void *arg);

static bool poll_on_completion(struct bus *b, int fd);
static int sender_id_of_socket(struct bus *b, int fd);
static int listener_id_of_socket(struct bus *b, int fd);
static void noop_log_cb(log_event_t event,
        int log_level, const char *msg, void *udata);
static void noop_error_cb(bus_unpack_cb_res_t result, void *socket_udata);
static bool attempt_to_increase_resource_limits(struct bus *b);

static void set_defaults(bus_config *cfg) {
    if (cfg->sender_count == 0) { cfg->sender_count = 1; }
    if (cfg->listener_count == 0) { cfg->listener_count = 1; }
}

#define DEF_FD_SET_SIZE2 4

bool bus_init(bus_config *config, struct bus_result *res) {
    if (res == NULL) { return false; }
    if (config == NULL) {
        res->status = BUS_INIT_ERROR_NULL;
        return false;
    }
    set_defaults(config);
    if (config->sink_cb == NULL) {
        res->status = BUS_INIT_ERROR_MISSING_SINK_CB;
        return false;
    }
    if (config->unpack_cb == NULL) {
        res->status = BUS_INIT_ERROR_MISSING_UNPACK_CB;
        return false;
    }
    if (config->log_cb == NULL) {
        config->log_cb = noop_log_cb;
        config->log_level = INT_MIN;
    }
    if (config->error_cb == NULL) {
        config->error_cb = noop_error_cb;
    }

    res->status = BUS_INIT_ERROR_ALLOC_FAIL;

    bool log_lock_init = false;
    struct sender **ss = NULL;
    struct listener **ls = NULL;
    struct threadpool *tp = NULL;
    bool *j = NULL;
    pthread_t *pts = NULL;
    struct yacht *fd_set = NULL;

    bus *b = calloc(1, sizeof(*b));
    if (b == NULL) { goto cleanup; }

    if (!bus_ssl_init(b)) { goto cleanup; }

    b->sink_cb = config->sink_cb;
    b->unpack_cb = config->unpack_cb;
    b->unexpected_msg_cb = config->unexpected_msg_cb;
    b->error_cb = config->error_cb;
    b->log_cb = config->log_cb;
    b->log_level = config->log_level;
    b->udata = config->bus_udata;
    if (0 != pthread_mutex_init(&b->log_lock, NULL)) {
        res->status = BUS_INIT_ERROR_MUTEX_INIT_FAIL;
        goto cleanup;
    }
    if (0 != pthread_mutex_init(&b->fd_set_lock, NULL)) {
        res->status = BUS_INIT_ERROR_MUTEX_INIT_FAIL;
        goto cleanup;
    }

    log_lock_init = true;

    attempt_to_increase_resource_limits(b);

    BUS_LOG_SNPRINTF(b, 3, LOG_INITIALIZATION, b->udata, 64,
        "Initialized bus at %p", (void*)b);

    ss = calloc(config->sender_count, sizeof(*ss));
    if (ss == NULL) {
        goto cleanup;
    }

    for (int i = 0; i < config->sender_count; i++) {
        ss[i] = sender_init(b, config);
        if (ss[i] == NULL) {
            res->status = BUS_INIT_ERROR_SENDER_INIT_FAIL;
            goto cleanup;
        } else {
            BUS_LOG_SNPRINTF(b, 3, LOG_INITIALIZATION, b->udata, 64,
                "Initialized sender %d at %p", i, (void*)ss[i]);
        }
    }

    ls = calloc(config->listener_count, sizeof(*ls));
    if (ls == NULL) {
        goto cleanup;
    }

    for (int i = 0; i < config->listener_count; i++) {
        ls[i] = listener_init(b, config);
        if (ls[i] == NULL) {
            res->status = BUS_INIT_ERROR_LISTENER_INIT_FAIL;
            goto cleanup;
        } else {
            BUS_LOG_SNPRINTF(b, 3, LOG_INITIALIZATION, b->udata, 64,
                "Initialized listener %d at %p", i, (void*)ls[i]);
        }
    }

    tp = threadpool_init(&config->threadpool_cfg);
    if (tp == NULL) {
        res->status = BUS_INIT_ERROR_THREADPOOL_INIT_FAIL;
        goto cleanup;
    }

    int thread_count = config->sender_count + config->listener_count;
    j = calloc(thread_count, sizeof(bool));
    pts = calloc(thread_count, sizeof(pthread_t));
    if (j == NULL || pts == NULL) {
        goto cleanup;
    }

    fd_set = yacht_init(DEF_FD_SET_SIZE2);
    if (fd_set == NULL) {
        goto cleanup;
    }

    b->sender_count = config->sender_count;
    b->senders = ss;
    b->listener_count = config->listener_count;
    b->listeners = ls;
    b->threadpool = tp;
    b->joined = j;
    b->threads = pts;

    for (int i = 0; i < b->sender_count; i++) {
        int pcres = pthread_create(&b->threads[i], NULL,
            sender_mainloop, (void *)b->senders[i]);
        if (pcres != 0) {
            res->status = BUS_INIT_ERROR_PTHREAD_INIT_FAIL;
            goto cleanup;
        }
    }

    for (int i = 0; i < b->listener_count; i++) {
        int pcres = pthread_create(&b->threads[i + b->sender_count], NULL,
            listener_mainloop, (void *)b->listeners[i]);
        if (pcres != 0) {
            res->status = BUS_INIT_ERROR_PTHREAD_INIT_FAIL;
            goto cleanup;
        }
    }

    b->fd_set = fd_set;
    res->bus = b;
    BUS_LOG(b, 2, LOG_INITIALIZATION, "initialized", config->bus_udata);
    return true;

cleanup:
    if (ss) {
        for (int i = 0; i < config->sender_count; i++) {
            if (ss[i]) { sender_free(ss[i]); }
        }
        free(ss);
    }
    if (ls) {
        for (int i = 0; i < config->listener_count; i++) {
            if (ls[i]) { listener_free(ls[i]); }
        }
        free(ls);
    }
    if (tp) { threadpool_free(tp); }
    if (j) { free(j); }
    if (b) {
        if (log_lock_init) {
            pthread_mutex_destroy(&b->fd_set_lock);
            pthread_mutex_destroy(&b->log_lock);
        }
        free(b);
    }

    if (pts) { free(pts); }
    if (fd_set) { yacht_free(fd_set, NULL, NULL); }

    return false;
}

static bool attempt_to_increase_resource_limits(struct bus *b) {
    struct rlimit info;
    if (-1 == getrlimit(RLIMIT_NOFILE, &info)) {
        fprintf(stderr, "getrlimit: %s", strerror(errno));
        errno = 0;
        return false;
    }

    const unsigned int nval = 1024;

    if (info.rlim_cur < nval && info.rlim_max > nval) {
        info.rlim_cur = nval;
        BUS_LOG_SNPRINTF(b, 3, LOG_MEMORY, b->udata, 256,
            "Current FD resource limits, [%lu, %lu], changing to %u",
            (unsigned long)info.rlim_cur, (unsigned long)info.rlim_max, nval);
        if (-1 == setrlimit(RLIMIT_NOFILE, &info)) {
            BUS_LOG_SNPRINTF(b, 3, LOG_MEMORY, b->udata, 256,
                "Failed to increase FD resource limit to %u, %s",
                nval, strerror(errno));
            fprintf(stderr, "getrlimit: %s", strerror(errno));
            errno = 0;
            return false;
        } else {
            BUS_LOG_SNPRINTF(b, 3, LOG_MEMORY, b->udata, 256,
                "Successfully increased FD resource limit to %u", nval);
        }
    } else {
        BUS_LOG_SNPRINTF(b, 3, LOG_MEMORY, b->udata, 256,
            "Current FD resource limits [%lu, %lu] are acceptable",
            (unsigned long)info.rlim_cur, (unsigned long)info.rlim_max);
    }
    return true;
}

/* Pack message to deliver on behalf of the user into an envelope
 * that can track status / routing along the way.
 *
 * The box should only ever be accessible on a single thread at a time. */
static boxed_msg *box_msg(struct bus *b, bus_user_msg *msg) {
    boxed_msg *box = calloc(1, sizeof(*box));
    if (box == NULL) { return NULL; }

    BUS_LOG_SNPRINTF(b, 3, LOG_MEMORY, b->udata, 64,
        "Allocated boxed message -- %p", (void*)box);

    box->fd = msg->fd;
    assert(msg->fd != 0);

    /* Lock hash table and check whether this FD uses SSL. */
    if (0 != pthread_mutex_lock(&b->fd_set_lock)) { assert(false); }
    void *value = NULL;
    SSL *ssl = NULL;
    if (yacht_get(b->fd_set, box->fd, &value)) {
        ssl = (SSL *)value;
        assert(ssl != NULL);
        box->ssl = ssl;
    }
    if (0 != pthread_mutex_unlock(&b->fd_set_lock)) { assert(false); }

    if (ssl == NULL) {
        /* socket isn't registered, fail out */
        free(box);
        return NULL;
    }

    box->timeout_sec = (time_t)msg->timeout_sec;
    if (box->timeout_sec == 0) {
        box->timeout_sec = BUS_DEFAULT_TIMEOUT_SEC;
    }

    box->out_seq_id = msg->seq_id;
    box->out_msg_size = msg->msg_size;

    /* Store message by pointer, since the client thread using it is blocked
     * until we are done sending. */
    box->out_msg = msg->msg;

    box->cb = msg->cb;
    box->udata = msg->udata;
    return box;
}

bool bus_send_request(struct bus *b, bus_user_msg *msg)
{
    if (b == NULL || msg == NULL || msg->fd == 0) {
        return false;
    }

    boxed_msg *box = box_msg(b, msg);
    if (box == NULL) {
        return false;
    }

    int s_id = sender_id_of_socket(b, msg->fd);
    struct sender *s = b->senders[s_id];

    BUS_LOG_SNPRINTF(b, 3-3, LOG_SENDING_REQUEST, b->udata, 64,
        "Sending request <fd:%d, seq_id:%lld>", msg->fd, (long long)msg->seq_id);
    bool res = sender_send_request(s, box);
    BUS_LOG_SNPRINTF(b, 3, LOG_SENDING_REQUEST, b->udata, 64,
        "...request sent, result %d", res);
    return res;
}

static bool poll_on_completion(struct bus *b, int fd) {
    /* POLL in a pipe */
    struct pollfd fds[1];
    fds[0].fd = fd;
    fds[0].events = POLLIN;

    /* TODO REFACTOR this should be reused between bus, sender, and listener,
     *     or be moved into the listener.
     *     The sender has its own blocking polling for commands. */

    /* FIXME: compare this to TCP timeouts -- try to prevent sender
     * succeeding but failing to notify client. */
    const int TIMEOUT_SECONDS = 10;
    const int ONE_SECOND = 1000;  // msec

    for (int i = 0; i < TIMEOUT_SECONDS; i++) {
        BUS_LOG(b, 5, LOG_SENDING_REQUEST, "Polling on completion...tick...", b->udata);
        int res = poll(fds, 1, ONE_SECOND);
        if (res == -1) {
            if (util_is_resumable_io_error(errno)) {
                BUS_LOG(b, 3, LOG_SENDING_REQUEST, "Polling on completion...EAGAIN", b->udata);
                if (errno == EINTR && i > 0) { i--; }<commit_msg>Reduce log level that was elevated for debugging<commit_after>/*
* kinetic-c-client
* Copyright (C) 2014 Seagate Technology.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
*
*/
#include <stdlib.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <poll.h>
#include <errno.h>
#include <assert.h>
#include <limits.h>
#include <sys/resource.h>

#include "bus.h"
#include "sender.h"
#include "listener.h"
#include "threadpool.h"
#include "bus_internal_types.h"
#include "bus_ssl.h"
#include "util.h"
#include "yacht.h"

/* Function pointers for pthreads. */
void *listener_mainloop(void *arg);
void *sender_mainloop(void *arg);

static bool poll_on_completion(struct bus *b, int fd);
static int sender_id_of_socket(struct bus *b, int fd);
static int listener_id_of_socket(struct bus *b, int fd);
static void noop_log_cb(log_event_t event,
        int log_level, const char *msg, void *udata);
static void noop_error_cb(bus_unpack_cb_res_t result, void *socket_udata);
static bool attempt_to_increase_resource_limits(struct bus *b);

static void set_defaults(bus_config *cfg) {
    if (cfg->sender_count == 0) { cfg->sender_count = 1; }
    if (cfg->listener_count == 0) { cfg->listener_count = 1; }
}

#define DEF_FD_SET_SIZE2 4

bool bus_init(bus_config *config, struct bus_result *res) {
    if (res == NULL) { return false; }
    if (config == NULL) {
        res->status = BUS_INIT_ERROR_NULL;
        return false;
    }
    set_defaults(config);
    if (config->sink_cb == NULL) {
        res->status = BUS_INIT_ERROR_MISSING_SINK_CB;
        return false;
    }
    if (config->unpack_cb == NULL) {
        res->status = BUS_INIT_ERROR_MISSING_UNPACK_CB;
        return false;
    }
    if (config->log_cb == NULL) {
        config->log_cb = noop_log_cb;
        config->log_level = INT_MIN;
    }
    if (config->error_cb == NULL) {
        config->error_cb = noop_error_cb;
    }

    res->status = BUS_INIT_ERROR_ALLOC_FAIL;

    bool log_lock_init = false;
    struct sender **ss = NULL;
    struct listener **ls = NULL;
    struct threadpool *tp = NULL;
    bool *j = NULL;
    pthread_t *pts = NULL;
    struct yacht *fd_set = NULL;

    bus *b = calloc(1, sizeof(*b));
    if (b == NULL) { goto cleanup; }

    if (!bus_ssl_init(b)) { goto cleanup; }

    b->sink_cb = config->sink_cb;
    b->unpack_cb = config->unpack_cb;
    b->unexpected_msg_cb = config->unexpected_msg_cb;
    b->error_cb = config->error_cb;
    b->log_cb = config->log_cb;
    b->log_level = config->log_level;
    b->udata = config->bus_udata;
    if (0 != pthread_mutex_init(&b->log_lock, NULL)) {
        res->status = BUS_INIT_ERROR_MUTEX_INIT_FAIL;
        goto cleanup;
    }
    if (0 != pthread_mutex_init(&b->fd_set_lock, NULL)) {
        res->status = BUS_INIT_ERROR_MUTEX_INIT_FAIL;
        goto cleanup;
    }

    log_lock_init = true;

    attempt_to_increase_resource_limits(b);

    BUS_LOG_SNPRINTF(b, 3, LOG_INITIALIZATION, b->udata, 64,
        "Initialized bus at %p", (void*)b);

    ss = calloc(config->sender_count, sizeof(*ss));
    if (ss == NULL) {
        goto cleanup;
    }

    for (int i = 0; i < config->sender_count; i++) {
        ss[i] = sender_init(b, config);
        if (ss[i] == NULL) {
            res->status = BUS_INIT_ERROR_SENDER_INIT_FAIL;
            goto cleanup;
        } else {
            BUS_LOG_SNPRINTF(b, 3, LOG_INITIALIZATION, b->udata, 64,
                "Initialized sender %d at %p", i, (void*)ss[i]);
        }
    }

    ls = calloc(config->listener_count, sizeof(*ls));
    if (ls == NULL) {
        goto cleanup;
    }

    for (int i = 0; i < config->listener_count; i++) {
        ls[i] = listener_init(b, config);
        if (ls[i] == NULL) {
            res->status = BUS_INIT_ERROR_LISTENER_INIT_FAIL;
            goto cleanup;
        } else {
            BUS_LOG_SNPRINTF(b, 3, LOG_INITIALIZATION, b->udata, 64,
                "Initialized listener %d at %p", i, (void*)ls[i]);
        }
    }

    tp = threadpool_init(&config->threadpool_cfg);
    if (tp == NULL) {
        res->status = BUS_INIT_ERROR_THREADPOOL_INIT_FAIL;
        goto cleanup;
    }

    int thread_count = config->sender_count + config->listener_count;
    j = calloc(thread_count, sizeof(bool));
    pts = calloc(thread_count, sizeof(pthread_t));
    if (j == NULL || pts == NULL) {
        goto cleanup;
    }

    fd_set = yacht_init(DEF_FD_SET_SIZE2);
    if (fd_set == NULL) {
        goto cleanup;
    }

    b->sender_count = config->sender_count;
    b->senders = ss;
    b->listener_count = config->listener_count;
    b->listeners = ls;
    b->threadpool = tp;
    b->joined = j;
    b->threads = pts;

    for (int i = 0; i < b->sender_count; i++) {
        int pcres = pthread_create(&b->threads[i], NULL,
            sender_mainloop, (void *)b->senders[i]);
        if (pcres != 0) {
            res->status = BUS_INIT_ERROR_PTHREAD_INIT_FAIL;
            goto cleanup;
        }
    }

    for (int i = 0; i < b->listener_count; i++) {
        int pcres = pthread_create(&b->threads[i + b->sender_count], NULL,
            listener_mainloop, (void *)b->listeners[i]);
        if (pcres != 0) {
            res->status = BUS_INIT_ERROR_PTHREAD_INIT_FAIL;
            goto cleanup;
        }
    }

    b->fd_set = fd_set;
    res->bus = b;
    BUS_LOG(b, 2, LOG_INITIALIZATION, "initialized", config->bus_udata);
    return true;

cleanup:
    if (ss) {
        for (int i = 0; i < config->sender_count; i++) {
            if (ss[i]) { sender_free(ss[i]); }
        }
        free(ss);
    }
    if (ls) {
        for (int i = 0; i < config->listener_count; i++) {
            if (ls[i]) { listener_free(ls[i]); }
        }
        free(ls);
    }
    if (tp) { threadpool_free(tp); }
    if (j) { free(j); }
    if (b) {
        if (log_lock_init) {
            pthread_mutex_destroy(&b->fd_set_lock);
            pthread_mutex_destroy(&b->log_lock);
        }
        free(b);
    }

    if (pts) { free(pts); }
    if (fd_set) { yacht_free(fd_set, NULL, NULL); }

    return false;
}

static bool attempt_to_increase_resource_limits(struct bus *b) {
    struct rlimit info;
    if (-1 == getrlimit(RLIMIT_NOFILE, &info)) {
        fprintf(stderr, "getrlimit: %s", strerror(errno));
        errno = 0;
        return false;
    }

    const unsigned int nval = 1024;

    if (info.rlim_cur < nval && info.rlim_max > nval) {
        info.rlim_cur = nval;
        BUS_LOG_SNPRINTF(b, 3, LOG_MEMORY, b->udata, 256,
            "Current FD resource limits, [%lu, %lu], changing to %u",
            (unsigned long)info.rlim_cur, (unsigned long)info.rlim_max, nval);
        if (-1 == setrlimit(RLIMIT_NOFILE, &info)) {
            BUS_LOG_SNPRINTF(b, 3, LOG_MEMORY, b->udata, 256,
                "Failed to increase FD resource limit to %u, %s",
                nval, strerror(errno));
            fprintf(stderr, "getrlimit: %s", strerror(errno));
            errno = 0;
            return false;
        } else {
            BUS_LOG_SNPRINTF(b, 3, LOG_MEMORY, b->udata, 256,
                "Successfully increased FD resource limit to %u", nval);
        }
    } else {
        BUS_LOG_SNPRINTF(b, 3, LOG_MEMORY, b->udata, 256,
            "Current FD resource limits [%lu, %lu] are acceptable",
            (unsigned long)info.rlim_cur, (unsigned long)info.rlim_max);
    }
    return true;
}

/* Pack message to deliver on behalf of the user into an envelope
 * that can track status / routing along the way.
 *
 * The box should only ever be accessible on a single thread at a time. */
static boxed_msg *box_msg(struct bus *b, bus_user_msg *msg) {
    boxed_msg *box = calloc(1, sizeof(*box));
    if (box == NULL) { return NULL; }

    BUS_LOG_SNPRINTF(b, 3, LOG_MEMORY, b->udata, 64,
        "Allocated boxed message -- %p", (void*)box);

    box->fd = msg->fd;
    assert(msg->fd != 0);

    /* Lock hash table and check whether this FD uses SSL. */
    if (0 != pthread_mutex_lock(&b->fd_set_lock)) { assert(false); }
    void *value = NULL;
    SSL *ssl = NULL;
    if (yacht_get(b->fd_set, box->fd, &value)) {
        ssl = (SSL *)value;
        assert(ssl != NULL);
        box->ssl = ssl;
    }
    if (0 != pthread_mutex_unlock(&b->fd_set_lock)) { assert(false); }

    if (ssl == NULL) {
        /* socket isn't registered, fail out */
        free(box);
        return NULL;
    }

    box->timeout_sec = (time_t)msg->timeout_sec;
    if (box->timeout_sec == 0) {
        box->timeout_sec = BUS_DEFAULT_TIMEOUT_SEC;
    }

    box->out_seq_id = msg->seq_id;
    box->out_msg_size = msg->msg_size;

    /* Store message by pointer, since the client thread using it is blocked
     * until we are done sending. */
    box->out_msg = msg->msg;

    box->cb = msg->cb;
    box->udata = msg->udata;
    return box;
}

bool bus_send_request(struct bus *b, bus_user_msg *msg)
{
    if (b == NULL || msg == NULL || msg->fd == 0) {
        return false;
    }

    boxed_msg *box = box_msg(b, msg);
    if (box == NULL) {
        return false;
    }

    int s_id = sender_id_of_socket(b, msg->fd);
    struct sender *s = b->senders[s_id];

    BUS_LOG_SNPRINTF(b, 3-0, LOG_SENDING_REQUEST, b->udata, 64,
        "Sending request <fd:%d, seq_id:%lld>", msg->fd, (long long)msg->seq_id);
    bool res = sender_send_request(s, box);
    BUS_LOG_SNPRINTF(b, 3, LOG_SENDING_REQUEST, b->udata, 64,
        "...request sent, result %d", res);
    return res;
}

static bool poll_on_completion(struct bus *b, int fd) {
    /* POLL in a pipe */
    struct pollfd fds[1];
    fds[0].fd = fd;
    fds[0].events = POLLIN;

    /* TODO REFACTOR this should be reused between bus, sender, and listener,
     *     or be moved into the listener.
     *     The sender has its own blocking polling for commands. */

    /* FIXME: compare this to TCP timeouts -- try to prevent sender
     * succeeding but failing to notify client. */
    const int TIMEOUT_SECONDS = 10;
    const int ONE_SECOND = 1000;  // msec

    for (int i = 0; i < TIMEOUT_SECONDS; i++) {
        BUS_LOG(b, 5, LOG_SENDING_REQUEST, "Polling on completion...tick...", b->udata);
        int res = poll(fds, 1, ONE_SECOND);
        if (res == -1) {
            if (util_is_resumable_io_error(errno)) {
                BUS_LOG(b, 3, LOG_SENDING_REQUEST, "Polling on completion...EAGAIN", b->udata);
                if (errno == EINTR && i > 0) { i--; }
{'fraction_non_alphanumeric': 0.13402201984504553, 'fraction_numerical': 0.008971047981514204, 'mean_word_length': 2.54969443550981, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 243, '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': '1309015', 'n_tokens_mistral': 7961, 'n_tokens_neox': 7608, 'n_words': 2019}