<commit_before>/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2019-2021 The Fluent Bit Authors
 *  Copyright (C) 2015-2018 Treasure Data 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 <fluent-bit/flb_info.h>
#include <fluent-bit/flb_pipe.h>
#include <fluent-bit/flb_engine.h>
#include <fluent-bit/flb_mem.h>
#include <fluent-bit/flb_mp.h>
#include <fluent-bit/flb_log.h>
#include <fluent-bit/flb_event_loop.h>
#include <fluent-bit/flb_utils.h>
#include <fluent-bit/flb_scheduler.h>
#include <fluent-bit/flb_input.h>
#include <fluent-bit/flb_input_plugin.h>
#include <fluent-bit/flb_input_thread.h>

static int input_thread_instance_set_status(struct flb_input_instance *ins, uint32_t status);
static int input_thread_instance_get_status(struct flb_input_instance *ins);

static void *worker(void *arg)
{
    struct flb_input_thread *it = arg;
    it->callback(it->write, it->data);
    fclose(it->write_file);
    return NULL;
}

static inline int handle_input_event(flb_pipefd_t fd, struct flb_input_instance *ins)
{
    int bytes;
    int ins_id;
    uint32_t type;
    uint32_t operation;
    uint64_t val;
    struct flb_config *config = ins->config;

    bytes = read(fd, &val, sizeof(val));
    if (bytes == -1) {
        flb_errno();
        return -1;
    }

    type   = FLB_BITS_U64_HIGH(val);
    operation = FLB_BITS_U64_LOW(val);

    /* At the moment we only support events coming from an input coroutine */
    if (type == FLB_ENGINE_IN_CORO) {
        ins_id = ins->id;
        flb_input_coro_finished(config, ins_id);
    }
    else if (type == FLB_INPUT_THREAD_TO_THREAD) {
        if (operation == FLB_INPUT_THREAD_PAUSE) {
            if (ins->p->cb_pause && ins->context) {
                ins->p->cb_pause(ins->context, ins->config);
            }
        }
        else if (operation == FLB_INPUT_THREAD_EXIT) {
            return FLB_INPUT_THREAD_EXIT;
        }
    }
    else {
        flb_error("[thread event loop] it happends on fd=%i, invalid type=%i", fd, type);
        return -1;
    }

    return 0;
}

static inline int handle_input_thread_event(flb_pipefd_t fd, struct flb_config *config)
{
    int bytes;
    uint32_t type;
    uint32_t ins_id;
    uint64_t val;

    bytes = flb_pipe_r(fd, &val, sizeof(val));
    if (bytes == -1) {
        flb_errno();
        return -1;
    }

    /* Get type and key */
    type   = FLB_BITS_U64_HIGH(val);
    ins_id = FLB_BITS_U64_LOW(val);

    /* At the moment we only support events coming from an input coroutine */
    if (type == FLB_ENGINE_IN_CORO) {
        flb_input_coro_finished(config, ins_id);
    }
    else {
        flb_error("[thread event loop] invalid thread event type %i for input handler",
                  type);
        return -1;
    }

    return 0;
}

static int input_collector_fd(flb_pipefd_t fd, struct flb_input_instance *ins)
{
    struct mk_list *head;
    struct flb_input_collector *collector = NULL;
    struct flb_input_coro *input_coro;
    struct flb_config *config = ins->config;

    mk_list_foreach(head, &ins->collectors) {
        collector = mk_list_entry(head, struct flb_input_collector, _head);
        if (collector->fd_event == fd) {
            break;
        }
        else if (collector->fd_timer == fd) {
            flb_utils_timer_consume(fd);
            break;
        }
        collector = NULL;
    }

    /* No matches */
    if (!collector) {
        return -1;
    }

    if (collector->running == FLB_FALSE) {
        return -1;
    }

    /* Trigger the collector callback */
    if (collector->instance->runs_in_coroutine) {
        input_coro = flb_input_coro_collect(collector, config);
        if (!input_coro) {
            return -1;
        }
        flb_input_coro_resume(input_coro);
    }
    else {
        collector->cb_collect(collector->instance, config,
                              collector->instance->context);
    }

    return 0;
}

static FLB_INLINE int engine_handle_event(flb_pipefd_t fd, int mask,
                                          struct flb_input_instance *ins,
                                          struct flb_config *config)
{
    int ret;

    if (mask & MK_EVENT_READ) {
        /* Try to match the file descriptor with a collector event */
        ret = input_collector_fd(fd, ins);
        if (ret != -1) {
            return ret;
        }
    }

