BPI-CM9-Linux/aml-comp/vendor/amlogic/ccdatabase/server/dataserver_socketserver.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <sys/select.h>
#include <errno.h>
#include <sys/prctl.h>
#include <poll.h>
#include <sys/eventfd.h>

#ifdef BUILD_FOR_ANDROID
#include <netinet/tcp.h>
#endif

#include "dataserver_inner.h"

#include "list.h"

#ifdef LOG_LOGCAT

#define LOG_TAG "ccdataserver"
#include <log/log.h>
#define ds_inf(_fmt_,...) ALOGI("%s:%d "_fmt_, __func__, __LINE__, ##__VA_ARGS__)
#define ds_err(_fmt_,...) ALOGE("%s:%d "_fmt_, __func__, __LINE__, ##__VA_ARGS__)

#else

#define ds_inf(_fmt_,...) printf("I %s:%d "_fmt_"\n", __func__, __LINE__, ##__VA_ARGS__)
#define ds_err(_fmt_,...) printf("E %s:%d "_fmt_"\n", __func__, __LINE__, ##__VA_ARGS__)

#endif


/*The global ipc server*/
static ds_server_t server;


static int do_read(ds_connection_t *conn, void *buf, size_t count)
{
    int fd = conn->fd;
    int n = read(fd, buf, count);
    if (n < 0) {
        ds_err("fd[%d] read error, err(%d:%s)", fd, errno, strerror(errno));
        return -1;
    } else if (n == 0) {
        ds_inf("fd[%d] EOF", fd);
        return -1;
    }
    return n;
}

static int do_write(ds_connection_t *conn, void *buf, size_t count)
{
    int fd = conn->fd;
    int sent = 0;
    while (sent < count) {
        int n = write(fd, (uint8_t *)buf + sent, count - sent);
        if (n <= 0)
            return -1;
        sent += n;
    }
    return sent;
}

static void *connection_read_task(void *args)
{
    ds_connection_t *conn = (ds_connection_t *)args;

    {
        char name[32];
        snprintf(name, 32, "DAConnR-%d", conn->fd);
        prctl(PR_SET_NAME, name);
    }

    ds_inf("connection fd[%d] session(read) started", conn->fd);

    int n;

    while (conn->active) {
        fd_set rfds;
        struct timeval timeout = {1, 0};
        int maxfd;
        int ret;

        FD_ZERO(&rfds);
        FD_SET(conn->fd, &rfds);
        maxfd = conn->fd + 1;

        ret = select(maxfd, &rfds, NULL, NULL, &timeout);
        if (ret < 0) /*error*/
            break;
        else if (ret == 0) /*timeout*/
            continue;

        if (!conn->active)
            break;

        if (FD_ISSET(conn->fd, &rfds)) {//data_in proc

            ret = ds_connection_manage_data_in(conn);
            if (ret == -3)
                break;
            if (ret == -1)
                continue;
        }
    }

end:
    /*cleanup the connection, [connection death state]*/
    ds_connection_cleanup(conn);

    ds_inf("connection to [%s] fd[%d] session(read) finished", conn->client, conn->fd);

    return NULL;
}

static void *connection_write_task(void *args)
{
    ds_connection_t *conn = (ds_connection_t *)args;

    {
        char name[32];
        snprintf(name, 32, "DAConnW-%d", conn->fd);
        prctl(PR_SET_NAME, name);
    }

    ds_inf("connection fd[%d] session(write) started", conn->fd);

    int n;

    while (conn->active) {
        fd_set wfds;
        struct timeval timeout = {1, 0};
        int maxfd;
        int ret;

        FD_ZERO(&wfds);
        FD_SET(conn->fd, &wfds);
        maxfd = conn->fd + 1;

        ret = select(maxfd, NULL, &wfds, NULL, &timeout);
        if (ret < 0) /*error*/
            break;
        else if (ret == 0) /*timeout*/
            continue;

        if (!conn->active)
            break;

        if (FD_ISSET(conn->fd, &wfds)) {//data_out proc

            ret = ds_connection_manage_data_out(conn);
            if (ret == -3)
                break;
            if (ret == -1)
                continue;
        }
    }

end:
    /*cleanup the connection, [connection death state]*/
    ds_connection_cleanup(conn);

    ds_inf("connection to [%s] fd[%d] session(write) finished", conn->client, conn->fd);

    return NULL;
}

