yb_arm/osp/src/ospUdp.c

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <errno.h>
#include <unistd.h>
#include <fcntl.h>

#include "osp.h"
#include "ucp_printf.h"


typedef struct tag_OSP_UDP_SOCKET                                                                                       
{   
    lx_sem_t mutex;
    char ip[64];
    int32_t port;
    int32_t taskid;
    int32_t direction;/**/
    int32_t pkgnum;
    int64_t totalsize;
    int32_t errcnt;
    struct sockaddr_in servAddr;
    int32_t socket;
    int32_t valid;    
}OSP_UDP_SOCKET;



#define OSP_MAX_UDP_CONNECT_NUM 256

OSP_UDP_SOCKET osp_udp[OSP_MAX_UDP_CONNECT_NUM] = {0};

#define OSPUDP_MSG_LEN (1500 + MSG_HEAD_SIZE)



void *udp_sem;

int32_t osp_get_udp(char *ip, int32_t port, int32_t direction)
{
    int32_t i;
    
    for (i = 0; i < OSP_MAX_UDP_CONNECT_NUM; i++)
    {
        if ((osp_udp[i].valid) && (strcmp(osp_udp[i].ip, ip) == 0) && (osp_udp[i].port == port) && (osp_udp[i].direction == direction))
        {
            return i;
        }
    }

    return -1;
}

int32_t osp_get_rxudp(char *ip, int32_t port)
{
    return osp_get_udp(ip, port, OSP_UDP_RX);
}

int32_t osp_get_txudp(char *ip, int32_t port)
{
    return osp_get_udp(ip, port, OSP_UDP_TX);
}


int32_t osp_get_ip_port(int32_t udpid, char *ip, int32_t *port)
{
    if ((udpid < 0) || (udpid > OSP_MAX_UDP_CONNECT_NUM) || NULL == ip || NULL == port)
    {
        return -1;
    }
    strcpy(ip, osp_udp[udpid].ip);
    *port = osp_udp[udpid].port;

    return -1;
}




int32_t osp_check_udpid(int32_t udpid)
{
    if ((udpid < 0) || (udpid > OSP_MAX_UDP_CONNECT_NUM) || !(osp_udp[udpid].valid))
    {
        return 0;
    }

    return 1;
}
int32_t osp_set_udp_noblock(int32_t udpid)
{
    if (osp_check_udpid(udpid))
        return -1;
    return fcntl(osp_udp[udpid].socket, F_SETFL, O_NONBLOCK);
}




int32_t osp_add_udp(char *ip, int32_t port, int32_t direction)
{
    int32_t i;
    osp_sem_take(udp_sem, -1);

    for (i = 0; i < OSP_MAX_UDP_CONNECT_NUM; i++)
    {
        if (osp_udp[i].valid == 0)
        {
            strcpy(osp_udp[i].ip, ip);
            osp_udp[i].port = port;
            osp_udp[i].valid = 1;
            osp_udp[i].direction = direction;
            osp_sem_give(udp_sem);
            return i;
        }
    }
    osp_sem_give(udp_sem);
    return -1;

}





int32_t osp_del_udp(int32_t udpid)
{
    if (!osp_check_udpid(udpid))
    {
        return 0;
    }
    
    osp_udp[udpid].valid = 0;
    close(osp_udp[udpid].socket);

    return 0;


}


int32_t osp_create_rxucp(int32_t port)
{
    int32_t res;
    int32_t sockset;
    int32_t tmp = 1;
    int32_t addrlen;
    int32_t udp_id;
    int32_t idx;
    
    if (osp_get_rxudp("127.0.0.1", port) >= 0)
    {
        return -1;
    }


    
    idx = osp_add_udp("127.0.0.1", port, OSP_UDP_RX);
    if (idx < 0)
    {
        return -1;
    }    

    osp_udp[idx].taskid = CURRENT_TASKID;
    osp_udp[idx].direction = OSP_UDP_RX;
    
    sockset = socket(AF_INET, SOCK_DGRAM, 0);
    if (sockset == -1) {
        perror("socket");
        return -1;
    }    

    setsockopt(sockset, SOL_SOCKET, SO_REUSEADDR, &tmp, sizeof(int32_t));
    
    osp_udp[idx].socket = sockset;
    osp_udp[idx].servAddr.sin_family = AF_INET;
    osp_udp[idx].servAddr.sin_port = htons(port);
    osp_udp[idx].servAddr.sin_addr.s_addr = htonl(INADDR_ANY);

    addrlen = sizeof(struct sockaddr);
    res = bind(sockset, (struct sockaddr *)&osp_udp[idx].servAddr, addrlen);
    if (res == -1) {
        perror("bind");
        return -1;
    }

    udp_id = idx;
    return udp_id;
}

