src/lynq/lib/liblynq-network/lynq_module_socket.cpp

#include <stdio.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <string.h>
#include <unistd.h>
#include <binder/Parcel.h>
#include <log/log.h>
#include <cutils/jstring.h>
#include <pthread.h>
#include <list>
#include <vendor-ril/telephony/ril.h>
#include <vendor-ril/telephony/mtk_ril_sp.h>
#include "lynq_network.h"
#include "lynq_module_common.h"
#include "lynq_module_socket.h"
#include "liblog/lynq_deflog.h"

#define LYNQ_SERVICE_PORT 8088
#define LYNQ_ADDRESS "127.0.0.1"
#define LYNQ_URC_SERVICE_PORT 8086
#define LYNQ_URC_ADDRESS "0.0.0.0"
#define LYNQ_REQUEST_PARAM_BUF 8192
#define LYNQ_REC_BUF 8192

#define USER_LOG_TAG "LYNQ_NETWORK"

typedef struct{
    int uToken;
    int request;
    int paramLen;
    char param[LYNQ_REQUEST_PARAM_BUF];
}lynq_client_t;

typedef struct{
    int resp_type;
    int token;
    int request;
    int slot_id;
    int error;
}lynq_resp_t;

lynq_client_t client_t;
lynq_resp_t response;

int module_len_rc_addr_serv;
int module_len_urc_addr_serv;
struct sockaddr_in module_rc_addr_serv;
struct sockaddr_in module_urc_addr_serv;
static int module_rc_sock_fd = -1;
static int module_urc_sock_fd = -1;
int module_urc_status = 1;
int module_rc_status = 1;
pthread_t module_urc_tid = -1;
pthread_t module_rc_tid = -1;
static pthread_mutex_t s_urc_mutex = PTHREAD_MUTEX_INITIALIZER;

/*hq add for set waiting time 2022/09/13 begin*/
static pthread_mutex_t s_sendto_mutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_mutex_t s_RecvMsgBlockMutex = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t s_WaitRecvMsgCond = PTHREAD_COND_INITIALIZER;
#define BLOCK_RECV_MSG_LOCK() pthread_mutex_lock(&s_RecvMsgBlockMutex)
#define BLOCK_RECV_MSG_UNLOCK() pthread_mutex_unlock(&s_RecvMsgBlockMutex)
#define BLOCK_WAIT_RECV_MSG_TIME_OUT(a) pthread_cond_timedwait(&s_WaitRecvMsgCond, &s_RecvMsgBlockMutex,(a))
#define BLOCK_WAKEUP_RECV_MSG() pthread_cond_broadcast(&s_WaitRecvMsgCond)
static std::list<Parcel*> s_recv_parcel_list;
const int waitResponse(int token,int time_out);
/*hq add for set waiting time 2022/09/13 end*/

int g_module_Global_uToken = 0;
int g_wait_time=5;

/*inner test*/
static int s_inner_test=0;

/**@brief just for inner test
* @param test_mode [IN]: test mode
*  0:success
*  other:failure
*/
void lynq_set_test_mode(const int test_mode)
{
    if(test_mode<0)
    {
        g_wait_time = -test_mode;          
    }
    else if(test_mode==9999)
    {
        LYERRLOG("%s inner test para %d %d",__func__,s_inner_test,g_wait_time);         
    }    
    else 
    {
        s_inner_test = test_mode;    
    }
}

void cleanup_RecvMsgBlock_mutex(void *arg)
{
    BLOCK_RECV_MSG_UNLOCK(); 
}

/**@brief wait response with expected token and write msg to parcel in some time
* @param p [IN]: pointer the parcel
* @param token [IN]: the expected token for the response msg
* @param time_out [IN]: timeout.
* @return
*  0:success
*  other:failure
*/
const int waitResponse(Parcel*& p, int token,int time_out)
{
    int waitToken = token;
    int wakeup_token=-1;
    int resp_type;
    struct timeval now;
    struct timespec timeout;  
    int ret;
    std::list<Parcel*>::iterator iter;
    int cnt=0;
    
	gettimeofday(&now,NULL);
    timeout.tv_sec = now.tv_sec+time_out; //timeout is 1min
    timeout.tv_nsec = now.tv_usec*1000;

    LYINFLOG("%s wait token is %d, wait time is %d",__func__,waitToken,time_out);
    
    BLOCK_RECV_MSG_LOCK();
    pthread_cleanup_push(cleanup_RecvMsgBlock_mutex, NULL); // thread cleanup handler
    while(module_rc_status) { 
        cnt++;
        for(iter=s_recv_parcel_list.begin();iter!=s_recv_parcel_list.end();++iter)
        {
            (*iter)->setDataPosition(0);
            (*iter)->readInt32(&resp_type);
            (*iter)->readInt32(&wakeup_token);
            if(wakeup_token==waitToken)
            {
                LYINFLOG("%s get waitToken",__func__);
                p=(*iter);
                p->setDataPosition(0);
                s_recv_parcel_list.erase(iter);
                goto waitResponse_end;
            }
        }
        LYINFLOG("%s no wait Token in msg list, list size is %d",__func__,s_recv_parcel_list.size());
        ret=BLOCK_WAIT_RECV_MSG_TIME_OUT(&timeout);
        if(ret!=0)
        {
            LYERRLOG("%s no expected token %d after %d second",__func__,waitToken,time_out);
            break;
        }        
    }
waitResponse_end:
    LYINFLOG("%s wait token is %d wakeup_token is %d, cnt is %d",__func__,waitToken,wakeup_token,cnt);
    pthread_cleanup_pop(0);
    BLOCK_RECV_MSG_UNLOCK(); 
    return wakeup_token==waitToken ? 0:LYNQ_E_TIME_OUT;
}

