diff options
Diffstat (limited to 'fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native')
15 files changed, 1158 insertions, 0 deletions
diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection.inl b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection.inl new file mode 100644 index 000000000..b2d01aa9f --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection.inl @@ -0,0 +1,9 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +EXPORT_WASM_API_WITH_SIG(wasm_open_connection, "($*~)i"), +EXPORT_WASM_API_WITH_SIG(wasm_close_connection, "(i)"), +EXPORT_WASM_API_WITH_SIG(wasm_send_on_connection, "(i*~)i"), +EXPORT_WASM_API_WITH_SIG(wasm_config_connection, "(i*~)i"), diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection_lib.h b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection_lib.h new file mode 100644 index 000000000..3e182cbb8 --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection_lib.h @@ -0,0 +1,75 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +#ifndef CONNECTION_LIB_H_ +#define CONNECTION_LIB_H_ + +#include "bi-inc/attr_container.h" +#include "wasm_export.h" + +#ifdef __cplusplus +extern "C" { +#endif + +/** + * This file defines connection library which should be implemented by + * different platforms + */ + +/* + * @brief Open a connection. + * + * @param name name of the connection, "TCP", "UDP" or "UART" + * @param args connection arguments, such as: ip:127.0.0.1, port:8888 + * + * @return 0~0xFFFFFFFE means id of the connection, otherwise(-1) means fail + */ +typedef uint32 (*connection_open_f)(wasm_module_inst_t module_inst, + const char *name, attr_container_t *args); + +/* + * @brief Close a connection. + * + * @param handle of the connection + */ +typedef void (*connection_close_f)(uint32 handle); + +/* + * @brief Send data to the connection in non-blocking manner. + * + * @param handle of the connection + * @param data data buffer to be sent + * @param len length of the data in byte + * + * @return actual length sent, -1 if fail + */ +typedef int (*connection_send_f)(uint32 handle, const char *data, int len); + +/* + * @brief Configure connection. + * + * @param handle of the connection + * @param cfg configurations + * + * @return true if success, false otherwise + */ +typedef bool (*connection_config_f)(uint32 handle, attr_container_t *cfg); + +/* Raw connection interface for platform to implement */ +typedef struct _connection_interface { + connection_open_f _open; + connection_close_f _close; + connection_send_f _send; + connection_config_f _config; +} connection_interface_t; + +/* Platform must define this interface */ +extern connection_interface_t connection_impl; + +#ifdef __cplusplus +} +#endif + +#endif /* CONNECTION_LIB_H_ */ diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection_native_api.h b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection_native_api.h new file mode 100644 index 000000000..42a2508f1 --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection_native_api.h @@ -0,0 +1,36 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +#ifndef CONNECTION_API_H_ +#define CONNECTION_API_H_ + +#include "bh_platform.h" +#include "wasm_export.h" + +#ifdef __cplusplus +extern "C" { +#endif + +/* + * connection interfaces + */ + +uint32 +wasm_open_connection(wasm_exec_env_t exec_env, char *name, char *args_buf, + uint32 len); +void +wasm_close_connection(wasm_exec_env_t exec_env, uint32 handle); +int +wasm_send_on_connection(wasm_exec_env_t exec_env, uint32 handle, char *data, + uint32 len); +bool +wasm_config_connection(wasm_exec_env_t exec_env, uint32 handle, char *cfg_buf, + uint32 len); + +#ifdef __cplusplus +} +#endif + +#endif /* end of CONNECTION_API_H_ */ diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection_wrapper.c b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection_wrapper.c new file mode 100644 index 000000000..7c20b51d0 --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/connection_wrapper.c @@ -0,0 +1,61 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +#include "connection_lib.h" +#include "wasm_export.h" +#include "native_interface.h" +#include "connection_native_api.h" + +/* Note: + * + * This file is the consumer of connection lib which is implemented by different + * platforms + */ + +uint32 +wasm_open_connection(wasm_exec_env_t exec_env, char *name, char *args_buf, + uint32 len) +{ + wasm_module_inst_t module_inst = get_module_inst(exec_env); + attr_container_t *args; + + args = (attr_container_t *)args_buf; + + if (connection_impl._open != NULL) + return connection_impl._open(module_inst, name, args); + + return -1; +} + +void +wasm_close_connection(wasm_exec_env_t exec_env, uint32 handle) +{ + if (connection_impl._close != NULL) + connection_impl._close(handle); +} + +int +wasm_send_on_connection(wasm_exec_env_t exec_env, uint32 handle, char *data, + uint32 len) +{ + if (connection_impl._send != NULL) + return connection_impl._send(handle, data, len); + + return -1; +} + +bool +wasm_config_connection(wasm_exec_env_t exec_env, uint32 handle, char *cfg_buf, + uint32 len) +{ + attr_container_t *cfg; + + cfg = (attr_container_t *)cfg_buf; + + if (connection_impl._config != NULL) + return connection_impl._config(handle, cfg); + + return false; +} diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_tcp.c b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_tcp.c new file mode 100644 index 000000000..054eb59fd --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_tcp.c @@ -0,0 +1,54 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +#include "conn_tcp.h" + +#include <sys/socket.h> +#include <netdb.h> +#include <arpa/inet.h> +#include <fcntl.h> +#include <unistd.h> + +int +tcp_open(char *address, uint16 port) +{ + int sock, ret; + struct sockaddr_in servaddr; + + memset(&servaddr, 0, sizeof(servaddr)); + servaddr.sin_family = AF_INET; + servaddr.sin_addr.s_addr = inet_addr(address); + servaddr.sin_port = htons(port); + + sock = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if (sock == -1) + return -1; + + ret = connect(sock, (struct sockaddr *)&servaddr, sizeof(servaddr)); + if (ret == -1) { + close(sock); + return -1; + } + + /* Put the socket in non-blocking mode */ + if (fcntl(sock, F_SETFL, fcntl(sock, F_GETFL) | O_NONBLOCK) < 0) { + close(sock); + return -1; + } + + return sock; +} + +int +tcp_send(int sock, const char *data, int size) +{ + return send(sock, data, size, 0); +} + +int +tcp_recv(int sock, char *buffer, int buf_size) +{ + return recv(sock, buffer, buf_size, 0); +} diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_tcp.h b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_tcp.h new file mode 100644 index 000000000..c4d5cc86a --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_tcp.h @@ -0,0 +1,28 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +#ifndef CONN_LINUX_TCP_H_ +#define CONN_LINUX_TCP_H_ + +#include "bh_platform.h" + +#ifdef __cplusplus +extern "C" { +#endif + +int +tcp_open(char *address, uint16 port); + +int +tcp_send(int sock, const char *data, int size); + +int +tcp_recv(int sock, char *buffer, int buf_size); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_uart.c b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_uart.c new file mode 100644 index 000000000..0bcdc93f7 --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_uart.c @@ -0,0 +1,103 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +#include "conn_uart.h" + +#include <fcntl.h> +#include <termios.h> +#include <unistd.h> + +static int +parse_baudrate(int baud) +{ + switch (baud) { + case 9600: + return B9600; + case 19200: + return B19200; + case 38400: + return B38400; + case 57600: + return B57600; + case 115200: + return B115200; + case 230400: + return B230400; + case 460800: + return B460800; + case 500000: + return B500000; + case 576000: + return B576000; + case 921600: + return B921600; + case 1000000: + return B1000000; + case 1152000: + return B1152000; + case 1500000: + return B1500000; + case 2000000: + return B2000000; + case 2500000: + return B2500000; + case 3000000: + return B3000000; + case 3500000: + return B3500000; + case 4000000: + return B4000000; + default: + return -1; + } +} + +int +uart_open(char *device, int baudrate) +{ + int uart_fd; + struct termios uart_term; + + uart_fd = open(device, O_RDWR | O_NOCTTY); + + if (uart_fd < 0) + return -1; + + memset(&uart_term, 0, sizeof(uart_term)); + uart_term.c_cflag = parse_baudrate(baudrate) | CS8 | CLOCAL | CREAD; + uart_term.c_iflag = IGNPAR; + uart_term.c_oflag = 0; + + /* set noncanonical mode */ + uart_term.c_lflag = 0; + uart_term.c_cc[VTIME] = 30; + uart_term.c_cc[VMIN] = 1; + tcflush(uart_fd, TCIFLUSH); + + if (tcsetattr(uart_fd, TCSANOW, &uart_term) != 0) { + close(uart_fd); + return -1; + } + + /* Put the fd in non-blocking mode */ + if (fcntl(uart_fd, F_SETFL, fcntl(uart_fd, F_GETFL) | O_NONBLOCK) < 0) { + close(uart_fd); + return -1; + } + + return uart_fd; +} + +int +uart_send(int fd, const char *data, int size) +{ + return write(fd, data, size); +} + +int +uart_recv(int fd, char *buffer, int buf_size) +{ + return read(fd, buffer, buf_size); +} diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_uart.h b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_uart.h new file mode 100644 index 000000000..443167026 --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_uart.h @@ -0,0 +1,28 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +#ifndef CONN_LINUX_UART_H_ +#define CONN_LINUX_UART_H_ + +#include "bh_platform.h" + +#ifdef __cplusplus +extern "C" { +#endif + +int +uart_open(char *device, int baudrate); + +int +uart_send(int fd, const char *data, int size); + +int +uart_recv(int fd, char *buffer, int buf_size); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_udp.c b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_udp.c new file mode 100644 index 000000000..61652b14d --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_udp.c @@ -0,0 +1,58 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +#include "conn_udp.h" + +#include <sys/socket.h> +#include <netdb.h> +#include <arpa/inet.h> +#include <fcntl.h> +#include <unistd.h> + +int +udp_open(uint16 port) +{ + int sock, ret; + struct sockaddr_in addr; + + sock = socket(AF_INET, SOCK_DGRAM, 0); + if (sock == -1) + return -1; + + memset(&addr, 0, sizeof(addr)); + addr.sin_family = AF_INET; + addr.sin_addr.s_addr = htonl(INADDR_ANY); + addr.sin_port = htons(port); + + ret = bind(sock, (struct sockaddr *)&addr, sizeof(addr)); + if (ret == -1) { + close(sock); + return -1; + } + + /* Put the socket in non-blocking mode */ + if (fcntl(sock, F_SETFL, fcntl(sock, F_GETFL) | O_NONBLOCK) < 0) { + close(sock); + return -1; + } + + return sock; +} + +int +udp_send(int sock, struct sockaddr *dest, const char *data, int size) +{ + return sendto(sock, data, size, MSG_CONFIRM, dest, sizeof(*dest)); +} + +int +udp_recv(int sock, char *buffer, int buf_size) +{ + struct sockaddr_in remaddr; + socklen_t addrlen = sizeof(remaddr); + + return recvfrom(sock, buffer, buf_size, 0, (struct sockaddr *)&remaddr, + &addrlen); +} diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_udp.h b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_udp.h new file mode 100644 index 000000000..377c26eb1 --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/conn_udp.h @@ -0,0 +1,28 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +#ifndef CONN_LINUX_UDP_H_ +#define CONN_LINUX_UDP_H_ + +#include "bh_platform.h" + +#ifdef __cplusplus +extern "C" { +#endif + +int +udp_open(uint16 port); + +int +udp_send(int sock, struct sockaddr *dest, const char *data, int size); + +int +udp_recv(int sock, char *buffer, int buf_size); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/connection_mgr.c b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/connection_mgr.c new file mode 100644 index 000000000..001446206 --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/connection_mgr.c @@ -0,0 +1,609 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +/* + * Note: + * This file implements the linux version connection library which is + * defined in connection_lib.h. + * It also provides a reference implementation of connections manager. + */ + +#include "connection_lib.h" +#include "bh_platform.h" +#include "app_manager_export.h" +#include "module_wasm_app.h" +#include "conn_tcp.h" +#include "conn_udp.h" +#include "conn_uart.h" + +#include <unistd.h> +#include <sys/epoll.h> +#include <sys/types.h> +#include <arpa/inet.h> +#include <fcntl.h> + +#define MAX_EVENTS 10 +#define IO_BUF_SIZE 256 + +static bool polling_thread_run = true; + +/* Connection type */ +typedef enum conn_type { + CONN_TYPE_TCP, + CONN_TYPE_UDP, + CONN_TYPE_UART, + CONN_TYPE_UNKNOWN +} conn_type_t; + +/* Sys connection */ +typedef struct sys_connection { + /* Next connection */ + struct sys_connection *next; + + /* Type */ + conn_type_t type; + + /* Handle to interact with wasm app */ + uint32 handle; + + /* Underlying connection ID, may be socket fd */ + int fd; + + /* Module id that the connection belongs to */ + uint32 module_id; + + /* Argument, such as dest addr for udp */ + void *arg; +} sys_connection_t; + +/* Epoll instance */ +static int epollfd; + +/* Connections list */ +static sys_connection_t *g_connections = NULL; + +/* Max handle */ +static uint32 g_handle_max = 0; + +/* Lock to protect g_connections and g_handle_max */ +static korp_mutex g_lock; + +/* Epoll events */ +static struct epoll_event epoll_events[MAX_EVENTS]; + +/* Buffer to receive data */ +static char io_buf[IO_BUF_SIZE]; + +static uint32 +_conn_open(wasm_module_inst_t module_inst, const char *name, + attr_container_t *args); +static void +_conn_close(uint32 handle); +static int +_conn_send(uint32 handle, const char *data, int len); +static bool +_conn_config(uint32 handle, attr_container_t *cfg); + +/* clang-format off */ +/* + * Platform implementation of connection library + */ +connection_interface_t connection_impl = { + ._open = _conn_open, + ._close = _conn_close, + ._send = _conn_send, + ._config = _conn_config +}; +/* clang-format on */ + +static void +add_connection(sys_connection_t *conn) +{ + os_mutex_lock(&g_lock); + + g_handle_max++; + if (g_handle_max == -1) + g_handle_max++; + conn->handle = g_handle_max; + + if (g_connections) { + conn->next = g_connections; + g_connections = conn; + } + else { + g_connections = conn; + } + + os_mutex_unlock(&g_lock); +} + +#define FREE_CONNECTION(conn) \ + do { \ + if (conn->arg) \ + wasm_runtime_free(conn->arg); \ + wasm_runtime_free(conn); \ + } while (0) + +static int +get_app_conns_num(uint32 module_id) +{ + sys_connection_t *conn; + int num = 0; + + os_mutex_lock(&g_lock); + + conn = g_connections; + while (conn) { + if (conn->module_id == module_id) + num++; + conn = conn->next; + } + + os_mutex_unlock(&g_lock); + + return num; +} + +static sys_connection_t * +find_connection(uint32 handle, bool remove_found) +{ + sys_connection_t *conn, *prev = NULL; + + os_mutex_lock(&g_lock); + + conn = g_connections; + while (conn) { + if (conn->handle == handle) { + if (remove_found) { + if (prev != NULL) { + prev->next = conn->next; + } + else { + g_connections = conn->next; + } + } + os_mutex_unlock(&g_lock); + return conn; + } + else { + prev = conn; + conn = conn->next; + } + } + + os_mutex_unlock(&g_lock); + + return NULL; +} + +static void +cleanup_connections(uint32 module_id) +{ + sys_connection_t *conn, *prev = NULL; + + os_mutex_lock(&g_lock); + + conn = g_connections; + while (conn) { + if (conn->module_id == module_id) { + epoll_ctl(epollfd, EPOLL_CTL_DEL, conn->fd, NULL); + close(conn->fd); + + if (prev != NULL) { + prev->next = conn->next; + FREE_CONNECTION(conn); + conn = prev->next; + } + else { + g_connections = conn->next; + FREE_CONNECTION(conn); + conn = g_connections; + } + } + else { + prev = conn; + conn = conn->next; + } + } + + os_mutex_unlock(&g_lock); +} + +static conn_type_t +get_conn_type(const char *name) +{ + if (strcmp(name, "TCP") == 0) + return CONN_TYPE_TCP; + if (strcmp(name, "UDP") == 0) + return CONN_TYPE_UDP; + if (strcmp(name, "UART") == 0) + return CONN_TYPE_UART; + + return CONN_TYPE_UNKNOWN; +} + +/* --- connection lib function --- */ +static uint32 +_conn_open(wasm_module_inst_t module_inst, const char *name, + attr_container_t *args) +{ + int fd; + sys_connection_t *conn; + struct epoll_event ev; + uint32 module_id = app_manager_get_module_id(Module_WASM_App, module_inst); + bh_assert(module_id != ID_NONE); + + if (get_app_conns_num(module_id) >= MAX_CONNECTION_PER_APP) + return -1; + + conn = (sys_connection_t *)wasm_runtime_malloc(sizeof(*conn)); + if (conn == NULL) + return -1; + + memset(conn, 0, sizeof(*conn)); + conn->module_id = module_id; + conn->type = get_conn_type(name); + + /* Generate a handle and add to list */ + add_connection(conn); + + if (conn->type == CONN_TYPE_TCP) { + char *address; + uint16 port; + + /* Check and parse connection parameters */ + if (!attr_container_contain_key(args, "address") + || !attr_container_contain_key(args, "port")) + goto fail; + + address = attr_container_get_as_string(args, "address"); + port = attr_container_get_as_uint16(args, "port"); + + /* Connect to TCP server */ + if (!address || (fd = tcp_open(address, port)) == -1) + goto fail; + } + else if (conn->type == CONN_TYPE_UDP) { + uint16 port; + + /* Check and parse connection parameters */ + if (!attr_container_contain_key(args, "bind port")) + goto fail; + port = attr_container_get_as_uint16(args, "bind port"); + + /* Bind port */ + if ((fd = udp_open(port)) == -1) + goto fail; + } + else if (conn->type == CONN_TYPE_UART) { + char *device; + int baud; + + /* Check and parse connection parameters */ + if (!attr_container_contain_key(args, "device") + || !attr_container_contain_key(args, "baudrate")) + goto fail; + device = attr_container_get_as_string(args, "device"); + baud = attr_container_get_as_int(args, "baudrate"); + + /* Open device */ + if (!device || (fd = uart_open(device, baud)) == -1) + goto fail; + } + else { + goto fail; + } + + conn->fd = fd; + + /* Set current connection as event data */ + ev.events = EPOLLIN; + ev.data.ptr = conn; + + /* Monitor incoming data */ + if (epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &ev) == -1) { + close(fd); + goto fail; + } + + return conn->handle; + +fail: + find_connection(conn->handle, true); + wasm_runtime_free(conn); + return -1; +} + +/* --- connection lib function --- */ +static void +_conn_close(uint32 handle) +{ + sys_connection_t *conn = find_connection(handle, true); + + if (conn != NULL) { + epoll_ctl(epollfd, EPOLL_CTL_DEL, conn->fd, NULL); + close(conn->fd); + FREE_CONNECTION(conn); + } +} + +/* --- connection lib function --- */ +static int +_conn_send(uint32 handle, const char *data, int len) +{ + sys_connection_t *conn = find_connection(handle, false); + + if (conn == NULL) + return -1; + + if (conn->type == CONN_TYPE_TCP) + return tcp_send(conn->fd, data, len); + + if (conn->type == CONN_TYPE_UDP) { + struct sockaddr *addr = (struct sockaddr *)conn->arg; + return udp_send(conn->fd, addr, data, len); + } + + if (conn->type == CONN_TYPE_UART) + return uart_send(conn->fd, data, len); + + return -1; +} + +/* --- connection lib function --- */ +static bool +_conn_config(uint32 handle, attr_container_t *cfg) +{ + sys_connection_t *conn = find_connection(handle, false); + + if (conn == NULL) + return false; + + if (conn->type == CONN_TYPE_UDP) { + char *address; + uint16_t port; + struct sockaddr_in *addr; + + /* Parse remote address/port */ + if (!attr_container_contain_key(cfg, "address") + || !attr_container_contain_key(cfg, "port")) + return false; + if (!(address = attr_container_get_as_string(cfg, "address"))) + return false; + port = attr_container_get_as_uint16(cfg, "port"); + + if (conn->arg == NULL) { + addr = (struct sockaddr_in *)wasm_runtime_malloc(sizeof(*addr)); + if (addr == NULL) + return false; + + memset(addr, 0, sizeof(*addr)); + addr->sin_family = AF_INET; + addr->sin_addr.s_addr = inet_addr(address); + addr->sin_port = htons(port); + + /* Set remote address as connection arg */ + conn->arg = addr; + } + else { + addr = (struct sockaddr_in *)conn->arg; + addr->sin_addr.s_addr = inet_addr(address); + addr->sin_port = htons(port); + } + + return true; + } + + return false; +} + +/* --- connection manager reference implementation ---*/ + +typedef struct connection_event { + uint32 handle; + char *data; + uint32 len; +} connection_event_t; + +static void +connection_event_cleaner(connection_event_t *conn_event) +{ + if (conn_event->data != NULL) + wasm_runtime_free(conn_event->data); + wasm_runtime_free(conn_event); +} + +static void +post_msg_to_module(sys_connection_t *conn, char *data, uint32 len) +{ + module_data *module = module_data_list_lookup_id(conn->module_id); + char *data_copy = NULL; + connection_event_t *conn_data_event; + bh_message_t msg; + + if (module == NULL) + return; + + conn_data_event = + (connection_event_t *)wasm_runtime_malloc(sizeof(*conn_data_event)); + if (conn_data_event == NULL) + return; + + if (len > 0) { + data_copy = (char *)wasm_runtime_malloc(len); + if (data_copy == NULL) { + wasm_runtime_free(conn_data_event); + return; + } + bh_memcpy_s(data_copy, len, data, len); + } + + memset(conn_data_event, 0, sizeof(*conn_data_event)); + conn_data_event->handle = conn->handle; + conn_data_event->data = data_copy; + conn_data_event->len = len; + + msg = bh_new_msg(CONNECTION_EVENT_WASM, conn_data_event, + sizeof(*conn_data_event), connection_event_cleaner); + if (!msg) { + connection_event_cleaner(conn_data_event); + return; + } + + bh_post_msg2(module->queue, msg); +} + +static void * +polling_thread_routine(void *arg) +{ + while (polling_thread_run) { + int i, n; + + n = epoll_wait(epollfd, epoll_events, MAX_EVENTS, -1); + + if (n == -1 && errno != EINTR) + continue; + + for (i = 0; i < n; i++) { + sys_connection_t *conn = + (sys_connection_t *)epoll_events[i].data.ptr; + + if (conn->type == CONN_TYPE_TCP) { + int count = tcp_recv(conn->fd, io_buf, IO_BUF_SIZE); + if (count <= 0) { + /* Connection is closed by peer */ + post_msg_to_module(conn, NULL, 0); + _conn_close(conn->handle); + } + else { + /* Data is received */ + post_msg_to_module(conn, io_buf, count); + } + } + else if (conn->type == CONN_TYPE_UDP) { + int count = udp_recv(conn->fd, io_buf, IO_BUF_SIZE); + if (count > 0) + post_msg_to_module(conn, io_buf, count); + } + else if (conn->type == CONN_TYPE_UART) { + int count = uart_recv(conn->fd, io_buf, IO_BUF_SIZE); + if (count > 0) + post_msg_to_module(conn, io_buf, count); + } + } + } + + return NULL; +} + +void +app_mgr_connection_event_callback(module_data *m_data, bh_message_t msg) +{ + uint32 argv[3]; + wasm_function_inst_t func_on_conn_data; + bh_assert(CONNECTION_EVENT_WASM == bh_message_type(msg)); + wasm_data *wasm_app_data = (wasm_data *)m_data->internal_data; + wasm_module_inst_t inst = wasm_app_data->wasm_module_inst; + connection_event_t *conn_event = + (connection_event_t *)bh_message_payload(msg); + int32 data_offset; + + if (conn_event == NULL) + return; + + func_on_conn_data = wasm_runtime_lookup_function( + inst, "_on_connection_data", "(i32i32i32)"); + if (!func_on_conn_data) + func_on_conn_data = wasm_runtime_lookup_function( + inst, "on_connection_data", "(i32i32i32)"); + if (!func_on_conn_data) { + printf("Cannot find function on_connection_data\n"); + return; + } + + /* 0 len means connection closed */ + if (conn_event->len == 0) { + argv[0] = conn_event->handle; + argv[1] = 0; + argv[2] = 0; + if (!wasm_runtime_call_wasm(wasm_app_data->exec_env, func_on_conn_data, + 3, argv)) { + const char *exception = wasm_runtime_get_exception(inst); + bh_assert(exception); + printf(":Got exception running wasm code: %s\n", exception); + wasm_runtime_clear_exception(inst); + return; + } + } + else { + data_offset = wasm_runtime_module_dup_data(inst, conn_event->data, + conn_event->len); + if (data_offset == 0) { + const char *exception = wasm_runtime_get_exception(inst); + if (exception) { + printf("Got exception running wasm code: %s\n", exception); + wasm_runtime_clear_exception(inst); + } + return; + } + + argv[0] = conn_event->handle; + argv[1] = (uint32)data_offset; + argv[2] = conn_event->len; + if (!wasm_runtime_call_wasm(wasm_app_data->exec_env, func_on_conn_data, + 3, argv)) { + const char *exception = wasm_runtime_get_exception(inst); + bh_assert(exception); + printf(":Got exception running wasm code: %s\n", exception); + wasm_runtime_clear_exception(inst); + wasm_runtime_module_free(inst, data_offset); + return; + } + wasm_runtime_module_free(inst, data_offset); + } +} + +bool +init_connection_framework() +{ + korp_tid tid; + + epollfd = epoll_create(MAX_EVENTS); + if (epollfd == -1) + return false; + + if (os_mutex_init(&g_lock) != 0) { + close(epollfd); + return false; + } + + if (!wasm_register_cleanup_callback(cleanup_connections)) { + goto fail; + } + + if (!wasm_register_msg_callback(CONNECTION_EVENT_WASM, + app_mgr_connection_event_callback)) { + goto fail; + } + + if (os_thread_create(&tid, polling_thread_routine, NULL, + BH_APPLET_PRESERVED_STACK_SIZE) + != 0) { + goto fail; + } + + return true; + +fail: + os_mutex_destroy(&g_lock); + close(epollfd); + return false; +} + +void +exit_connection_framework() +{ + polling_thread_run = false; +} diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/connection_mgr.cmake b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/connection_mgr.cmake new file mode 100644 index 000000000..c8f2b487e --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/linux/connection_mgr.cmake @@ -0,0 +1,13 @@ +# Copyright (C) 2019 Intel Corporation. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +set (WASM_LIB_CONN_MGR_DIR ${CMAKE_CURRENT_LIST_DIR}) + +include_directories(${WASM_LIB_CONN_MGR_DIR}) + + +file (GLOB_RECURSE source_all ${WASM_LIB_CONN_MGR_DIR}/*.c) + +set (WASM_LIB_CONN_MGR_SOURCE ${source_all}) + + diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/wasm_lib.cmake b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/wasm_lib.cmake new file mode 100644 index 000000000..58db0c1d8 --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/wasm_lib.cmake @@ -0,0 +1,18 @@ +# Copyright (C) 2019 Intel Corporation. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +set (WASM_LIB_CONN_DIR ${CMAKE_CURRENT_LIST_DIR}) + +include_directories(${WASM_LIB_CONN_DIR}) + +add_definitions (-DAPP_FRAMEWORK_CONNECTION) + + +include (${CMAKE_CURRENT_LIST_DIR}/${WAMR_BUILD_PLATFORM}/connection_mgr.cmake) + +file (GLOB source_all + ${WASM_LIB_CONN_MGR_SOURCE} + ${WASM_LIB_CONN_DIR}/*.c +) + +set (WASM_APP_LIB_CURRENT_SOURCE ${source_all}) diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/zephyr/connection_lib_impl.c b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/zephyr/connection_lib_impl.c new file mode 100644 index 000000000..a812a71a2 --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/zephyr/connection_lib_impl.c @@ -0,0 +1,25 @@ +/* + * Copyright (C) 2019 Intel Corporation. All rights reserved. + * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + */ + +/* + * Note: + * This file implements the linux version connection library which is + * defined in connection_lib.h. + * It also provides a reference impl of connections manager. + */ + +#include "connection_lib.h" + +/* clang-format off */ +/* + * Platform implementation of connection library + */ +connection_interface_t connection_impl = { + ._open = NULL, + ._close = NULL, + ._send = NULL, + ._config = NULL +}; +/* clang-format on */ diff --git a/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/zephyr/connection_mgr.cmake b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/zephyr/connection_mgr.cmake new file mode 100644 index 000000000..c8f2b487e --- /dev/null +++ b/fluent-bit/lib/wasm-micro-runtime-WAMR-1.2.2/core/app-framework/connection/native/zephyr/connection_mgr.cmake @@ -0,0 +1,13 @@ +# Copyright (C) 2019 Intel Corporation. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +set (WASM_LIB_CONN_MGR_DIR ${CMAKE_CURRENT_LIST_DIR}) + +include_directories(${WASM_LIB_CONN_MGR_DIR}) + + +file (GLOB_RECURSE source_all ${WASM_LIB_CONN_MGR_DIR}/*.c) + +set (WASM_LIB_CONN_MGR_SOURCE ${source_all}) + + |