update

wangzhengquan

2020-08-05 9e64ecf008d2258cab142b10f07eca6b538d6c19

update

 src/libshm_queue.a

Binary files differ

 src/queue/include/shm_queue.h

@@ -40,6 +40,7 @@

  static void remove_queues_exclude(int keys[], size_t length);
  static void remove_queues_include(int keys[], size_t length);
  static void remove_queue(int key);

private:
protected:
@@ -78,6 +79,7 @@
}



template <typename ELEM_T>
void SHMQueue<ELEM_T>::remove_queues_include(int keys[], size_t length) {
  hashtable_t *hashtable = mm_get_hashtable();
@@ -91,6 +93,12 @@
}

template <typename ELEM_T>
void SHMQueue<ELEM_T>::remove_queue(int key) {
  int keys[] = {key};
  remove_queues_include(keys, 1);
}

template <typename ELEM_T>
SHMQueue<ELEM_T>::SHMQueue(int key, size_t qsize) : KEY(key) {

  hashtable_t *hashtable = mm_get_hashtable();

 src/queue/include/shm_queue_wrapper.h

@@ -13,6 +13,8 @@

//移除不包含在keys中的队列
void shm_remove_queues_exclude(void *keys, int length);
//移除不包含在keys中的队列
void shm_remove_queue(int key);
/**
 * 创建队列
 * @ shmqueue 

 src/queue/shm_queue_wrapper.c

@@ -20,6 +20,9 @@
void shm_remove_queues_exclude(void *keys, int length) {
    SHMQueue<ele_t>::remove_queues_exclude((int*)keys, (size_t)length);
}
void shm_remove_queue(int key) {
    SHMQueue<ele_t>::remove_queue(key);
}

/**
 * 创建队列

 src/socket/dgram_mod_socket.c

File was deleted

 src/socket/dmod_socket.c

New file
@@ -0,0 +1,371 @@
#include "dmod_socket.h"


void DModSocket::foreach_subscripters(std::function<void(SHMKeySet *, SHMKeySet::iterator)>  cb) {
    SHMTopicSubMap *topic_sub_map = mem_pool_attach<SHMTopicSubMap>(BUS_MAP_KEY);
    SHMKeySet *subscripter_set;
    SHMKeySet::iterator set_iter;
    SHMTopicSubMap::iterator map_iter;

    if(topic_sub_map != NULL) {
        for (map_iter = topic_sub_map->begin(); map_iter != topic_sub_map->end(); map_iter++) {
            subscripter_set = map_iter->second;
            if(subscripter_set != NULL) {
                for(set_iter = subscripter_set->begin(); set_iter != subscripter_set->end(); set_iter++) {
                    cb(subscripter_set, set_iter);
                }
            }
        }
    }
}


DModSocket::DModSocket() {
        shm_socket = shm_open_socket(SHM_SOCKET_DGRAM);
}

DModSocket::~DModSocket() {
    SHMKeySet *subscripter_set;
    SHMTopicSubMap::iterator map_iter;

    if(topic_sub_map != NULL) {
        for (map_iter = topic_sub_map->begin(); map_iter != topic_sub_map->end(); map_iter++) {
            subscripter_set = map_iter->second;
            subscripter_set->clear();
            mm_free((void *)subscripter_set);
            //delete subscripter_set;
            // printf("=============delete subscripter_set\n");
        }
        topic_sub_map->clear();
        mem_pool_free_by_key(BUS_MAP_KEY);
    }
    // printf("=============close socket\n");
    shm_close_socket(shm_socket);
}

int DModSocket::bind(int port) {
    return  shm_socket_bind(shm_socket, port);
} 



/**
 * 强制绑定端口到socket, 适用于程序非正常关闭的情况下，重启程序绑定原来还没释放的key
 * @return 0 成功， 其他值 失败的错误码
*/
int DModSocket::force_bind(int port) {
    return shm_socket_force_bind(shm_socket, port);
}
/**
 * 发送信息
 * @port 发送给谁
 * @return 0 成功， 其他值 失败的错误码
 */
int DModSocket::sendto(const void *buf, const int size, const int port) {
        return shm_sendto(shm_socket, buf, size, port, NULL, 0);
}
// 发送信息超时返回。 @sec 秒 ， @nsec 纳秒
int DModSocket::sendto_timeout(const void *buf, const int size, const int port, const struct timespec *timeout) {
    return shm_sendto(shm_socket, buf, size, port, timeout, 0);
}
// 发送信息立刻返回。
int DModSocket::sendto_nowait( const void *buf, const int size, const int port){
    return shm_sendto(shm_socket, buf, size, port, NULL, (int)SHM_MSG_NOWAIT);
}


inline int DModSocket::_recvfrom_(void **buf, int *size, int *port,  struct timespec *timeout, int flags) {

    if(mod == BUS) {
        err_exit(0, "Can not use method recvfrom in a Bus");
    }
// printf("dgram_mod_recvfrom  before\n");
    int rv = shm_recvfrom(shm_socket, buf, size, port, timeout, flags);
// printf("dgram_mod_recvfrom  after\n");
    return rv;
}
/**
 * 接收信息
 * @port 从谁哪里收到的信息
 * @return 0 成功， 其他值 失败的错误码
*/
int DModSocket::recvfrom(void **buf, int *size, int *port) {
		
        return  _recvfrom_( buf, size, port, NULL, 0);
}


// 接受信息超时返回。 @sec 秒 ， @nsec 纳秒
int DModSocket::recvfrom_timeout( void **buf, int *size, int *port, struct timespec *timeout) {
    return _recvfrom_(buf, size, port, timeout, 0);
}

int DModSocket::recvfrom_nowait( void **buf, int *size, int *port){
    return _recvfrom_(buf, size, port, NULL, (int)SHM_MSG_NOWAIT);
}

/**
 * 发送请求信息并等待接收应答
 * @port 发送给谁
 * @return 0 成功， 其他值 失败的错误码
*/
int DModSocket::sendandrecv( const void *send_buf, const int send_size, const int send_port, void **recv_buf, int *recv_size){
    return shm_sendandrecv(shm_socket, send_buf, send_size, send_port, recv_buf, recv_size, NULL, 0);
}
// 超时返回。 @sec 秒 ， @nsec 纳秒
int DModSocket::sendandrecv_timeout(const void *send_buf, const int send_size, const int send_port, void **recv_buf, int *recv_size, struct timespec *timeout){
    return shm_sendandrecv(shm_socket, send_buf, send_size, send_port, recv_buf, recv_size, timeout, 0);
}
int DModSocket::sendandrecv_nowait(const void *send_buf, const int send_size, const int send_port, void **recv_buf, int *recv_size){
    return shm_sendandrecv(shm_socket, send_buf, send_size, send_port, recv_buf, recv_size, 0, (int)SHM_MSG_NOWAIT);
}


/**
 * 启动bus
 * 
 * @return 0 成功， 其他值 失败的错误码
*/
int  DModSocket::start_bus(){
    mod = BUS;
    topic_sub_map =    mem_pool_attach<SHMTopicSubMap>(BUS_MAP_KEY);
 
    run_pubsub_proxy();
    // pthread_t tid;
    // pthread_create(&tid, NULL, run_accept_sub_request, _socket);
    return 0;
}
/**
 * @port 总线端口
 */
int  DModSocket::_sub_( void *topic, int size, int port,  
    struct timespec *timeout, int flags) {
    char buf[8192];
    snprintf(buf,  8192, "%ssub%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, (char *)topic, TOPIC_RIDENTIFIER);
    return shm_sendto(shm_socket, buf, strlen(buf) + 1, port, timeout, flags);
}
/**
 * 订阅指定主题
 * @topic 主题
 * @size 主题长度
 * @port 总线端口
 */
int  DModSocket::sub( void *topic, int size, int port){
    return _sub_( topic, size, port, NULL, 0);
}
// 超时返回。 @sec 秒 ， @nsec 纳秒
int  DModSocket::sub_timeout(void *topic, int size, int port, struct timespec *timeout){
    return _sub_(topic, size, port, timeout, 0);
}
int  DModSocket::sub_nowait(void *topic, int size, int port) {
    return _sub_(topic, size, port, NULL,  (int)SHM_MSG_NOWAIT);
}


/**
 * @port 总线端口
 */
int  DModSocket::_pub_( void *topic, int topic_size, void *content, int content_size, int port,  
    struct timespec *timeout, int flags) {
    int head_len;
    char buf[8192+content_size];
    snprintf(buf, 8192, "%spub%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, (char *)topic, TOPIC_RIDENTIFIER);
    head_len = strlen(buf);
    memcpy(buf+head_len, content, content_size);
    return shm_sendto(shm_socket, buf, head_len+content_size, port, timeout, flags);

}

/**
 * 发布主题
 * @topic 主题
 * @content 主题内容
 * @port 总线端口
 */
int  DModSocket::pub(void *topic, int topic_size, void *content, int content_size, int port){
        return _pub_(topic, topic_size, content, content_size, port, NULL, 0);
}
//  超时返回。 @sec 秒 ， @nsec 纳秒
int  DModSocket::pub_timeout(void *topic, int topic_size, void *content, int content_size, int port, struct timespec * timeout){
    return _pub_( topic, topic_size, content, content_size, port, timeout, 0);
}
int  DModSocket::pub_nowait(void *topic, int topic_size, void *content, int content_size, int port){
    return _pub_(topic, topic_size, content, content_size, port, NULL, (int)SHM_MSG_NOWAIT);
}


/**
 * 获取soket端口号
 */
int DModSocket::get_port(){
    return shm_socket->port;
}



// ========================================================


/*
 * 处理订阅
*/
void DModSocket::_proxy_sub( char *topic, int port) {
    SHMKeySet *subscripter_set;

    SHMTopicSubMap::iterator map_iter;
    SHMKeySet::iterator set_iter;

    if( (map_iter = topic_sub_map->find(topic) ) != topic_sub_map->end()) {
        subscripter_set = map_iter->second;
    } else {
        void *set_ptr = mm_malloc(sizeof(SHMKeySet));
        subscripter_set = new(set_ptr) SHMKeySet;
        topic_sub_map->insert({topic, subscripter_set});
    }
    subscripter_set->insert(port);
}

/*
 * 处理发布，代理转发
*/
void DModSocket::_proxy_pub( char *topic, size_t head_len, void *buf, size_t size, int port) {
    SHMKeySet *subscripter_set;

    SHMTopicSubMap::iterator map_iter;
    SHMKeySet::iterator set_iter;

    std::vector<int> subscripter_to_del;
    std::vector<int>::iterator vector_iter;

    int send_port;
    struct timespec timeout = {1,0};

    if( (map_iter = topic_sub_map->find(topic) ) != topic_sub_map->end()) {
        subscripter_set = map_iter->second;
        for(set_iter = subscripter_set->begin(); set_iter != subscripter_set->end(); set_iter++) {
            send_port = *set_iter;
 // printf("_proxy_pub send before %d \n", send_port);
            if (shm_sendto(shm_socket, buf+head_len, size-head_len, send_port, &timeout) == SHM_SOCKET_ECONNFAILED ) {
                //对方已关闭的连接放到待删除队列里。如果直接删除会让iter指针出现错乱
                subscripter_to_del.push_back(send_port);
            } else {
// printf("_proxy_pub send after: %d \n", send_port);
            }

			
        }

        // 删除已关闭的端
        for(vector_iter = subscripter_to_del.begin(); vector_iter != subscripter_to_del.end(); vector_iter++) {
            if((set_iter = subscripter_set->find(*vector_iter)) != subscripter_set->end()) {
                subscripter_set->erase(set_iter);
                printf("remove closed subscripter %d \n", send_port);
            }
        }
        subscripter_to_del.clear();

    }
}

void * DModSocket::run_pubsub_proxy() {
    // pthread_detach(pthread_self());
    int size;
    int port;
    char * action, *topic, *topics, *buf;
    size_t head_len;

    const char *topic_delim = ",";
// printf("run_pubsub_proxy server receive before\n");
    while(shm_recvfrom(shm_socket, (void **)&buf, &size, &port) == 0) {
// printf("run_pubsub_proxy server recv after: %s \n", buf);
        if(parse_pubsub_topic(buf, size, &action, &topics, &head_len)) {
            if(strcmp(action, "sub") == 0) {
                // 订阅支持多主题订阅
                topic = trim(strtok(topics, topic_delim), NULL);
              while(topic) {
           _proxy_sub( topic, port);
            topic = trim(strtok(NULL, topic_delim), NULL);
              }

            } else if(strcmp(action, "pub") == 0) {
                _proxy_pub(topics, head_len, buf, size, port);
            }
			
            free(action);
            free(topics);
        } else {
            err_msg(0, "incorrect format msg");
        }
        free(buf);
    }
    return NULL;
}


/**
 * @str "<**sub**>{经济}"
 */

int DModSocket::parse_pubsub_topic(char *str, size_t size, char **_action, char **_topic, size_t *head_len ) {
 char *ptr = str;
 char *str_end_ptr = str + size;
 char *action_start_ptr;
 char *action_end_ptr;
 size_t action_len = 0;

 char *topic_start_ptr;
 char *topic_end_ptr;
 size_t topic_len = 0;

 // if (strlen(identifier) > strlen(str)) {
 //  return 0;
 // }

 if (strncmp(ptr, ACTION_LIDENTIFIER, strlen(ACTION_LIDENTIFIER)) == 0) {
  ptr += strlen(ACTION_LIDENTIFIER);
  action_start_ptr = ptr;
  while(strncmp(++ptr, ACTION_RIDENTIFIER, strlen(ACTION_RIDENTIFIER)) != 0) {
    if(ptr >= str_end_ptr) {
      return 0;
    }
  }
// printf("%s\n", ptr);
  action_end_ptr = ptr;
  action_len = action_end_ptr - action_start_ptr;
  ptr += strlen(ACTION_RIDENTIFIER);
// printf("%s\n", ptr);
// printf("%s\n", str_end_ptr-1);
  if(strncmp(ptr, TOPIC_LIDENTIFIER, strlen(TOPIC_LIDENTIFIER)) == 0 ) {
    topic_start_ptr = ptr+strlen(TOPIC_LIDENTIFIER);
   

    while(strncmp(++ptr, TOPIC_RIDENTIFIER, strlen(TOPIC_RIDENTIFIER)) != 0) {
      if(ptr >= str_end_ptr) {
        return 0;
      }
    }
    topic_end_ptr = ptr;
    topic_len = topic_end_ptr - topic_start_ptr;
    
    ptr += strlen(TOPIC_RIDENTIFIER);
   
  } else {
    return 0;
  }
 } else {
  return 0;
 }

 char *topic = (char *)calloc(1, topic_len+1);
 strncpy(topic, topic_start_ptr, topic_len);
 *_topic = topic;

 char *action = (char *)calloc(1, action_len+1);
 strncpy(action, action_start_ptr, action_len);
 *_action = action;
 *head_len = ptr-str;

 return 1;
}

 

 src/socket/include/dmod_socket.h

New file
@@ -0,0 +1,138 @@
#ifndef __DMODE_SOCKET_H__
#define __DMODE_SOCKET_H__
#include "usg_common.h"
#include "shm_socket.h"
#include "shm_allocator.h"
#include "mem_pool.h"
#include "hashtable.h"
#include "sem_util.h"
#include "logger_factory.h"
#include <set>

#define ACTION_LIDENTIFIER "<**"
#define ACTION_RIDENTIFIER "**>"
#define TOPIC_LIDENTIFIER "{"
#define TOPIC_RIDENTIFIER "}"

static Logger logger = LoggerFactory::getLogger();
#define BUS_MAP_KEY 1
//typedef std::basic_string<char, std::char_traits<char>, SHM_STL_Allocator<char> > SHMString;
typedef std::set<int,  std::less<int>, SHM_STL_Allocator<int> > SHMKeySet;
typedef std::map<SHMString, SHMKeySet *, std::less<SHMString>, SHM_STL_Allocator<std::pair<SHMString, SHMKeySet *> > > SHMTopicSubMap;

enum socket_mod_t
{
    PULL_PUSH = 1,
    REQ_REP = 2,
    PAIR = 3,
    PUB_SUB = 4,
    SURVEY = 5,
    BUS = 6
	
};

class DModSocket {
private:
    shm_socket_t *shm_socket;
  socket_mod_t mod;
  // pthread_t recv_thread;
  // <主题， 订阅者>
    SHMTopicSubMap *topic_sub_map;

private:
    inline int _recvfrom_(void **buf, int *size, int *port,  struct timespec *timeout, int flags);
    void _proxy_sub( char *topic, int port);
    void _proxy_pub( char *topic, size_t head_len, void *buf, size_t size, int port);
    void *run_pubsub_proxy();
    int parse_pubsub_topic(char *str, size_t size, char **_action, char **_topic, size_t *head_len );
    int _sub_( void *topic, int size, int port, struct timespec *timeout, int flags);
    int _pub_( void *topic, int topic_size, void *content, int content_size, int port, struct timespec *timeout, int flags);

    static void  foreach_subscripters(std::function<void(SHMKeySet *, SHMKeySet::iterator)>  cb) ;
public:
    DModSocket();
    ~DModSocket();
	 

    /**
     * 绑定端口到socket, 如果不绑定则系统自动分配一个
     * @return 0 成功， 其他值 失败的错误码
    */
    int bind(int port);

    /**
     * 强制绑定端口到socket, 适用于程序非正常关闭的情况下，重启程序绑定原来还没释放的key
     * @return 0 成功， 其他值 失败的错误码
    */
    int force_bind(int port);
    /**
     * 发送信息
     * @port 发送给谁
     * @return 0 成功， 其他值 失败的错误码
     */
    int sendto(const void *buf, const int size, const int port);
    // 发送信息超时返回。 @sec 秒 ， @nsec 纳秒
    int sendto_timeout(const void *buf, const int size, const int port, const struct timespec *timeout);
    // 发送信息立刻返回。
    int sendto_nowait(const void *buf, const int size, const int port);

    /**
     * 接收信息
     * @port 从谁哪里收到的信息
     * @return 0 成功， 其他值 失败的错误码
    */
    int recvfrom(void **buf, int *size, int *port);
    // 接受信息超时返回。 @sec 秒 ， @nsec 纳秒
    int recvfrom_timeout(void **buf, int *size, int *port,  struct timespec *timeout);
    int recvfrom_nowait(void **buf, int *size, int *port);

    /**
     * 发送请求信息并等待接收应答
     * @port 发送给谁
     * @return 0 成功， 其他值 失败的错误码
    */
    int sendandrecv(const void *send_buf, const int send_size, const int port, void **recv_buf, int *recv_size) ;
    // 超时返回。 @sec 秒 ， @nsec 纳秒
    int sendandrecv_timeout(const void *send_buf, const int send_size, const int port, void **recv_buf, int *recv_size,  struct timespec *timeout) ;
    int sendandrecv_nowait(const void *send_buf, const int send_size, const int port, void **recv_buf, int *recv_size) ;


    /**
     * 启动bus
     * 
     * @return 0 成功， 其他值 失败的错误码
    */
    int  start_bus();

    /**
     * 订阅指定主题
     * @topic 主题
     * @size 主题长度
     * @port 总线端口
     */
    int  sub(void *topic, int size, int port);
    // 超时返回。 @sec 秒 ， @nsec 纳秒
    int  sub_timeout(void *topic, int size, int port,  struct timespec *timeout);
    int  sub_nowait(void *topic, int size, int port);



    /**
     * 发布主题
     * @topic 主题
     * @content 主题内容
     * @port 总线端口
     */
    int  pub(void *topic, int topic_size, void *content, int content_size, int port);
    //  超时返回。 @sec 秒 ， @nsec 纳秒
    int  pub_timeout(void *topic, int topic_size, void *content, int content_size, int port,  struct timespec *timeout);
    int  pub_nowait(void *topic, int topic_size, void *content, int content_size, int port);


    /**
     * 获取soket端口号
     */
    int get_port() ;
};

#endif

 src/socket/include/shm_socket.h

@@ -30,8 +30,8 @@
};

enum shm_socket_error_type_t {
    SHM_SOCKET_CONN_FAILED = 1,
    SHM_SOCKET_TIMEOUT = 2	
    SHM_SOCKET_ECONNFAILED = 1,
    SHM_SOCKET_ETIMEOUT = 2	
};

enum shm_connection_status_t {

 src/socket/shm_socket.c

@@ -270,7 +270,7 @@
  SHMQueue<shm_msg_t> *remoteQueue;
  if ((remoteQueue = _attach_remote_queue(port)) == NULL) {
      err_msg(0, "shm_sendto failed, the other end has been closed, or has not been opened!");
    return SHM_SOCKET_CONN_FAILED;
    return SHM_SOCKET_ECONNFAILED;
  }

  shm_msg_t dest;

 test_socket/dgram_mod_req_rep.c

@@ -160,13 +160,14 @@
  port = atoi(argv[2]);

  if (strcmp("server", argv[1]) == 0) {
    int temp = shm_alloc_key();
    printf("tmp=%d\n", temp);
    server(temp);
    // int temp = shm_alloc_key();
    // printf("tmp=%d\n", temp);
    server(port);
  } else if (strcmp("client", argv[1]) == 0) {
    startClients(port);
  } else {
    printf("input invalidate arguments\n");
  }

  if (strcmp("client", argv[1]) == 0)
    client2(port);

  
  return 0;