int32_t osp_create_txucp(int32_t port, char *targetIp)
{
    int32_t sockset;
    int32_t udp_id;
    int32_t idx;

    if (osp_get_txudp(targetIp, port) >= 0)
    {
        return -1;
    }
    idx = osp_add_udp(targetIp, port, OSP_UDP_TX);
    if (idx < 0)
    {
        return -1;
    }

    osp_udp[idx].taskid = CURRENT_TASKID;
    osp_udp[idx].direction = OSP_UDP_TX;
    
    sockset = socket(AF_INET, SOCK_DGRAM, 0);
    if (sockset == -1) {
        perror("socket");
        return -1;
    }
    

    
    osp_udp[idx].socket = sockset;
    osp_udp[idx].servAddr.sin_family = AF_INET;
    osp_udp[idx].servAddr.sin_addr.s_addr = inet_addr(targetIp);
    osp_udp[idx].servAddr.sin_port = htons(port);

    udp_id = idx;
    
    return udp_id;
}


int32_t osp_udp_receive(char *pMsg, int32_t len, int32_t udp_id)
{
//    struct sockaddr_in sin;
    socklen_t sin_len;

    sin_len = sizeof(struct sockaddr); 
    len = recvfrom(osp_udp[udp_id].socket, pMsg, len, 0, (struct sockaddr *)&osp_udp[udp_id].servAddr, &sin_len);
    if (len <= 0)
    {
        osp_atomic_inc(&osp_udp[udp_id].errcnt);
        return len;
    }
    
    osp_atomic_inc(&osp_udp[udp_id].pkgnum);
    osp_sem_take(&osp_udp[udp_id].mutex, -1);
    osp_udp[udp_id].totalsize += len;
    osp_sem_give(&osp_udp[udp_id].mutex);

    return len;
    
}

int32_t osp_udp_send(char *pMsg, int32_t msgLen, int32_t udp_id)
{
    int32_t Len;

    if ((pMsg == NULL) || !osp_check_udpid(udp_id))
    {
        return (-1);
    }

    Len = sendto(osp_udp[udp_id].socket, pMsg, msgLen, 0, (struct sockaddr *)&osp_udp[udp_id].servAddr, sizeof(struct sockaddr_in));
    if (Len != msgLen)
    {
        //osp_atomic_inc(&osp_udp[udp_id].errcnt);
        return (-1);
    }

#if 0
    osp_atomic_inc(&osp_udp[udp_id].pkgnum);

    osp_sem_take(&osp_udp[udp_id].mutex, -1);
    osp_udp[udp_id].totalsize += Len;
    osp_sem_give(&osp_udp[udp_id].mutex);
#endif
    return (0);

}

int32_t osp_send_udpmsg(Osp_Msg_Head *pMsg, uint64_t udp_id)
{
    int32_t ret;
    ret = osp_udp_send((char*)pMsg, pMsg->MsgSize + MSG_HEAD_SIZE, OSP_UDPID_TO_TX(udp_id));

    osp_free_msg(pMsg);
    return ret;
}

Osp_Msg_Head *osp_rev_udpmsg(uint64_t udp_id)
{
    char *pbuf;
    int32_t ret;
    
    pbuf = (char *) osp_alloc_msg(OSPUDP_MSG_LEN);
    if (NULL == pbuf)
    {
        return NULL;
    }
    
    ret = osp_udp_receive(pbuf, OSPUDP_MSG_LEN, OSP_UDPID_TO_RX(udp_id));
    if (ret <= 0)
    {
        return NULL;
    }

    return (Osp_Msg_Head *)pbuf;
    
}



int32_t osp_udp_init(void)
{
    int32_t i;
    
    udp_sem = osp_semm_create();
    
    for (i = 0; i < OSP_MAX_UDP_CONNECT_NUM; i++)
    {
        osp_semsm_create(&(osp_udp[i].mutex));
    }
    return 0;
}





void osp_show_udp(void)
{
    int32_t i;
    char buf[4096];
    int32_t len = 0;
    
    len += sprintf(buf + len, "\n%-16s%-16s%-16s%-16s%-16s%-16s%-16s%-16s%-16s\n", "direction", "udpid", "taskid", "socket", "ip", "port", "pkgnum", "totalsize", "errcnt");
    for (i = 0; i < OSP_MAX_UDP_CONNECT_NUM; i++)
    {
        if ((osp_udp[i].valid) && (osp_udp[i].direction == OSP_UDP_RX))
        {
            len += sprintf(buf + len, "%-16s%-16d%-16d%-16d%-16s%-16d%-16d%-16lu%-16d\n", 
                "rx", i, osp_udp[i].taskid, osp_udp[i].socket, osp_udp[i].ip, osp_udp[i].port, osp_udp[i].pkgnum, osp_udp[i].totalsize, osp_udp[i].errcnt);
        }
    }

    len += sprintf(buf + len, "\n");
    
    for (i = 0; i < OSP_MAX_UDP_CONNECT_NUM; i++)
    {
        if ((osp_udp[i].valid) && (osp_udp[i].direction == OSP_UDP_TX))
        {
            len += sprintf(buf + len, "%-16s%-16d%-16d%-16d%-16s%-16d%-16d%-16lu%-16d\n", 
                "tx", i, osp_udp[i].taskid, osp_udp[i].socket, osp_udp[i].ip, osp_udp[i].port, osp_udp[i].pkgnum,  osp_udp[i].totalsize, osp_udp[i].errcnt);

            
        }
    }
    
    osp_net_shell_out(buf, len);

    UCP_PRINT_SHELL("%s\n", buf);
}