/**@brief print solicied response msg's head information
* @param head [IN]: head information
* @return none
*/
void PrintHeader(lynq_resp_t& head)
{
    LYINFLOG("resp_type=%d,token=%d,request=%d,slot_id=%d,error_code=%d",head.resp_type,head.token,head.request,head.slot_id,head.error);
}

/**@brief get solicied response msg's head
* @param head [OUT]: head information
* @return
*  0:success
*  other:failure
*/
int GetHeader(Parcel* &p, lynq_resp_t& head)
{
    LYINFLOG("get header");
    if(p->dataAvail() > 0)
    {
        p->readInt32(&(head.resp_type));
        p->readInt32(&(head.token));
        p->readInt32(&(head.request));
        p->readInt32(&(head.slot_id));
        p->readInt32(&(head.error));
        PrintHeader(head);
        return RESULT_OK;  
    }
    else
    {
        return RESULT_ERROR;  
    }
}

/**@brief send msg to service and get response from service
* @param request_id [IN]: request id
* @param time_out [IN]: wait time uplimit
* @param p [IN]: point to response msg's parcel
* @param argc [IN]: how many parameters in parameter string
* @param format [IN]: parameter string's format
* @param ... [IN]: the specific parameter
* @return
*  0:success
*  other:failure
*/
int lynq_send_common_request(Parcel*& p, int time_out, int request_id, int argc, const char* format,...)
{
    lynq_client_t client;
    int ret;
    int send_num;
    
    client.uToken = g_module_Global_uToken;
    g_module_Global_uToken=(g_module_Global_uToken+1)%10000;/*0-10000*/
    client.request = request_id;
    client.paramLen = argc;
    bzero(client.param,LYNQ_REQUEST_PARAM_BUF);
    if(argc!=0)
    {
        va_list args;
        va_start(args, format);
        vsnprintf(client.param, LYNQ_REQUEST_PARAM_BUF, format, args);
        va_end(args);    
    }
    LYINFLOG("uToken=%d,request=%d,paralen=%d,param=%s",client.uToken,client.request,client.paramLen,client.param);
    pthread_mutex_lock(&s_sendto_mutex);
     if(s_inner_test==1)
    {
        send_num = 1;          
    }
    else
    {
        send_num = sendto(module_rc_sock_fd,&client,sizeof(client),0,(struct sockaddr *)&module_rc_addr_serv,module_len_rc_addr_serv); 
    }
    if(s_inner_test>=1000)
    {
        time_out = s_inner_test-1000;
    }        
    pthread_mutex_unlock(&s_sendto_mutex);

    if(send_num <= 0)       
    {
        LYERRLOG("send request fail, send num is %d", send_num);
        return LYNQ_E_SEND_REQUEST_FAIL;
    }
    ret = waitResponse(p, client.uToken,time_out);

    if(ret==0)
    {
        lynq_resp_t head; 
        ret=GetHeader(p,head);
        if(ret!=0)
        {
            LYERRLOG("%s %d get head error %d",__func__,client.uToken,ret);
            delete p;
            return LYNQ_E_GET_HEAD_ERROR;
        }
        if(head.error!=0)
        {
            LYERRLOG("%s %d mdm return head error %d",__func__,client.uToken,head.error);
            delete p;
            return head.error;
        }
        LYERRLOG("%s %d suc",__func__,client.uToken);
        return RESULT_OK;  
    }

    LYERRLOG("%s %d fail, ret is %d",__func__,client.uToken,ret);
   
    return ret;    
}

