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

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2019-2020 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 <stdio.h>
#include <stdlib.h>

#include <monkey/mk_core.h>
#include <fluent-bit/flb_time.h>
#include <fluent-bit/flb_info.h>
#include <fluent-bit/flb_bits.h>

#include <fluent-bit/flb_macros.h>
#include <fluent-bit/flb_pipe.h>
#include <fluent-bit/flb_input.h>
#include <fluent-bit/flb_output.h>
#include <fluent-bit/flb_error.h>
#include <fluent-bit/flb_utils.h>
#include <fluent-bit/flb_config.h>
#include <fluent-bit/flb_engine.h>
#include <fluent-bit/flb_engine_dispatch.h>
#include <fluent-bit/flb_task.h>
#include <fluent-bit/flb_router.h>
#include <fluent-bit/flb_http_server.h>
#include <fluent-bit/flb_scheduler.h>
#include <fluent-bit/flb_parser.h>
#include <fluent-bit/flb_sosreport.h>
#include <fluent-bit/flb_storage.h>
#include <fluent-bit/flb_http_server.h>

#ifdef FLB_HAVE_METRICS
#include <fluent-bit/flb_metrics_exporter.h>
#endif

#ifdef FLB_HAVE_STREAM_PROCESSOR
#include <fluent-bit/stream_processor/flb_sp.h>
#endif

FLB_TLS_DEFINE(struct mk_event_loop, flb_engine_evl);


void flb_engine_evl_init()
{
    FLB_TLS_INIT(flb_engine_evl);
}

struct mk_event_loop *flb_engine_evl_get()
{
    struct mk_event_loop *evl;

    evl = FLB_TLS_GET(flb_engine_evl);
    return evl;
}

void flb_engine_evl_set(struct mk_event_loop *evl)
{
    FLB_TLS_SET(flb_engine_evl, evl);
}

int flb_engine_destroy_tasks(struct mk_list *tasks)
{
    int c = 0;
    struct mk_list *tmp;
    struct mk_list *head;
    struct flb_task *task;

    mk_list_foreach_safe(head, tmp, tasks) {
        task = mk_list_entry(head, struct flb_task, _head);
        flb_task_destroy(task, FLB_FALSE);
        c++;
    }

    return c;
}

int flb_engine_flush(struct flb_config *config,
                     struct flb_input_plugin *in_force)
{
    struct flb_input_instance *in;
    struct flb_input_plugin *p;
    struct mk_list *head;

    mk_list_foreach(head, &config->inputs) {
        in = mk_list_entry(head, struct flb_input_instance, _head);
        p = in->p;

        if (in_force != NULL && p != in_force) {
            continue;
        }
        flb_engine_dispatch(0, in, config);
    }

    return 0;
}

/* Cleanup function that runs every 1.5 second */
static void cb_engine_sched_timer(struct flb_config *ctx, void *data)
{
    (void) data;

    /* Upstream connections timeouts handling */
    flb_upstream_conn_timeouts(&ctx->upstreams);
}

static inline int handle_output_event(flb_pipefd_t fd, struct flb_config *config)
{
    int ret;
    int bytes;
    int task_id;
    int out_id;
    int retries;
    int retry_seconds;
    uint32_t type;
    uint32_t key;
    uint64_t val;
    struct flb_task *task;
    struct flb_task_retry *retry;
    struct flb_output_instance *ins;

    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);
    key  = FLB_BITS_U64_LOW(val);

    if (type != FLB_ENGINE_TASK) {
        flb_error("[engine] invalid event type %i for output handler",
                  type);
        return -1;
    }

    /*
     * The notion of ENGINE_TASK is associated to outputs. All thread
     * references below belongs to flb_output_coro's.
     */
    ret     = FLB_TASK_RET(key);
    task_id = FLB_TASK_ID(key);
    out_id  = FLB_TASK_OUT(key);

#ifdef FLB_HAVE_TRACE
    char *trace_st = NULL;

    if (ret == FLB_OK) {
        trace_st = "OK";
    }
    else if (ret == FLB_ERROR) {
        trace_st = "ERROR";
    }
    else if (ret == FLB_RETRY) {
        trace_st = "RETRY";
    }

    flb_trace("%s[engine] [task event]%s task_id=%i out_id=%i return=%s",
              ANSI_YELLOW, ANSI_RESET,
              task_id, out_id, trace_st);
