<commit_before>/*
 * D-Bus Socket Abstraction
 *
 * The Socket objects wraps a single connection between two DBus peers
 * using streaming sockets. File-desciptor management is done by the caller.
 * This object is mainly used for line and message buffering. It supports
 * dual-mode: Line-based buffers for initial SASL transactions, and
 * message-based buffers for DBus transactions.
 *
 * The first line (if any) of a SASL exchange sent from a client to a server
 * must be prepended with a null byte. The wrapper handles this internally.
 *
 * Note that once the first real DBus message was read, you must not use the
 * line-helpers, anymore!
 */

#include <c-list.h>
#include <c-macro.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/un.h>
#include "dbus/message.h"
#include "socket.h"
#include "util/error.h"
#include "util/fdlist.h"

static char *socket_buffer_get_base(SocketBuffer *buffer) {
        return (char *)(buffer->vecs + buffer->n_vecs);
}

static int socket_buffer_new(SocketBuffer **bufferp, size_t n_vecs, size_t n_line) {
        SocketBuffer *buffer;

        buffer = malloc(sizeof(*buffer) + n_vecs * sizeof(*buffer->vecs) + n_line);
        if (!buffer)
                return error_origin(-ENOMEM);

        buffer->link = (CList)C_LIST_INIT(buffer->link);
        buffer->n_total = n_line;
        buffer->message = NULL;
        buffer->n_vecs = n_vecs;
        buffer->writer = NULL;

        *bufferp = buffer;
        return 0;
}

static int socket_buffer_new_line(SocketBuffer **bufferp, size_t n) {
        SocketBuffer *buffer;
        int r;

        r = socket_buffer_new(&buffer, 1, c_max(n, SOCKET_LINE_PREALLOC));
        if (r)
                return r;

        buffer->vecs[0] = (struct iovec){ socket_buffer_get_base(buffer), 0 };

        *bufferp = buffer;
        return 0;
}

int socket_buffer_new_message(SocketBuffer **bufferp, Message *message) {
        SocketBuffer *buffer;
        int r;

        r = socket_buffer_new(&buffer, C_ARRAY_SIZE(message->vecs), 0);
        if (r)
                return r;

        buffer->message = message_ref(message);
        memcpy(buffer->vecs, message->vecs, sizeof(message->vecs));

        *bufferp = buffer;
        return 0;
}

SocketBuffer *socket_buffer_free(SocketBuffer *buffer) {
        if (!buffer)
                return NULL;

        c_list_unlink_init(&buffer->link);
        message_unref(buffer->message);
        free(buffer);

        return NULL;
}

static size_t socket_buffer_get_line_space(SocketBuffer *buffer) {
        size_t n_remaining;

        assert(!buffer->message);

        n_remaining = buffer->n_total;
        n_remaining -= (char *)buffer->vecs[0].iov_base - socket_buffer_get_base(buffer);
        n_remaining -= buffer->vecs[0].iov_len;

        return n_remaining;
}

static void socket_buffer_get_line_cursor(SocketBuffer *buffer, char **datap, size_t **posp) {
        assert(!buffer->message);

        *datap = buffer->vecs[0].iov_base + buffer->vecs[0].iov_len;
        *posp = &buffer->vecs[0].iov_len;
}

static bool socket_buffer_is_uncomsumed(SocketBuffer *buffer) {
        return !buffer->writer;
}

static bool socket_buffer_is_consumed(SocketBuffer *buffer) {
        return buffer->writer >= buffer->vecs + buffer->n_vecs;
}

static bool socket_buffer_consume(SocketBuffer *buffer, size_t n) {
        size_t t;

        if (!buffer->writer)
                buffer->writer = buffer->vecs;

        for ( ; !socket_buffer_is_consumed(buffer); ++buffer->writer) {
                t = c_min(buffer->writer->iov_len, n);
                buffer->writer->iov_len -= t;
                buffer->writer->iov_base += t;
                n -= t;
                if (buffer->writer->iov_len)
                        break;
        }

        assert(!n);

        return socket_buffer_is_consumed(buffer);
}

int socket_new(Socket **socketp, int fd, bool server) {
        _c_cleanup_(socket_freep) Socket *socket = NULL;

        socket = calloc(1, sizeof(*socket));
        if (!socket)
                return error_origin(-ENOMEM);

        socket->fd = fd;
        socket->server = server;

        socket->out.queue = (CList)C_LIST_INIT(socket->out.queue);

        socket->in.data_size = SOCKET_DATA_PREALLOC;
        socket->in.data = malloc(socket->in.data_size);
        if (!socket->in.data)
                return error_origin(-ENOMEM);

        *socketp = socket;
        socket = NULL;
        return 0;
}