static int init_connection(ds_connection_t *conn, void *priv)
{
    conn->fd = (int)priv;
    return 0;
}

static void deinit_connection(ds_connection_t *conn)
{
    close(conn->fd);
}

static int start_connection(ds_connection_t *conn)
{
#ifdef BUILD_FOR_ANDROID
    {
        int val = 1;
        errno = 0;
        if (setsockopt(conn->fd, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(val))) {
            ds_err("setsockopt SO_KEEPALIVE failed. err(%d/%s)", errno, strerror(errno));
        }
        int keepidle = 20;
        int keepinterval = 3;
        setsockopt(conn->fd, SOL_TCP, TCP_KEEPIDLE, &keepidle, sizeof(keepidle));
        setsockopt(conn->fd, SOL_TCP, TCP_KEEPINTVL, &keepinterval, sizeof(keepinterval));

    }
#endif

    int ret = pthread_create(&conn->wr_thread, NULL, connection_write_task, conn);
    if (ret == 0)
        ret = pthread_create(&conn->rd_thread, NULL, connection_read_task, conn);

    return ret;
}

static void stop_connection(ds_connection_t *conn)
{
    ds_inf("connection closing... fd[%d] dev[%s] cli[%s]",
            conn->fd, conn->device, conn->client);

    conn->active = 0;

    //todo: fixme
    //close(conn->fd);
    ds_rb_set_flag(&conn->rb, DS_RB_FLAG_EXIT);

    pthread_join(conn->wr_thread, NULL);
    pthread_join(conn->rd_thread, NULL);

    ds_inf("connection closed, fd[%d] dev[%s] cli[%s]",
            conn->fd, conn->device, conn->client);
}

static void *server_task(void *args)
{
    ds_server_t *server = (ds_server_t *)args;

    {
        char name[32];
        snprintf(name, 32, "DAServer-%d", server->fd);
        prctl(PR_SET_NAME, name);
    }

    ds_inf("daserver fd[%d] accepting connections....",
            server->fd);

    struct sockaddr_un c_un;
    socklen_t c_len = sizeof(c_un);
    int connfd;

    while (server->running) {
        int ret;
        struct pollfd fds[2];

        memset(fds, 0, sizeof(fds));

        fds[0].fd = server->fd;
        fds[1].fd = server->evt_fd;
        fds[0].events = fds[1].events = POLLIN | POLLERR;

        ret = poll(fds, 2, 1000);
        if (ret < 0)
            break;
        else if (ret == 0)
            goto next;

        if (!server->running)
            break;

        if (!(fds[0].revents & POLLIN))
            goto next;

        if ((connfd = accept(server->fd, (struct sockaddr *)&c_un, &c_len)) < 0) {
            ds_err("accept error, err(%d:%s)", errno, strerror(errno));
            goto next;
        }

        ds_inf("connection fd[%d], start session... ",
                connfd);

        ds_connection_t *conn = ds_connection_create();

        if (conn) {

            conn->init = init_connection;
            conn->deinit = deinit_connection;
            conn->start = start_connection;
            conn->stop = stop_connection;
            conn->read = do_read;
            conn->write = do_write;

            ret = ds_connection_start(server, conn, (void *)(long)connfd);
            if (ret != 0) {
                ds_err("connection fd[%d], session start fail", connfd);
            }

        } else {
            ds_err("memory fail");
        }

next:
        ds_server_cleanup_connections(server, 0);
    }

    ds_server_cleanup_connections(server, 1);

    server->running = 0;

    ds_inf("daserver fd[%d] exit", server->fd);

    return NULL;
}

