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

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2015-2016 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.
 */

#ifndef FLB_OUTPUT_H
#define FLB_OUTPUT_H

#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE
#endif

#include <fluent-bit/flb_info.h>
#include <fluent-bit/flb_bits.h>
#include <fluent-bit/flb_io.h>
#include <fluent-bit/flb_stats.h>
#include <fluent-bit/flb_config.h>
#include <fluent-bit/flb_network.h>
#include <fluent-bit/flb_engine.h>
#include <fluent-bit/flb_task.h>

#include <unistd.h>

/* Output plugin masks */
#define FLB_OUTPUT_NET         32  /* output address may set host and port */

struct flb_output_instance;

struct flb_output_plugin {
    int flags;

    /* The plugin name */
    char *name;

    /* Plugin description */
    char *description;

    /*
     * Output network info:
     *
     * An output plugin can be specified just using it shortname or using the
     * complete network address format, e.g:
     *
     *  $ fluent-bit -i cpu -o plugin://hostname:port/uri
     *
     * where:
     *
     *   plugin   = the output plugin shortname
     *   name     = IP address or hostname of the target
     *   port     = target TCP port
     *   uri      = extra information that may be used by the plugin
     */
    struct flb_net_host host;

    /* Socket connection */
    //int conn;

    /* Initalization */
    int (*cb_init)    (struct flb_output_instance *, struct flb_config *, void *);

    /* Pre run */
    int (*cb_pre_run) (void *, struct flb_config *);

    /* Flush callback */
    int (*cb_flush) (void *, size_t,
                     char *, int,
                     struct flb_input_instance *,
                     void *,
                     struct flb_config *);

    /* Exit */
    int (*cb_exit) (void *, struct flb_config *);

    /* Link to global list from flb_config->outputs */
    struct mk_list _head;
};

/*
 * Each initialized plugin must have an instance, same plugin may be
 * loaded more than one time.
 *
 * An instance try to contain plugin data separating what is fixed data
 * and the variable one that is generated when the plugin is invoked.
 */
struct flb_output_instance {
    uint64_t mask_id;                    /* internal bitmask for routing */
    char name[16];                       /* numbered name (cpu -> cpu.0) */
    struct flb_output_plugin *p;         /* original plugin              */
    void *context;                       /* plugin configuration context */

    /* Plugin properties */
    int retry_limit;                     /* max of retries allowed       */
    int use_tls;                         /* bool, try to use TLS for I/O */
    char *match;                         /* match rule for tag/routing   */

#ifdef FLB_HAVE_TLS
    int tls_verify;                      /* Verify certs (default: true) */
    char *tls_ca_file;                   /* CA root cert                 */
    char *tls_crt_file;                  /* Certificate                  */
    char *tls_key_file;                  /* Cert Key                     */
    char *tls_key_passwd;                /* Cert Key Password            */
#endif

    /*
     * network info:
     *
     * An input plugin can be specified just using it shortname or using the
     * complete network address format, e.g:
     *
     *  $ fluent-bit -i cpu -o plugin://hostname:port/uri
     *
     * where:
     *
     *   plugin   = the output plugin shortname
     *   name     = IP address or hostname of the target
     *   port     = target TCP port
     *   uri      = extra information that may be used by the plugin
     */
    struct flb_net_host host;

    /*
     * Optional data passed to the plugin, this info is useful when
     * running Fluent Bit in library mode and the target plugin needs
     * some specific data from it caller.
     */
    void *data;

    /* Output handler configuration */
    void *out_context;

    /* IO upstream context, if flags & (FLB_OUTPUT_TCP | FLB_OUTPUT TLS)) */
    struct flb_upstream *upstream;



    /*
     * The threads_queue is the head for the linked list that holds co-routines
     * nodes information that needs to be processed.
     */
    struct mk_list th_queue;

#ifdef FLB_HAVE_STATS
    int stats_fd;
#endif

#ifdef FLB_HAVE_TLS
    struct flb_tls tls;
#else
    void *tls;
#endif

    struct mk_list properties;           /* properties / configuration   */
    struct mk_list _head;                /* link to config->inputs       */
};

struct flb_output_thread {
    int id;                            /* out-thread ID      */
    int retries;                       /* number of retries  */
    void *buffer;                      /* output buffer      */
    struct flb_task *task;             /* Parent flb_task    */
    struct flb_config *config;         /* FLB context        */
    struct flb_output_instance *o_ins; /* output instance    */
    struct flb_thread *parent;         /* parent thread addr */
    struct mk_list _head;              /* Link to struct flb_task->threads */
};

#ifdef FLB_HAVE_FLUSH_UCONTEXT

static FLB_INLINE
struct flb_output_thread *flb_output_thread_get(int id, struct flb_task *task)
{
    struct mk_list *head;
    struct flb_output_thread *out_th = NULL;

    mk_list_foreach(head, &task->threads) {
        out_th = mk_list_entry(head, struct flb_output_thread, _head);
        if (out_th->id == id) {
            return out_th;
        }
    }

    return NULL;
}

static FLB_INLINE int flb_output_thread_destroy_id(int id, struct flb_task *task)
{
    struct flb_output_thread *out_th;
    struct flb_thread *thread;

    out_th = flb_output_thread_get(id, task);
    if (!out_th) {
        return -1;
    }

    thread = out_th->parent;
    flb_thread_destroy(thread);

    return 0;
}