/**@ a thread just for recv\buffer solicited msg's response and notice waiting thread
* @param p [IN]: no meaning
* @return
* always null
*/
void *thread_rc_recv(void *p)
{
    Parcel* rc_p;
    std::list<Parcel*>::iterator iter;
    int resp_type = -1;   
    char rc_data[LYNQ_REC_BUF];
    int rc_len;
    int null_cnt=0;
    int wakeup_token; 
 
    LYINFLOG("rc thread is running");
    while(module_rc_status)
    {
        bzero(rc_data,LYNQ_REC_BUF);
        while(true)
        {
            rc_len = recvfrom(module_rc_sock_fd,rc_data,sizeof(rc_data),0,(struct sockaddr *)&module_rc_addr_serv,(socklen_t *)&module_len_rc_addr_serv);
            if(rc_len<sizeof(int32_t)*2)
            {
                 LYERRLOG("%s recv len %d less %d",__func__, rc_len,sizeof(int)*2);
                 continue;
            } 
            rc_p= new Parcel;
            if(rc_p==NULL)
            {
                null_cnt++;
                LYERRLOG("%s rc_p is NULL, cnt is %d",__func__, null_cnt);
                if(null_cnt>20)
                {
                    goto rc_recv_end;
                }
                continue;
            }
            else 
            {
                null_cnt=0;
            }
                
            rc_p->setData((uint8_t *)rc_data,rc_len); // p.setData((uint8_t *) buffer, buflen);
            rc_p->setDataPosition(0);
            if(rc_p->dataAvail()>0)
            {
                rc_p->readInt32(&resp_type); 
                rc_p->readInt32(&wakeup_token); 
                BLOCK_RECV_MSG_LOCK();
                s_recv_parcel_list.push_back(rc_p);                
                LYINFLOG("%s wakeup token is %d, list size is %d!",__func__,wakeup_token,s_recv_parcel_list.size());
                if(s_recv_parcel_list.size()>20) //max 20
                {
                    iter=s_recv_parcel_list.begin();
                    (*iter)->setDataPosition(0);
                    (*iter)->readInt32(&resp_type); 
                    (*iter)->readInt32(&wakeup_token); 
                    delete (*(s_recv_parcel_list.begin()));
                    LYERRLOG("%s wakeup token %d is deleted!",__func__,wakeup_token);
                    s_recv_parcel_list.erase(iter);
                }                   
                BLOCK_WAKEUP_RECV_MSG();
                BLOCK_RECV_MSG_UNLOCK();
                break;
            }
            else
            {
                LYERRLOG("%s rc_p data Avail %d not greater than 0",__func__, rc_p->dataAvail());
                delete rc_p;
            }
        }
    }                 

rc_recv_end:    
    LYINFLOG("rc thread ended");
    return NULL;
}

void *thread_urc_recv(void *p)
{
    Parcel *urc_p =NULL;
    char urc_data[LYNQ_REC_BUF];    
    int res = 0;
   
    LYINFLOG("urc thread is running");
    while(module_urc_status)
    {
        bzero(urc_data,LYNQ_REC_BUF);
        res = recvfrom(module_urc_sock_fd,urc_data,sizeof(urc_data),0,(struct sockaddr *)&module_urc_addr_serv,(socklen_t*)&module_len_urc_addr_serv);
        if(res<=0)
        {
            LYERRLOG("thread_urc_recv step2 fail:");
            break;
        }
        urc_p = new Parcel();
        if(urc_p == NULL)
        {
            LYERRLOG("new parcel failure!!!");
            break;
        }
        urc_p->setData((uint8_t *)urc_data,res); // p.setData((uint8_t *) buffer, buflen);
        urc_p->setDataPosition(0);
        if(urc_p->dataAvail()>0)
        {
            pthread_mutex_lock(&s_urc_mutex);
            urc_msg_process(urc_p);            
            pthread_mutex_unlock(&s_urc_mutex);
        }
        delete urc_p;
        urc_p = NULL;
    }
    LYINFLOG("urc thread ended");
    return NULL;
}

int lynq_server_socket_start()
{
    module_rc_sock_fd = socket(AF_INET, SOCK_DGRAM, 0);
    if(module_rc_sock_fd < 0)
    {
        LYERRLOG("socket open error");
        return -1;
    }

    LYINFLOG("module_rc_sock_fd = %d",module_rc_sock_fd);

    memset(&module_rc_addr_serv, 0, sizeof(module_rc_addr_serv));
    module_rc_addr_serv.sin_family = AF_INET;
    module_rc_addr_serv.sin_addr.s_addr = inet_addr(LYNQ_ADDRESS);
    module_rc_addr_serv.sin_port = htons(LYNQ_SERVICE_PORT);
    module_len_rc_addr_serv = sizeof(module_rc_addr_serv);

    BLOCK_RECV_MSG_LOCK(); 
    std::list<Parcel*>::iterator iter;
    for(iter=s_recv_parcel_list.begin();iter!=s_recv_parcel_list.end();++iter)
    {
        delete (*iter);        
    }
    s_recv_parcel_list.clear();  
    BLOCK_RECV_MSG_UNLOCK(); 

//    pthread_attr_t attr;
    int ret;

    pthread_mutex_init(&s_sendto_mutex, NULL);   
    pthread_mutex_init(&s_RecvMsgBlockMutex, NULL);   

    module_rc_status = 1;
//    pthread_attr_init(&attr);
//    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    ret = pthread_create(&module_rc_tid,/*&attr*/NULL,thread_rc_recv,NULL);
    if(ret <0)
    {
        LYERRLOG("rc pthread create error");
        module_rc_status =0;
        close(module_rc_sock_fd);
        module_rc_sock_fd =-1;        
        return ret;
    }   
    
    LYINFLOG("rc start success");

    return RESULT_OK;  
}