static int init_server(ds_server_t *server, void *arg1, void *arg2)
{
    server->fd = (int)arg1;
    server->evt_fd = (int)arg2;
    return 0;
}

static void deinit_server(ds_server_t *server)
{
    if (server->fd != -1) {
        close(server->fd);
        server->fd = -1;
    }
    if (server->evt_fd != -1) {
        close(server->evt_fd);
        server->evt_fd = -1;
    }
}

static void stop_server(ds_server_t *server)
{
    {
        int64_t pad = 1;
        write(server->evt_fd, &pad, sizeof(pad));
    }
}

int socket_server_start_unix(const char *socket_path)
{
    int fd;
    int evt_fd;

    if ((fd = socket(AF_UNIX, SOCK_STREAM, 0)) < 0) {
        ds_err("open socket fail");
        return -1;
    }

    struct sockaddr_un s_un;
    int size;
    memset(&s_un, 0, sizeof(s_un));
    s_un.sun_family = AF_UNIX;
    strncpy(s_un.sun_path, socket_path, sizeof(s_un.sun_path) - 1);
    size = offsetof(struct sockaddr_un, sun_path) + strlen(s_un.sun_path);
    unlink(socket_path);

    if (bind(fd, (struct sockaddr *)&s_un, size) < 0) {
        ds_err("bind error, err(%d:%s)", errno, strerror(errno));
        close(fd);
        return -1;
    }

    ds_inf("socket ready.");

    if (listen(fd, 20) < 0) {
        ds_err("listen error, err(%d:%s)", errno, strerror(errno));
        close(fd);
        return -1;
    }

    evt_fd = eventfd(0, 0);

    server.init = init_server;
    server.deinit = deinit_server;
    server.stop = stop_server;

    int ret = ds_server_start(&server, (void*)(long)fd, (void*)(long)evt_fd, server_task);

    if (ret != 0) {
        close(evt_fd);
        close(fd);
    }

    return ret;
}

int socket_server_stop_unix(void)
{
    ds_inf("unix server closing...");
    ds_server_stop(&server);
    ds_inf("unix server closed.");

    return 0;
}

#ifdef DS_SUPPORT_IP

#include <netinet/in.h>

static ds_server_t ip_server;

int socket_server_start_ip(int socket_port)
{
    /*init the server inst*/
    INIT_LIST_HEAD(&ip_server.conns);

    ip_server.fd = -1;
    ip_server.running = 0;

    if ((ip_server.fd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
        ds_err("open socket fail");
        return -1;
    }

    int opt = 1;
    if (setsockopt(ip_server.fd, SOL_SOCKET,
                SO_REUSEADDR | SO_REUSEPORT,
                &opt, sizeof(opt))) {
        ds_err("setsockopt fail");
        close(ip_server.fd);
        return -1;
    }

    struct sockaddr_in s_in;
    memset(&s_in, 0, sizeof(s_in));
    s_in.sin_family = AF_INET;
    s_in.sin_addr.s_addr = INADDR_ANY;
    s_in.sin_port = htons(socket_port);

    if (bind(ip_server.fd, (struct sockaddr *)&s_in, sizeof(s_in)) < 0) {
        ds_err("bind fail");
        close(ip_server.fd);
        return -1;
    }

    ds_inf("socket ready.");

    if (listen(ip_server.fd, 20) < 0) {
        ds_err("listen fail");
        close(ip_server.fd);
        return -1;
    }

    ip_server.running = 1;
    int ret = pthread_create(&ip_server.thread, NULL, server_task, &ip_server);

    if (ret != 0)
        close(ip_server.fd);

    return ret;
}

int socket_server_stop_ip(void)
{
    ip_server.running = 0;
    close(ip_server.fd);

    ds_inf("ip server closing...");

    pthread_join(ip_server.thread, NULL);

    ds_inf("ip server closed.");

    return 0;
}

#endif