/* When an output_thread is going to be destroyed, this callback is triggered */
static void cb_output_thread_destroy(void *data)<commit_msg>output: on destroying thread, decrement the task users counter (#113)<commit_after>/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2015-2016 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.
 */

#ifndef FLB_OUTPUT_H
#define FLB_OUTPUT_H

#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE
#endif

#include <fluent-bit/flb_info.h>
#include <fluent-bit/flb_bits.h>
#include <fluent-bit/flb_io.h>
#include <fluent-bit/flb_stats.h>
#include <fluent-bit/flb_config.h>
#include <fluent-bit/flb_network.h>
#include <fluent-bit/flb_engine.h>
#include <fluent-bit/flb_task.h>

#include <unistd.h>

/* Output plugin masks */
#define FLB_OUTPUT_NET         32  /* output address may set host and port */

struct flb_output_instance;

struct flb_output_plugin {
    int flags;

    /* The plugin name */
    char *name;

    /* Plugin description */
    char *description;

    /*
     * Output network info:
     *
     * An output plugin can be specified just using it shortname or using the
     * complete network address format, e.g:
     *
     *  $ fluent-bit -i cpu -o plugin://hostname:port/uri
     *
     * where:
     *
     *   plugin   = the output plugin shortname
     *   name     = IP address or hostname of the target
     *   port     = target TCP port
     *   uri      = extra information that may be used by the plugin
     */
    struct flb_net_host host;

    /* Socket connection */
    //int conn;

    /* Initalization */
    int (*cb_init)    (struct flb_output_instance *, struct flb_config *, void *);

    /* Pre run */
    int (*cb_pre_run) (void *, struct flb_config *);

    /* Flush callback */
    int (*cb_flush) (void *, size_t,
                     char *, int,
                     struct flb_input_instance *,
                     void *,
                     struct flb_config *);

    /* Exit */
    int (*cb_exit) (void *, struct flb_config *);

    /* Link to global list from flb_config->outputs */
    struct mk_list _head;
};

/*
 * Each initialized plugin must have an instance, same plugin may be
 * loaded more than one time.
 *
 * An instance try to contain plugin data separating what is fixed data
 * and the variable one that is generated when the plugin is invoked.
 */
struct flb_output_instance {
    uint64_t mask_id;                    /* internal bitmask for routing */
    char name[16];                       /* numbered name (cpu -> cpu.0) */
    struct flb_output_plugin *p;         /* original plugin              */
    void *context;                       /* plugin configuration context */

    /* Plugin properties */
    int retry_limit;                     /* max of retries allowed       */
    int use_tls;                         /* bool, try to use TLS for I/O */
    char *match;                         /* match rule for tag/routing   */

#ifdef FLB_HAVE_TLS
    int tls_verify;                      /* Verify certs (default: true) */
    char *tls_ca_file;                   /* CA root cert                 */
    char *tls_crt_file;                  /* Certificate                  */
    char *tls_key_file;                  /* Cert Key                     */
    char *tls_key_passwd;                /* Cert Key Password            */
#endif

    /*
     * network info:
     *
     * An input plugin can be specified just using it shortname or using the
     * complete network address format, e.g:
     *
     *  $ fluent-bit -i cpu -o plugin://hostname:port/uri
     *
     * where:
     *
     *   plugin   = the output plugin shortname
     *   name     = IP address or hostname of the target
     *   port     = target TCP port
     *   uri      = extra information that may be used by the plugin
     */
    struct flb_net_host host;

    /*
     * Optional data passed to the plugin, this info is useful when
     * running Fluent Bit in library mode and the target plugin needs
     * some specific data from it caller.
     */
    void *data;

    /* Output handler configuration */
    void *out_context;

    /* IO upstream context, if flags & (FLB_OUTPUT_TCP | FLB_OUTPUT TLS)) */
    struct flb_upstream *upstream;



    /*
     * The threads_queue is the head for the linked list that holds co-routines
     * nodes information that needs to be processed.
     */
    struct mk_list th_queue;

#ifdef FLB_HAVE_STATS
    int stats_fd;
#endif

#ifdef FLB_HAVE_TLS
    struct flb_tls tls;
#else
    void *tls;
#endif

    struct mk_list properties;           /* properties / configuration   */
    struct mk_list _head;                /* link to config->inputs       */
};

struct flb_output_thread {
    int id;                            /* out-thread ID      */
    int retries;                       /* number of retries  */
    void *buffer;                      /* output buffer      */
    struct flb_task *task;             /* Parent flb_task    */
    struct flb_config *config;         /* FLB context        */
    struct flb_output_instance *o_ins; /* output instance    */
    struct flb_thread *parent;         /* parent thread addr */
    struct mk_list _head;              /* Link to struct flb_task->threads */
};

#ifdef FLB_HAVE_FLUSH_UCONTEXT

static FLB_INLINE
struct flb_output_thread *flb_output_thread_get(int id, struct flb_task *task)
{
    struct mk_list *head;
    struct flb_output_thread *out_th = NULL;

    mk_list_foreach(head, &task->threads) {
        out_th = mk_list_entry(head, struct flb_output_thread, _head);
        if (out_th->id == id) {
            return out_th;
        }
    }

    return NULL;
}

static FLB_INLINE int flb_output_thread_destroy_id(int id, struct flb_task *task)
{
    struct flb_output_thread *out_th;
    struct flb_thread *thread;

    out_th = flb_output_thread_get(id, task);
    if (!out_th) {
        return -1;
    }

    thread = out_th->parent;
    flb_thread_destroy(thread);
    task->users--;

    return 0;
}

/* When an output_thread is going to be destroyed, this callback is triggered */
static void cb_output_thread_destroy(void *data)
{'fraction_non_alphanumeric': 0.09189806009889692, 'fraction_numerical': 0.0037276531000380374, 'mean_word_length': 2.0068618481244282, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5876702', 'n_tokens_mistral': 3980, 'n_tokens_neox': 3826, 'n_words': 1275}