    return 0;
}

static void input_thread_instance_destroy(struct flb_input_thread_instance *thi)
{
    if (thi->evl) {
        mk_event_loop_destroy(thi->evl);
    }

    /* ch_parent_events */
    if (thi->ch_parent_events[0] > 0) {
        mk_event_closesocket(thi->ch_parent_events[0]);
    }
    if (thi->ch_parent_events[1] > 0) {
        mk_event_closesocket(thi->ch_parent_events[1]);
    }

    /* ch_thread_events */
    if (thi->ch_thread_events[0] > 0) {
        mk_event_closesocket(thi->ch_thread_events[0]);
    }
    if (thi->ch_thread_events[1] > 0) {
        mk_event_closesocket(thi->ch_thread_events[1]);
    }

    flb_tp_destroy(thi->tp);
    flb_free(thi);
}

static struct flb_input_thread_instance *input_thread_instance_create(struct flb_input_instance *ins)
{
    int ret;
    struct flb_input_thread_instance *thi;

    /* context for thread */
    thi = flb_calloc(1, sizeof(struct flb_input_thread_instance));
    if (!thi) {
        flb_errno();
        return NULL;
    }
    thi->ins = ins;
    thi->config = ins->config;

    /* init status */
    thi->init_status = 0;
    pthread_mutex_init(&thi->init_mutex, NULL);

    /* initialize lists */
    mk_list_init(&thi->input_coro_list);
    mk_list_init(&thi->input_coro_list_destroy);

    /* event loop */
    thi->evl = mk_event_loop_create(256);
    if (!thi->evl) {
        input_thread_instance_destroy(thi);
        return NULL;
    }