#endif

    task = config->tasks_map[task_id].task;
    ins  = flb_output_get_instance(config, out_id);

    /* A thread has finished, delete it */
    if (ret == FLB_OK) {

        /* Update metrics */
#ifdef FLB_HAVE_METRICS
        if (ins->metrics) {
            flb_metrics_sum(FLB_METRIC_OUT_OK_RECORDS, task->records, ins->metrics);
            flb_metrics_sum(FLB_METRIC_OUT_OK_BYTES, task->size, ins->metrics);
        }
#endif
        /* Inform the user if a 'retry' succedeed */
        if (mk_list_size(&task->retries) > 0) {
            retries = flb_task_retry_count(task, ins);
            if (retries > 0) {
                flb_info("[engine] flush chunk '%s' succeeded at retry %i: "
                         "task_id=%i, input=%s > output=%s (out_id=%i)",
                         flb_input_chunk_get_name(task->ic),
                         retries, task_id,
                         flb_input_name(task->i_ins),
                         flb_output_name(ins), out_id);
            }
        }<commit_msg>engine: report finished coroutine only if is not threaded mode<commit_after>/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2019-2020 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 <stdio.h>
#include <stdlib.h>

#include <monkey/mk_core.h>
#include <fluent-bit/flb_time.h>
#include <fluent-bit/flb_info.h>
#include <fluent-bit/flb_bits.h>

#include <fluent-bit/flb_macros.h>
#include <fluent-bit/flb_pipe.h>
#include <fluent-bit/flb_input.h>
#include <fluent-bit/flb_output.h>
#include <fluent-bit/flb_error.h>
#include <fluent-bit/flb_utils.h>
#include <fluent-bit/flb_config.h>
#include <fluent-bit/flb_engine.h>
#include <fluent-bit/flb_engine_dispatch.h>
#include <fluent-bit/flb_task.h>
#include <fluent-bit/flb_router.h>
#include <fluent-bit/flb_http_server.h>
#include <fluent-bit/flb_scheduler.h>
#include <fluent-bit/flb_parser.h>
#include <fluent-bit/flb_sosreport.h>
#include <fluent-bit/flb_storage.h>
#include <fluent-bit/flb_http_server.h>

#ifdef FLB_HAVE_METRICS
#include <fluent-bit/flb_metrics_exporter.h>
#endif

#ifdef FLB_HAVE_STREAM_PROCESSOR
#include <fluent-bit/stream_processor/flb_sp.h>
#endif

FLB_TLS_DEFINE(struct mk_event_loop, flb_engine_evl);


void flb_engine_evl_init()
{
    FLB_TLS_INIT(flb_engine_evl);
}

struct mk_event_loop *flb_engine_evl_get()
{
    struct mk_event_loop *evl;

    evl = FLB_TLS_GET(flb_engine_evl);
    return evl;
}

void flb_engine_evl_set(struct mk_event_loop *evl)
{
    FLB_TLS_SET(flb_engine_evl, evl);
}

int flb_engine_destroy_tasks(struct mk_list *tasks)
{
    int c = 0;
    struct mk_list *tmp;
    struct mk_list *head;
    struct flb_task *task;

    mk_list_foreach_safe(head, tmp, tasks) {
        task = mk_list_entry(head, struct flb_task, _head);
        flb_task_destroy(task, FLB_FALSE);
        c++;
    }

    return c;
}

int flb_engine_flush(struct flb_config *config,
                     struct flb_input_plugin *in_force)
{
    struct flb_input_instance *in;
    struct flb_input_plugin *p;
    struct mk_list *head;

    mk_list_foreach(head, &config->inputs) {
        in = mk_list_entry(head, struct flb_input_instance, _head);
        p = in->p;

        if (in_force != NULL && p != in_force) {
            continue;
        }
        flb_engine_dispatch(0, in, config);
    }

    return 0;
}

/* Cleanup function that runs every 1.5 second */
static void cb_engine_sched_timer(struct flb_config *ctx, void *data)
{
    (void) data;

    /* Upstream connections timeouts handling */
    flb_upstream_conn_timeouts(&ctx->upstreams);
}

static inline int handle_output_event(flb_pipefd_t fd, struct flb_config *config)
{
    int ret;
    int bytes;
    int task_id;
    int out_id;
    int retries;
    int retry_seconds;
    uint32_t type;
    uint32_t key;
    uint64_t val;
    struct flb_task *task;
    struct flb_task_retry *retry;
    struct flb_output_instance *ins;

    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);
    key  = FLB_BITS_U64_LOW(val);

    if (type != FLB_ENGINE_TASK) {
        flb_error("[engine] invalid event type %i for output handler",
                  type);
        return -1;
    }

    /*
     * The notion of ENGINE_TASK is associated to outputs. All thread
     * references below belongs to flb_output_coro's.
     */
    ret     = FLB_TASK_RET(key);
    task_id = FLB_TASK_ID(key);
    out_id  = FLB_TASK_OUT(key);

#ifdef FLB_HAVE_TRACE
    char *trace_st = NULL;

    if (ret == FLB_OK) {
        trace_st = "OK";
    }
    else if (ret == FLB_ERROR) {
        trace_st = "ERROR";
    }
    else if (ret == FLB_RETRY) {
        trace_st = "RETRY";
    }

    flb_trace("%s[engine] [task event]%s task_id=%i out_id=%i return=%s",
              ANSI_YELLOW, ANSI_RESET,
              task_id, out_id, trace_st);
#endif

    task = config->tasks_map[task_id].task;
    ins  = flb_output_get_instance(config, out_id);
    if (flb_output_is_threaded(ins) == FLB_FALSE) {
        flb_output_flush_finished(config, out_id);
    }

    /* A thread has finished, delete it */
    if (ret == FLB_OK) {

        /* Update metrics */
#ifdef FLB_HAVE_METRICS
        if (ins->metrics) {
            flb_metrics_sum(FLB_METRIC_OUT_OK_RECORDS, task->records, ins->metrics);
            flb_metrics_sum(FLB_METRIC_OUT_OK_BYTES, task->size, ins->metrics);
        }
#endif
        /* Inform the user if a 'retry' succedeed */
        if (mk_list_size(&task->retries) > 0) {
            retries = flb_task_retry_count(task, ins);
            if (retries > 0) {
                flb_info("[engine] flush chunk '%s' succeeded at retry %i: "
                         "task_id=%i, input=%s > output=%s (out_id=%i)",
                         flb_input_chunk_get_name(task->ic),
                         retries, task_id,
                         flb_input_name(task->i_ins),
                         flb_output_name(ins), out_id);
            }
        }
{'fraction_non_alphanumeric': 0.11570764832022873, 'fraction_numerical': 0.007505360972122945, 'mean_word_length': 2.8227459016393444, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 83, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '5853763', 'n_tokens_mistral': 4204, 'n_tokens_neox': 4090, 'n_words': 912}