int lynq_urc_socket_start()
{
//    pthread_t tid;
//    pthread_attr_t attr;
    int on = 1;
    int ret = 0;
    module_len_urc_addr_serv = sizeof(sockaddr_in);
    module_urc_sock_fd = socket(AF_INET, SOCK_DGRAM, 0);
    if (module_urc_sock_fd <0){
        LYERRLOG("urc socket error");
        return RESULT_ERROR;  
    }
    module_urc_addr_serv.sin_family = AF_INET;
    module_urc_addr_serv.sin_port = htons(LYNQ_URC_SERVICE_PORT);
    module_urc_addr_serv.sin_addr.s_addr = inet_addr(LYNQ_URC_ADDRESS);
    /* Set socket to allow reuse of address and port, SO_REUSEADDR value is 2*/
    ret = setsockopt(module_urc_sock_fd,SOL_SOCKET,SO_REUSEADDR,&on,sizeof on);
    if(ret <0)
    {
        LYERRLOG("urc socket set error");
        close(module_urc_sock_fd);
        module_urc_sock_fd =-1;
        return RESULT_ERROR;  
    }
    ret = bind(module_urc_sock_fd ,(struct sockaddr*)&module_urc_addr_serv, sizeof(module_urc_addr_serv));
    if(ret <0)
    {
        LYERRLOG("urc socket bind error");
        close(module_urc_sock_fd);
        module_urc_sock_fd =-1;
        return RESULT_ERROR;  
    }

    module_urc_status = 1;
  //  pthread_attr_init(&attr);
  //  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    ret = pthread_create(&module_urc_tid,/*&attr*/NULL,thread_urc_recv,NULL);
    if(ret <0)
    {
        LYERRLOG("urc pthread create error");
        module_urc_status = 0;
        close(module_urc_sock_fd);
        module_urc_sock_fd =-1;
        return RESULT_ERROR;  
    }
    LYINFLOG("urc start success");
    
    return RESULT_OK;  
}

void lynq_close_urc_thread()
{
    int ret;

    pthread_mutex_lock(&s_urc_mutex); //just cancel urc tid when recvfrom avoid mutual lock for tid may call pthread_cond_wait
    module_urc_status = 0;    
    if(module_urc_tid!=-1)
    {
        ret = pthread_cancel(module_urc_tid);
        LYINFLOG("pthread cancel urc ret = %d",ret);            
    }
    pthread_mutex_unlock(&s_urc_mutex);
     if(module_urc_tid != -1)
    {
        ret = pthread_join(module_urc_tid,NULL);
        LYINFLOG("pthread join urc tid ret = %d",ret);
        module_urc_tid =-1;
    }
    if (module_urc_sock_fd > 0)
    {
        close(module_urc_sock_fd);
        module_urc_sock_fd =-1;
    }  
}

void lynq_close_rc_thread()
{
    int ret;
    BLOCK_RECV_MSG_LOCK(); 
    module_rc_status = 0;    
    BLOCK_WAKEUP_RECV_MSG(); 
    if(module_rc_tid != -1)
    {
        ret = pthread_cancel(module_rc_tid);
        LYINFLOG("pthread cancel rc ret = %d",ret);     
    }    
    BLOCK_RECV_MSG_UNLOCK();     
    if(module_rc_tid != -1)
    {
        ret = pthread_join(module_rc_tid,NULL);
        module_rc_tid =-1;
        LYINFLOG("pthread join rc tid ret = %d",ret);
        
    }    

    if (module_rc_sock_fd > 0)
    {
        close(module_rc_sock_fd);
        module_rc_sock_fd =-1;
    }
    
    BLOCK_RECV_MSG_LOCK();    
    std::list<Parcel*>::iterator iter;
    for(iter=s_recv_parcel_list.begin();iter!=s_recv_parcel_list.end();++iter)
    {
        delete (*iter);        
    }
    s_recv_parcel_list.clear(); 
    BLOCK_RECV_MSG_UNLOCK(); 
}