    /* channel to receive parent (engine) notifications */
    ret = mk_event_channel_create(thi->evl,
                                  &thi->ch_parent_events[0],
                                  &thi->ch_parent_events[1],
                                  &thi->event);
    if (ret == -1) {
        flb_error("could not initialize parent channels for %s",<commit_msg>input_thread: macos: Add a missing pthread_cond_init<commit_after>/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2019-2021 The Fluent Bit Authors
 *  Copyright (C) 2015-2018 Treasure Data 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 <fluent-bit/flb_info.h>
#include <fluent-bit/flb_pipe.h>
#include <fluent-bit/flb_engine.h>
#include <fluent-bit/flb_mem.h>
#include <fluent-bit/flb_mp.h>
#include <fluent-bit/flb_log.h>
#include <fluent-bit/flb_event_loop.h>
#include <fluent-bit/flb_utils.h>
#include <fluent-bit/flb_scheduler.h>
#include <fluent-bit/flb_input.h>
#include <fluent-bit/flb_input_plugin.h>
#include <fluent-bit/flb_input_thread.h>

static int input_thread_instance_set_status(struct flb_input_instance *ins, uint32_t status);
static int input_thread_instance_get_status(struct flb_input_instance *ins);

static void *worker(void *arg)
{
    struct flb_input_thread *it = arg;
    it->callback(it->write, it->data);
    fclose(it->write_file);
    return NULL;
}

static inline int handle_input_event(flb_pipefd_t fd, struct flb_input_instance *ins)
{
    int bytes;
    int ins_id;
    uint32_t type;
    uint32_t operation;
    uint64_t val;
    struct flb_config *config = ins->config;

    bytes = read(fd, &val, sizeof(val));
    if (bytes == -1) {
        flb_errno();
        return -1;
    }

    type   = FLB_BITS_U64_HIGH(val);
    operation = FLB_BITS_U64_LOW(val);

    /* At the moment we only support events coming from an input coroutine */
    if (type == FLB_ENGINE_IN_CORO) {
        ins_id = ins->id;
        flb_input_coro_finished(config, ins_id);
    }
    else if (type == FLB_INPUT_THREAD_TO_THREAD) {
        if (operation == FLB_INPUT_THREAD_PAUSE) {
            if (ins->p->cb_pause && ins->context) {
                ins->p->cb_pause(ins->context, ins->config);
            }
        }
        else if (operation == FLB_INPUT_THREAD_EXIT) {
            return FLB_INPUT_THREAD_EXIT;
        }
    }
    else {
        flb_error("[thread event loop] it happends on fd=%i, invalid type=%i", fd, type);
        return -1;
    }

    return 0;
}

static inline int handle_input_thread_event(flb_pipefd_t fd, struct flb_config *config)
{
    int bytes;
    uint32_t type;
    uint32_t ins_id;
    uint64_t val;

    bytes = flb_pipe_r(fd, &val, sizeof(val));
    if (bytes == -1) {
        flb_errno();
        return -1;
    }

    /* Get type and key */
    type   = FLB_BITS_U64_HIGH(val);
    ins_id = FLB_BITS_U64_LOW(val);

    /* At the moment we only support events coming from an input coroutine */
    if (type == FLB_ENGINE_IN_CORO) {
        flb_input_coro_finished(config, ins_id);
    }
    else {
        flb_error("[thread event loop] invalid thread event type %i for input handler",
                  type);
        return -1;
    }

    return 0;
}

static int input_collector_fd(flb_pipefd_t fd, struct flb_input_instance *ins)
{
    struct mk_list *head;
    struct flb_input_collector *collector = NULL;
    struct flb_input_coro *input_coro;
    struct flb_config *config = ins->config;

    mk_list_foreach(head, &ins->collectors) {
        collector = mk_list_entry(head, struct flb_input_collector, _head);
        if (collector->fd_event == fd) {
            break;
        }
        else if (collector->fd_timer == fd) {
            flb_utils_timer_consume(fd);
            break;
        }
        collector = NULL;
    }

    /* No matches */
    if (!collector) {
        return -1;
    }

    if (collector->running == FLB_FALSE) {
        return -1;
    }

    /* Trigger the collector callback */
    if (collector->instance->runs_in_coroutine) {
        input_coro = flb_input_coro_collect(collector, config);
        if (!input_coro) {
            return -1;
        }
        flb_input_coro_resume(input_coro);
    }
    else {
        collector->cb_collect(collector->instance, config,
                              collector->instance->context);
    }

    return 0;
}

static FLB_INLINE int engine_handle_event(flb_pipefd_t fd, int mask,
                                          struct flb_input_instance *ins,
                                          struct flb_config *config)
{
    int ret;

    if (mask & MK_EVENT_READ) {
        /* Try to match the file descriptor with a collector event */
        ret = input_collector_fd(fd, ins);
        if (ret != -1) {
            return ret;
        }
    }

    return 0;
}

static void input_thread_instance_destroy(struct flb_input_thread_instance *thi)
{
    if (thi->evl) {
        mk_event_loop_destroy(thi->evl);
    }

    /* ch_parent_events */
    if (thi->ch_parent_events[0] > 0) {
        mk_event_closesocket(thi->ch_parent_events[0]);
    }
    if (thi->ch_parent_events[1] > 0) {
        mk_event_closesocket(thi->ch_parent_events[1]);
    }

    /* ch_thread_events */
    if (thi->ch_thread_events[0] > 0) {
        mk_event_closesocket(thi->ch_thread_events[0]);
    }
    if (thi->ch_thread_events[1] > 0) {
        mk_event_closesocket(thi->ch_thread_events[1]);
    }

    flb_tp_destroy(thi->tp);
    flb_free(thi);
}

static struct flb_input_thread_instance *input_thread_instance_create(struct flb_input_instance *ins)
{
    int ret;
    struct flb_input_thread_instance *thi;

    /* context for thread */
    thi = flb_calloc(1, sizeof(struct flb_input_thread_instance));
    if (!thi) {
        flb_errno();
        return NULL;
    }
    thi->ins = ins;
    thi->config = ins->config;

    /* init status */
    thi->init_status = 0;
    pthread_mutex_init(&thi->init_mutex, NULL);

    /* init condition */
    pthread_cond_init(&thi->init_condition, NULL);

    /* initialize lists */
    mk_list_init(&thi->input_coro_list);
    mk_list_init(&thi->input_coro_list_destroy);

    /* event loop */
    thi->evl = mk_event_loop_create(256);
    if (!thi->evl) {
        input_thread_instance_destroy(thi);
        return NULL;
    }

    /* channel to receive parent (engine) notifications */
    ret = mk_event_channel_create(thi->evl,
                                  &thi->ch_parent_events[0],
                                  &thi->ch_parent_events[1],
                                  &thi->event);
    if (ret == -1) {
        flb_error("could not initialize parent channels for %s",
{'fraction_non_alphanumeric': 0.11314243759177679, 'fraction_numerical': 0.01145374449339207, 'mean_word_length': 2.47829417773238, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 132, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7424271', 'n_tokens_mistral': 4779, 'n_tokens_neox': 4588, 'n_words': 1032}