<commit_before>/*
    Copyright (C) 2009-2016 Red Hat, Inc.

   This library is free software; you can redistribute it and/or
   modify it under the terms of the GNU Lesser General Public
   License as published by the Free Software Foundation; either
   version 2.1 of the License, or (at your option) any later version.

   This library 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
   Lesser General Public License for more details.

   You should have received a copy of the GNU Lesser General Public
   License along with this library; if not, see <http://www.gnu.org/licenses/>.
*/

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include <glib.h>
#include <stdio.h>
#include <stdint.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <sys/ioctl.h>
#ifdef HAVE_LINUX_SOCKIOS_H
#include <linux/sockios.h> /* SIOCOUTQ */
#endif
#include <common/generated_server_marshallers.h>

#include "red-channel-client.h"
#include "red-client.h"
#include "glib-compat.h"

#define CLIENT_ACK_WINDOW 20

#define MAX_HEADER_SIZE sizeof(SpiceDataHeader)

#ifndef IOV_MAX
#define IOV_MAX 1024
#endif

typedef struct SpiceDataHeaderOpaque SpiceDataHeaderOpaque;

typedef uint16_t (*get_msg_type_proc)(SpiceDataHeaderOpaque *header);
typedef uint32_t (*get_msg_size_proc)(SpiceDataHeaderOpaque *header);
typedef void (*set_msg_type_proc)(SpiceDataHeaderOpaque *header, uint16_t type);
typedef void (*set_msg_size_proc)(SpiceDataHeaderOpaque *header, uint32_t size);
typedef void (*set_msg_serial_proc)(SpiceDataHeaderOpaque *header, uint64_t serial);
typedef void (*set_msg_sub_list_proc)(SpiceDataHeaderOpaque *header, uint32_t sub_list);

struct SpiceDataHeaderOpaque {
    uint8_t *data;
    uint16_t header_size;

    set_msg_type_proc set_msg_type;
    set_msg_size_proc set_msg_size;
    set_msg_serial_proc set_msg_serial;
    set_msg_sub_list_proc set_msg_sub_list;

    get_msg_type_proc get_msg_type;
    get_msg_size_proc get_msg_size;
};

typedef struct RedChannelClientLatencyMonitor {
    int state;
    uint64_t last_pong_time;
    SpiceTimer *timer;
    uint32_t id;
    int tcp_nodelay;
    int warmup_was_sent;

    int64_t roundtrip;
} RedChannelClientLatencyMonitor;

typedef struct RedChannelClientConnectivityMonitor {
    int state;
    bool sent_bytes;
    bool received_bytes;
    uint32_t timeout;
    SpiceTimer *timer;
} RedChannelClientConnectivityMonitor;

typedef struct OutgoingMessageBuffer {
    struct iovec vec[IOV_MAX];
    int vec_size;
    int pos;
    int size;
} OutgoingMessageBuffer;

typedef struct IncomingMessageBuffer {
    uint8_t header_buf[MAX_HEADER_SIZE];
    SpiceDataHeaderOpaque header;
    uint32_t header_pos;
    uint8_t *msg; // data of the msg following the header. allocated by alloc_msg_buf.
    uint32_t msg_pos;
} IncomingMessageBuffer;

struct RedChannelClientPrivate
{
    RedChannel *channel;
    RedClient  *client;
    RedsStream *stream;
    gboolean monitor_latency;

    struct {
        uint32_t generation;
        uint32_t client_generation;
        uint32_t messages_window;
        uint32_t client_window;
    } ack_data;

    struct {
        /* this can be either main.marshaller or urgent.marshaller */
        SpiceMarshaller *marshaller;
        SpiceDataHeaderOpaque header;
        uint32_t size;
        int blocked;
        uint64_t last_sent_serial;

        struct {
            SpiceMarshaller *marshaller;
            uint8_t *header_data;
        } main;

        struct {
            SpiceMarshaller *marshaller;
        } urgent;
    } send_data;

    int during_send;
    GQueue pipe;

    RedChannelCapabilities remote_caps;
    int is_mini_header;
    gboolean destroying;

    int wait_migrate_data;
    int wait_migrate_flush_mark;

    RedChannelClientLatencyMonitor latency_monitor;
    RedChannelClientConnectivityMonitor connectivity_monitor;

    IncomingMessageBuffer incoming;
    OutgoingMessageBuffer outgoing;

    RedStatCounter out_messages;
    RedStatCounter out_bytes;
};

static const SpiceDataHeaderOpaque full_header_wrapper;
static const SpiceDataHeaderOpaque mini_header_wrapper;
static void red_channel_client_clear_sent_item(RedChannelClient *rcc);
static void red_channel_client_initable_interface_init(GInitableIface *iface);
static void red_channel_client_set_message_serial(RedChannelClient *channel, uint64_t);
static bool red_channel_client_config_socket(RedChannelClient *rcc);

/*
 * When an error occurs over a channel, we treat it as a warning
 * for spice-server and shutdown the channel.
 */