Socket *socket_free(Socket *socket) {
        SocketBuffer *buffer;

        if (!socket)
                return NULL;

        socket->fd = c_close(socket->fd);

        while ((buffer = c_list_first_entry(&socket->out.queue, SocketBuffer, link)))<commit_msg>socket: order constructor calls according to struct definition<commit_after>/*
 * D-Bus Socket Abstraction
 *
 * The Socket objects wraps a single connection between two DBus peers
 * using streaming sockets. File-desciptor management is done by the caller.
 * This object is mainly used for line and message buffering. It supports
 * dual-mode: Line-based buffers for initial SASL transactions, and
 * message-based buffers for DBus transactions.
 *
 * The first line (if any) of a SASL exchange sent from a client to a server
 * must be prepended with a null byte. The wrapper handles this internally.
 *
 * Note that once the first real DBus message was read, you must not use the
 * line-helpers, anymore!
 */

#include <c-list.h>
#include <c-macro.h>
#include <stdlib.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/un.h>
#include "dbus/message.h"
#include "socket.h"
#include "util/error.h"
#include "util/fdlist.h"

static char *socket_buffer_get_base(SocketBuffer *buffer) {
        return (char *)(buffer->vecs + buffer->n_vecs);
}

static int socket_buffer_new(SocketBuffer **bufferp, size_t n_vecs, size_t n_line) {
        SocketBuffer *buffer;

        buffer = malloc(sizeof(*buffer) + n_vecs * sizeof(*buffer->vecs) + n_line);
        if (!buffer)
                return error_origin(-ENOMEM);

        buffer->link = (CList)C_LIST_INIT(buffer->link);
        buffer->n_total = n_line;
        buffer->message = NULL;
        buffer->n_vecs = n_vecs;
        buffer->writer = NULL;

        *bufferp = buffer;
        return 0;
}

static int socket_buffer_new_line(SocketBuffer **bufferp, size_t n) {
        SocketBuffer *buffer;
        int r;

        r = socket_buffer_new(&buffer, 1, c_max(n, SOCKET_LINE_PREALLOC));
        if (r)
                return r;

        buffer->vecs[0] = (struct iovec){ socket_buffer_get_base(buffer), 0 };

        *bufferp = buffer;
        return 0;
}

int socket_buffer_new_message(SocketBuffer **bufferp, Message *message) {
        SocketBuffer *buffer;
        int r;

        r = socket_buffer_new(&buffer, C_ARRAY_SIZE(message->vecs), 0);
        if (r)
                return r;

        buffer->message = message_ref(message);
        memcpy(buffer->vecs, message->vecs, sizeof(message->vecs));

        *bufferp = buffer;
        return 0;
}

SocketBuffer *socket_buffer_free(SocketBuffer *buffer) {
        if (!buffer)
                return NULL;

        c_list_unlink_init(&buffer->link);
        message_unref(buffer->message);
        free(buffer);

        return NULL;
}

static size_t socket_buffer_get_line_space(SocketBuffer *buffer) {
        size_t n_remaining;

        assert(!buffer->message);

        n_remaining = buffer->n_total;
        n_remaining -= (char *)buffer->vecs[0].iov_base - socket_buffer_get_base(buffer);
        n_remaining -= buffer->vecs[0].iov_len;

        return n_remaining;
}

static void socket_buffer_get_line_cursor(SocketBuffer *buffer, char **datap, size_t **posp) {
        assert(!buffer->message);

        *datap = buffer->vecs[0].iov_base + buffer->vecs[0].iov_len;
        *posp = &buffer->vecs[0].iov_len;
}

static bool socket_buffer_is_uncomsumed(SocketBuffer *buffer) {
        return !buffer->writer;
}

static bool socket_buffer_is_consumed(SocketBuffer *buffer) {
        return buffer->writer >= buffer->vecs + buffer->n_vecs;
}

static bool socket_buffer_consume(SocketBuffer *buffer, size_t n) {
        size_t t;

        if (!buffer->writer)
                buffer->writer = buffer->vecs;

        for ( ; !socket_buffer_is_consumed(buffer); ++buffer->writer) {
                t = c_min(buffer->writer->iov_len, n);
                buffer->writer->iov_len -= t;
                buffer->writer->iov_base += t;
                n -= t;
                if (buffer->writer->iov_len)
                        break;
        }

        assert(!n);

        return socket_buffer_is_consumed(buffer);
}

int socket_new(Socket **socketp, int fd, bool server) {
        _c_cleanup_(socket_freep) Socket *socket = NULL;

        socket = calloc(1, sizeof(*socket));
        if (!socket)
                return error_origin(-ENOMEM);

        socket->fd = fd;
        socket->server = server;
        socket->in.data_size = SOCKET_DATA_PREALLOC;
        socket->out.queue = (CList)C_LIST_INIT(socket->out.queue);

        socket->in.data = malloc(socket->in.data_size);
        if (!socket->in.data)
                return error_origin(-ENOMEM);

        *socketp = socket;
        socket = NULL;
        return 0;
}

Socket *socket_free(Socket *socket) {
        SocketBuffer *buffer;

        if (!socket)
                return NULL;

        socket->fd = c_close(socket->fd);

        while ((buffer = c_list_first_entry(&socket->out.queue, SocketBuffer, link)))
{'fraction_non_alphanumeric': 0.11821989528795812, 'fraction_numerical': 0.002931937172774869, 'mean_word_length': 2.565136244867488, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 121, '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': '2792176', 'n_tokens_mistral': 3042, 'n_tokens_neox': 2948, 'n_words': 708}