#define spice_channel_client_error(rcc, format, ...)                                     \
    do {                                                                                 \
        RedChannel *_ch = red_channel_client_get_channel(rcc);                           \
        uint32_t _type, _id;                                                             \
        g_object_get(_ch, "channel-type", &_type, "id", &_id, NULL);                     \
        spice_warning("rcc %p type %u id %u: " format, rcc,                              \
                    type, id, ## __VA_ARGS__);                                           \
        red_channel_client_shutdown(rcc);                                                \
    } while (0)

G_DEFINE_TYPE_WITH_CODE(RedChannelClient, red_channel_client, G_TYPE_OBJECT,
                        G_IMPLEMENT_INTERFACE(G_TYPE_INITABLE,
                                              red_channel_client_initable_interface_init))

#define CHANNEL_CLIENT_PRIVATE(o) \
    (G_TYPE_INSTANCE_GET_PRIVATE((o), RED_TYPE_CHANNEL_CLIENT, RedChannelClientPrivate))

static gboolean red_channel_client_initable_init(GInitable *initable,
                                                 GCancellable *cancellable,
                                                 GError **error);

enum {
    PROP0,
    PROP_STREAM,
    PROP_CHANNEL,
    PROP_CLIENT,
    PROP_MONITOR_LATENCY,
    PROP_CAPS
};

#define PING_TEST_TIMEOUT_MS (MSEC_PER_SEC * 15)
#define PING_TEST_IDLE_NET_TIMEOUT_MS (MSEC_PER_SEC / 10)

enum QosPingState {
    PING_STATE_NONE,
    PING_STATE_TIMER,
    PING_STATE_WARMUP,
    PING_STATE_LATENCY,
};

enum ConnectivityState {
    CONNECTIVITY_STATE_CONNECTED,
    CONNECTIVITY_STATE_BLOCKED,
    CONNECTIVITY_STATE_WAIT_PONG,
    CONNECTIVITY_STATE_DISCONNECTED,
};

typedef struct RedEmptyMsgPipeItem {
    RedPipeItem base;
    int msg;
} RedEmptyMsgPipeItem;

typedef struct MarkerPipeItem {
    RedPipeItem base;
    gboolean *item_in_pipe;
} MarkerPipeItem;

static void red_channel_client_start_ping_timer(RedChannelClient *rcc, uint32_t timeout)
{
    SpiceCoreInterfaceInternal *core;

    if (!rcc->priv->latency_monitor.timer) {<commit_msg>Use enum rather than int in RedChannelClient{Latency,Connectivity}Monitor<commit_after>/*
    Copyright (C) 2009-2016 Red Hat, Inc.

   This library is free software; you can redistribute it and/or
   modify it under the terms of the GNU Lesser General Public
   License as published by the Free Software Foundation; either
   version 2.1 of the License, or (at your option) any later version.

   This library 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
   Lesser General Public License for more details.

   You should have received a copy of the GNU Lesser General Public
   License along with this library; if not, see <http://www.gnu.org/licenses/>.
*/

#ifdef HAVE_CONFIG_H
#include <config.h>
#endif

#include <glib.h>
#include <stdio.h>
#include <stdint.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
#include <sys/ioctl.h>
#ifdef HAVE_LINUX_SOCKIOS_H
#include <linux/sockios.h> /* SIOCOUTQ */
#endif
#include <common/generated_server_marshallers.h>

#include "red-channel-client.h"
#include "red-client.h"
#include "glib-compat.h"

#define CLIENT_ACK_WINDOW 20

#define MAX_HEADER_SIZE sizeof(SpiceDataHeader)

#ifndef IOV_MAX
#define IOV_MAX 1024
#endif

typedef struct SpiceDataHeaderOpaque SpiceDataHeaderOpaque;

typedef uint16_t (*get_msg_type_proc)(SpiceDataHeaderOpaque *header);
typedef uint32_t (*get_msg_size_proc)(SpiceDataHeaderOpaque *header);
typedef void (*set_msg_type_proc)(SpiceDataHeaderOpaque *header, uint16_t type);
typedef void (*set_msg_size_proc)(SpiceDataHeaderOpaque *header, uint32_t size);
typedef void (*set_msg_serial_proc)(SpiceDataHeaderOpaque *header, uint64_t serial);
typedef void (*set_msg_sub_list_proc)(SpiceDataHeaderOpaque *header, uint32_t sub_list);

struct SpiceDataHeaderOpaque {
    uint8_t *data;
    uint16_t header_size;

    set_msg_type_proc set_msg_type;
    set_msg_size_proc set_msg_size;
    set_msg_serial_proc set_msg_serial;
    set_msg_sub_list_proc set_msg_sub_list;

    get_msg_type_proc get_msg_type;
    get_msg_size_proc get_msg_size;
};

typedef enum {
    PING_STATE_NONE,
    PING_STATE_TIMER,
    PING_STATE_WARMUP,
    PING_STATE_LATENCY,
} QosPingState;

typedef struct RedChannelClientLatencyMonitor {
    QosPingState state;
    uint64_t last_pong_time;
    SpiceTimer *timer;
    uint32_t id;
    int tcp_nodelay;
    int warmup_was_sent;

    int64_t roundtrip;
} RedChannelClientLatencyMonitor;

typedef enum {
    CONNECTIVITY_STATE_CONNECTED,
    CONNECTIVITY_STATE_BLOCKED,
    CONNECTIVITY_STATE_WAIT_PONG,
    CONNECTIVITY_STATE_DISCONNECTED,
} ConnectivityState;

typedef struct RedChannelClientConnectivityMonitor {
    ConnectivityState state;
    bool sent_bytes;
    bool received_bytes;
    uint32_t timeout;
    SpiceTimer *timer;
} RedChannelClientConnectivityMonitor;

typedef struct OutgoingMessageBuffer {
    struct iovec vec[IOV_MAX];
    int vec_size;
    int pos;
    int size;
} OutgoingMessageBuffer;

typedef struct IncomingMessageBuffer {
    uint8_t header_buf[MAX_HEADER_SIZE];
    SpiceDataHeaderOpaque header;
    uint32_t header_pos;
    uint8_t *msg; // data of the msg following the header. allocated by alloc_msg_buf.
    uint32_t msg_pos;
} IncomingMessageBuffer;

struct RedChannelClientPrivate
{
    RedChannel *channel;
    RedClient  *client;
    RedsStream *stream;
    gboolean monitor_latency;

    struct {
        uint32_t generation;
        uint32_t client_generation;
        uint32_t messages_window;
        uint32_t client_window;
    } ack_data;

    struct {
        /* this can be either main.marshaller or urgent.marshaller */
        SpiceMarshaller *marshaller;
        SpiceDataHeaderOpaque header;
        uint32_t size;
        int blocked;
        uint64_t last_sent_serial;

        struct {
            SpiceMarshaller *marshaller;
            uint8_t *header_data;
        } main;

        struct {
            SpiceMarshaller *marshaller;
        } urgent;
    } send_data;

    int during_send;
    GQueue pipe;

    RedChannelCapabilities remote_caps;
    int is_mini_header;
    gboolean destroying;

    int wait_migrate_data;
    int wait_migrate_flush_mark;

    RedChannelClientLatencyMonitor latency_monitor;
    RedChannelClientConnectivityMonitor connectivity_monitor;

    IncomingMessageBuffer incoming;
    OutgoingMessageBuffer outgoing;

    RedStatCounter out_messages;
    RedStatCounter out_bytes;
};

static const SpiceDataHeaderOpaque full_header_wrapper;
static const SpiceDataHeaderOpaque mini_header_wrapper;
static void red_channel_client_clear_sent_item(RedChannelClient *rcc);
static void red_channel_client_initable_interface_init(GInitableIface *iface);
static void red_channel_client_set_message_serial(RedChannelClient *channel, uint64_t);
static bool red_channel_client_config_socket(RedChannelClient *rcc);

/*
 * When an error occurs over a channel, we treat it as a warning
 * for spice-server and shutdown the channel.
 */
#define spice_channel_client_error(rcc, format, ...)                                     \
    do {                                                                                 \
        RedChannel *_ch = red_channel_client_get_channel(rcc);                           \
        uint32_t _type, _id;                                                             \
        g_object_get(_ch, "channel-type", &_type, "id", &_id, NULL);                     \
        spice_warning("rcc %p type %u id %u: " format, rcc,                              \
                    type, id, ## __VA_ARGS__);                                           \
        red_channel_client_shutdown(rcc);                                                \
    } while (0)

G_DEFINE_TYPE_WITH_CODE(RedChannelClient, red_channel_client, G_TYPE_OBJECT,
                        G_IMPLEMENT_INTERFACE(G_TYPE_INITABLE,
                                              red_channel_client_initable_interface_init))

#define CHANNEL_CLIENT_PRIVATE(o) \
    (G_TYPE_INSTANCE_GET_PRIVATE((o), RED_TYPE_CHANNEL_CLIENT, RedChannelClientPrivate))

static gboolean red_channel_client_initable_init(GInitable *initable,
                                                 GCancellable *cancellable,
                                                 GError **error);

enum {
    PROP0,
    PROP_STREAM,
    PROP_CHANNEL,
    PROP_CLIENT,
    PROP_MONITOR_LATENCY,
    PROP_CAPS
};

#define PING_TEST_TIMEOUT_MS (MSEC_PER_SEC * 15)
#define PING_TEST_IDLE_NET_TIMEOUT_MS (MSEC_PER_SEC / 10)

typedef struct RedEmptyMsgPipeItem {
    RedPipeItem base;
    int msg;
} RedEmptyMsgPipeItem;

typedef struct MarkerPipeItem {
    RedPipeItem base;
    gboolean *item_in_pipe;
} MarkerPipeItem;

static void red_channel_client_start_ping_timer(RedChannelClient *rcc, uint32_t timeout)
{
    SpiceCoreInterfaceInternal *core;

    if (!rcc->priv->latency_monitor.timer) {
{'fraction_non_alphanumeric': 0.06535619484898486, 'fraction_numerical': 0.010043762106320396, 'mean_word_length': 3.0417512322412295, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 30, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3683415', 'n_tokens_mistral': 4497, 'n_tokens_neox': 4302, 'n_words': 1010}