commit

wangzhengquan

2020-07-25 8ec1d776751f62c43335d36c3427dc1ab2d84a61

commit

 demo/Makefile

@@ -14,7 +14,7 @@
include $(ROOT)/Make.defines.$(PLATFORM)


PROGS = dgram_mod_req_rep dgram_mod_survey
PROGS =


build: $(PROGS)

 demo/dgram_mod_req_rep

Binary files differ

 demo/dgram_mod_survey

Binary files differ

 src/libshm_queue.a

Binary files differ

 src/logger_factory.h

@@ -6,8 +6,8 @@
public:

    static Logger getLogger() {
//ERROR ALL DEBUG INFO
        static Logger logger(Logger::ERROR);
//ERROR ALL DEBUG INFO WARN
        static Logger logger(Logger::WARN);
        return logger;
    }
};

 src/queue/include/lock_free_queue.h

@@ -200,13 +200,16 @@
    template <typename T, typename AT> class Q_TYPE>
bool LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::push(const ELEM_T &a_data)
{
// printf("==================LockFreeQueue push before\n");   
    if (SemUtil::dec(slots) == -1) {
        err_msg(errno, "LockFreeQueue push");
        return false;
    }

    if ( m_qImpl.push(a_data) ) {

        SemUtil::inc(items);      
// printf("==================LockFreeQueue push after\n");   
        return true;
    }
    return false;
@@ -244,6 +247,7 @@
bool LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::push_timeout(const ELEM_T &a_data, struct timespec * timeout)
{


    if (SemUtil::dec_timeout(slots, timeout) == -1) {
        if (errno == EAGAIN)
            return false;
@@ -270,6 +274,7 @@
    template <typename T, typename AT> class Q_TYPE>
bool LockFreeQueue<ELEM_T, Allocator, Q_TYPE>::pop(ELEM_T &a_data)
{
// printf("==================LockFreeQueue pop before\n");
    if (SemUtil::dec(items) == -1) {
        err_msg(errno, "LockFreeQueue pop");
        return false;
@@ -277,6 +282,7 @@

    if (m_qImpl.pop(a_data)) {
        SemUtil::inc(slots);      
// printf("==================LockFreeQueue pop after\n");      
        return true;
    }
    return false;

 src/queue/include/shm_queue.h

@@ -104,7 +104,7 @@
        delete queue;
        hashtable_t *hashtable = mm_get_hashtable();
        hashtable_remove(hashtable, KEY);
        LoggerFactory::getLogger().debug("SHMQueue destructor delete queue");
printf("SHMQueue destructor delete queue\n");
    } else {
        SemUtil::inc(queue->mutex);
    }
@@ -159,7 +159,10 @@
template < typename ELEM_T >
inline bool SHMQueue<ELEM_T>::pop(ELEM_T &a_data)
{
   return queue->pop(a_data);
// printf("SHMQueue pop before\n");
   int rv = queue->pop(a_data);
// printf("SHMQueue after before\n");
   return rv;
    
}


 src/socket/dgram_mod_socket.c

@@ -6,18 +6,28 @@
#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 int parse_pubsub_topic(char *str, size_t size, char **_action, char **_topic, size_t *head_len );
void * run_accept_pubsub_request(void * _socket) ;

typedef struct dgram_mod_socket_t {
    socket_mod_t mod;
  shm_socket_t *shm_socket;
  // pthread_t recv_thread;
    // std::map<int, LockFreeQueue<shm_msg_t, DM_Allocator> * > *recv_queue_map;
  // <主题， 订阅者>
    std::map<std::string, std::set<int> *> *topic_sub_map;
} dgram_mod_socket_t;


void *dgram_mod_open_socket(int mod) {
void *dgram_mod_open_socket() {
    dgram_mod_socket_t * socket = (dgram_mod_socket_t *)calloc(1, sizeof(dgram_mod_socket_t));
    socket->mod = (socket_mod_t)mod;
    // socket->mod = (socket_mod_t)mod;
    socket->shm_socket = shm_open_socket(SHM_SOCKET_DGRAM);
    return (void *)socket;
}
@@ -25,6 +35,19 @@

int dgram_mod_close_socket(void * _socket) {
    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
    std::map<std::string, std::set<int> *> *topic_sub_map = socket->topic_sub_map;
    std::set<int> *subscripter_set;
    std::map<std::string, std::set<int> *>::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;
            delete subscripter_set;
        }
        delete topic_sub_map;
    }
	

    shm_close_socket(socket->shm_socket);
    free(_socket);
}
@@ -44,7 +67,10 @@
int dgram_mod_recvfrom(void *_socket, void **buf, int *size, int *port) {

    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
    return shm_recvfrom(socket->shm_socket, buf, size, port);
// printf("dgram_mod_recvfrom  before\n");
    int rv = shm_recvfrom(socket->shm_socket, buf, size, port);
// printf("dgram_mod_recvfrom  after\n");
    return rv;
}
 

@@ -66,3 +92,166 @@
void dgram_mod_free(void *buf) {
    free(buf);
}

int start_bus(void * _socket) {
    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
    socket->topic_sub_map = new std::map<std::string, std::set<int> *>;
    run_accept_pubsub_request(_socket);
    // pthread_t tid;
    // pthread_create(&tid, NULL, run_accept_sub_request, _socket);
    return 0;

}

/**
 * @port 总线端口
 */
int sub(void * _socket, void *topic, int size, int port) {
    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
    char buf[8192];
    snprintf(buf,  8192, "%ssub%s%s%s%s", ACTION_LIDENTIFIER, ACTION_RIDENTIFIER, TOPIC_LIDENTIFIER, (char *)topic, TOPIC_RIDENTIFIER);
    return shm_sendto(socket->shm_socket, buf, strlen(buf) + 1, port);
}

/**
 * @port 总线端口
 */
int pub(void * _socket, void *topic, int topic_size, void *content, int content_size, int port) {

    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
    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(socket->shm_socket, buf, head_len+content_size, port);

}


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


void * run_accept_pubsub_request(void * _socket) {
    // pthread_detach(pthread_self());
    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
    int size;
    int port;
    char * action, *topic, *buf;

    size_t head_len;
    // void * send_buf;
    int send_port;
    struct timespec timeout = {1,0};
    std::map<std::string, std::set<int> *> *topic_sub_map = socket->topic_sub_map;
    std::set<int> *subscripter_set;

    std::map<std::string, std::set<int> *>::iterator map_iter;
    std::set<int>::iterator set_iter;
//printf("server receive before\n");
    while(shm_recvfrom(socket->shm_socket, (void **)&buf, &size, &port) == 0) {
//printf("server recv after: %s \n", buf);
        if(parse_pubsub_topic(buf, size, &action, &topic, &head_len)) {
            if(strcmp(action, "sub") == 0) {
                if( (map_iter = topic_sub_map->find(topic) ) != topic_sub_map->end()) {
                    subscripter_set = map_iter->second;
                } else {
                    subscripter_set = new std::set<int>;
                    topic_sub_map->insert({topic, subscripter_set});
                }
                subscripter_set->insert(port);

            } else if(strcmp(action, "pub") == 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;
                        // send_buf = malloc(size-head_len);
                        // memcpy(send_buf, buf+head_len, );
printf("run_accept_sub_request send before %d \n", send_port);
                        if (shm_sendto(socket->shm_socket, buf+head_len, size-head_len, send_port, &timeout) !=0 ) {
printf("erase %d \n", send_port);
                            subscripter_set->erase(set_iter);
                            set_iter++;
                        }
printf("run_accept_sub_request send after: %d \n", send_port);
						
                    }
                }
            }
            free(buf);
            free(action);
            free(topic);
        } else {
            err_msg(0, "incorrect format msg");
        }
    }
    return NULL;
}


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

static int 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/dgram_mod_socket.h

@@ -18,7 +18,7 @@
    
};

void *dgram_mod_open_socket(int mod);
void *dgram_mod_open_socket();

int dgram_mod_close_socket(void * _socket);

@@ -34,6 +34,19 @@

void dgram_mod_free(void *buf) ;

int start_bus(void * _socket);

/**
 * @port 总线端口
 */
int sub(void * _socket, void *topic, int size, int port);

/**
 * @port 总线端口
 */
int pub(void * _socket, void *topic, int topic_size, void *content, int content_size, int port);


#ifdef __cplusplus
}
#endif

 src/socket/include/shm_socket.h

@@ -77,7 +77,7 @@

int shm_recv(shm_socket_t * socket, void **buf, int *size) ;

int shm_sendto(shm_socket_t *socket, const void *buf, const int size, const int port);
int shm_sendto(shm_socket_t *socket, const void *buf, const int size, const int port, const struct timespec * timeout = NULL);

int shm_recvfrom(shm_socket_t *socket, void **buf, int *size, int *port);


 src/socket/shm_socket.c

@@ -3,8 +3,6 @@
#include "logger_factory.h"
#include <map>



static Logger logger = LoggerFactory::getLogger();

void print_msg(char *head, shm_msg_t& msg) {
@@ -16,7 +14,6 @@
void * _client_run_msg_rev(void* _socket);

int _shm_close_dgram_socket(shm_socket_t *socket);


int _shm_close_stream_socket(shm_socket_t *socket, bool notifyRemote);

@@ -42,7 +39,6 @@
            return -1;
    }
    return -1;
	
}

int shm_socket_bind(shm_socket_t * socket, int port) {
@@ -53,7 +49,8 @@
int shm_listen(shm_socket_t* socket) {

    if(socket->socket_type != SHM_SOCKET_STREAM) {
        err_exit(0, "can not invoke shm_listen method with a socket which is not a SHM_SOCKET_STREAM socket");
    err_exit(0, "can not invoke shm_listen method with a socket which is not a "
                "SHM_SOCKET_STREAM socket");
    }

    int  port;
@@ -72,12 +69,11 @@
    socket->acceptQueue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
    socket->clientSocketMap = new std::map<int, shm_socket_t* >;
    socket->status = SHM_CONN_LISTEN;
    pthread_create(&(socket->dispatch_thread), NULL, _server_run_msg_rev , (void *)socket);

  pthread_create(&(socket->dispatch_thread), NULL, _server_run_msg_rev,
                 (void *)socket);
    
    return 0;
}


/**
 * 接受客户端建立新连接的请求
@@ -85,7 +81,8 @@
*/
shm_socket_t* shm_accept(shm_socket_t* socket) {
    if(socket->socket_type != SHM_SOCKET_STREAM) {
        err_exit(0, "can not invoke shm_accept method with a socket which is not a SHM_SOCKET_STREAM socket");
    err_exit(0, "can not invoke shm_accept method with a socket which is not a "
                "SHM_SOCKET_STREAM socket");
    }
    hashtable_t *hashtable = mm_get_hashtable();
    int client_port;
@@ -101,14 +98,16 @@
        client_socket->port = socket->port;
        // client_socket->queue= socket->queue;
        //初始化消息queue
        client_socket->messageQueue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16); 
    client_socket->messageQueue =
        new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
        //连接到对方queue
        client_socket->remoteQueue = _attach_remote_queue(client_port);

        socket->clientSocketMap->insert({client_port, client_socket});

        /*
         * shm_accept 用户执行的方法 与_server_run_msg_rev在两个不同的限制工作,accept要保证在客户的发送消息之前完成资源的准备工作，以避免出现竞态问题
* shm_accept 用户执行的方法
* 与_server_run_msg_rev在两个不同的限制工作,accept要保证在客户的发送消息之前完成资源的准备工作，以避免出现竞态问题
        */
        //发送open_reply,回应客户端的connect请求
        struct timespec timeout = {1, 0};
@@ -117,8 +116,7 @@
        msg.size = 0;
        msg.type = SHM_SOCKET_OPEN_REPLY;

        if (client_socket->remoteQueue->push_timeout(msg, &timeout) )
        {
    if (client_socket->remoteQueue->push_timeout(msg, &timeout)) {
            client_socket->status = SHM_CONN_ESTABLISHED;
            return client_socket;
        } else {
@@ -126,18 +124,16 @@
            return NULL;
        }

		
    } else {
        err_exit(errno, "shm_accept");
    }
    return NULL;
	
}


int shm_connect(shm_socket_t* socket, int port) {
    if(socket->socket_type != SHM_SOCKET_STREAM) {
        err_exit(0, "can not invoke shm_connect method with a socket which is not a SHM_SOCKET_STREAM socket");
    err_exit(0, "can not invoke shm_connect method with a socket which is not "
                "a SHM_SOCKET_STREAM socket");
    }
    hashtable_t *hashtable = mm_get_hashtable();
    if(hashtable_get(hashtable, port)== NULL) {
@@ -154,9 +150,11 @@
    }

    socket->queue = new SHMQueue<shm_msg_t>(socket->port, 16);
    socket->remoteQueue = _attach_remote_queue(port);
    socket->messageQueue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16); 
    
  if ((socket->remoteQueue = _attach_remote_queue(port)) == NULL) {
    err_exit(0, "connect to %d failted", port);
  }
  socket->messageQueue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);

    //发送open请求
    struct timespec timeout = {1, 0};
@@ -171,7 +169,8 @@
        // 在这里server端已经准备好接受客户端发送请求了,完成与服务端的连接
        if(msg.type == SHM_SOCKET_OPEN_REPLY) {
            socket->status = SHM_CONN_ESTABLISHED;
            pthread_create(&(socket->dispatch_thread), NULL, _client_run_msg_rev , (void *)socket);
      pthread_create(&(socket->dispatch_thread), NULL, _client_run_msg_rev,
                     (void *)socket);
        } else {
            err_exit(0, "shm_connect: 不匹配的应答信息!");
        }
@@ -183,10 +182,10 @@
    return 0;
}


int shm_send(shm_socket_t *socket, const void *buf, const int size) {
    if(socket->socket_type != SHM_SOCKET_STREAM) {
        err_exit(0, "can not invoke shm_send method with a socket which is not a SHM_SOCKET_STREAM socket");
    err_exit(0, "can not invoke shm_send method with a socket which is not a "
                "SHM_SOCKET_STREAM socket");
    }
    // hashtable_t *hashtable = mm_get_hashtable();
    // if(socket->remoteQueue == NULL) {
@@ -200,7 +199,6 @@
    dest.buf = mm_malloc(size);
    memcpy(dest.buf, buf, size);

 
    if(socket->remoteQueue->push(dest)) {
        return 0;
    } else {
@@ -209,10 +207,11 @@
    }
}


int shm_recv(shm_socket_t* socket, void **buf, int *size) {
    if(socket->socket_type != SHM_SOCKET_STREAM) {
        err_exit(0, "can not invoke shm_recv method in a %d type socket  which is not a SHM_SOCKET_STREAM socket ", socket->socket_type);
    err_exit(0, "can not invoke shm_recv method in a %d type socket  which is "
                "not a SHM_SOCKET_STREAM socket ",
             socket->socket_type);
    }
    shm_msg_t src;

@@ -226,15 +225,15 @@
    } else {
        return -1;
    }
	
}



// 短连接方式发送
int shm_sendto(shm_socket_t *socket, const void *buf, const int size, const int port) {
int shm_sendto(shm_socket_t *socket, const void *buf, const int size,
               const int port, const struct timespec *timeout) {
    if(socket->socket_type != SHM_SOCKET_DGRAM) {
        err_exit(0, "Can't invoke shm_sendto method in a %d type socket  which is not a SHM_SOCKET_DGRAM socket ", socket->socket_type);
    err_exit(0, "Can't invoke shm_sendto method in a %d type socket  which is "
                "not a SHM_SOCKET_DGRAM socket ",
             socket->socket_type);
    }
    hashtable_t *hashtable = mm_get_hashtable();

@@ -262,8 +261,21 @@
    dest.buf = mm_malloc(size);
    memcpy(dest.buf, buf, size);

    SHMQueue<shm_msg_t> *remoteQueue =  _attach_remote_queue(port);
    if(remoteQueue->push(dest)) {
  SHMQueue<shm_msg_t> *remoteQueue;
  if ((remoteQueue = _attach_remote_queue(port)) == NULL) {
      err_msg(0, "shm_sendto failed, then other end has been closed!");
    return -1;
  }
  // printf("shm_sendto push before\n");
  bool rv;
  if(timeout != NULL) {
      rv = remoteQueue->push_timeout(dest, timeout);
  } else {
      rv = remoteQueue->push(dest);
  }

  if (rv) {
    // printf("shm_sendto push after\n");
        delete remoteQueue;
        return 0;
    } else {
@@ -273,11 +285,12 @@
    }
}


// 短连接方式接受
int shm_recvfrom(shm_socket_t *socket, void **buf, int *size, int *port){
    if(socket->socket_type != SHM_SOCKET_DGRAM) {
        err_exit(0, "Can't invoke shm_recvfrom method in a %d type socket  which is not a SHM_SOCKET_DGRAM socket ", socket->socket_type);
    err_exit(0, "Can't invoke shm_recvfrom method in a %d type socket  which "
                "is not a SHM_SOCKET_DGRAM socket ",
             socket->socket_type);
    }
    hashtable_t *hashtable = mm_get_hashtable();
    if(socket->queue == NULL) {
@@ -294,7 +307,7 @@
    }

    shm_msg_t src;
// printf("shm_recvfrom pop before");
  // printf("shm_recvfrom pop before\n");
    if (socket->queue->pop(src)) {
        void * _buf = malloc(src.size);
        memcpy(_buf, src.buf, src.size);
@@ -302,16 +315,20 @@
        *size = src.size;
        *port = src.port;
        mm_free(src.buf);
// printf("shm_recvfrom pop after");
    // printf("shm_recvfrom pop after\n");
        return 0;
    } else {
        return -1;
    }
}

int shm_sendandrecv(shm_socket_t *socket, const void *send_buf, const int send_size, const int send_port, void **recv_buf, int *recv_size) {
int shm_sendandrecv(shm_socket_t *socket, const void *send_buf,
                    const int send_size, const int send_port, void **recv_buf,
                    int *recv_size) {
    if(socket->socket_type != SHM_SOCKET_DGRAM) {
        err_exit(0, "Can't invoke shm_sendandrecv method in a %d type socket  which is not a SHM_SOCKET_DGRAM socket ", socket->socket_type);
    err_exit(0, "Can't invoke shm_sendandrecv method in a %d type socket  "
                "which is not a SHM_SOCKET_DGRAM socket ",
             socket->socket_type);
    }
    int recv_port;
    int rv;
@@ -325,14 +342,13 @@
    return -1;
}


/**
 * 绑定key到队列，但是并不会创建队列。如果没有对应指定key的队列提示错误并退出
 */
SHMQueue<shm_msg_t> * _attach_remote_queue(int port) {
    hashtable_t *hashtable = mm_get_hashtable();
    if(hashtable_get(hashtable, port)== NULL) {
        err_exit(0, "_remote_queue_attach：connet at port %d  failed!", port);
    err_msg(0, "_remote_queue_attach：connet at port %d  failed!", port);
        return NULL;
    }
     
@@ -340,20 +356,15 @@
    return queue;
}





void _server_close_conn_to_client(shm_socket_t* socket, int port) {
    shm_socket_t *client_socket;
    std::map<int, shm_socket_t* >::iterator iter = socket->clientSocketMap->find(port);
  std::map<int, shm_socket_t *>::iterator iter =
      socket->clientSocketMap->find(port);
    if( iter !=  socket->clientSocketMap->end() ) {
        client_socket = iter->second;
        free((void *)client_socket);
        socket->clientSocketMap->erase(iter);
    }
	

}

/**
@@ -396,13 +407,10 @@
    return NULL;
}



void _client_close_conn_to_server(shm_socket_t* socket) {
     
    _shm_close_stream_socket(socket, false);
}


/**
 * client端的各种类型消息（）在这里进程分拣
@@ -429,7 +437,6 @@
 
    return NULL;
}


int _shm_close_stream_socket(shm_socket_t *socket, bool notifyRemote) {
    socket->status = SHM_CONN_CLOSED;
@@ -466,7 +473,8 @@

    if(socket->clientSocketMap != NULL) {
        shm_socket_t *client_socket;
        for(auto iter = socket->clientSocketMap->begin(); iter != socket->clientSocketMap->end(); iter++) {
    for (auto iter = socket->clientSocketMap->begin();
         iter != socket->clientSocketMap->end(); iter++) {
            client_socket= iter->second;

            client_socket->remoteQueue->push_timeout(close_msg, &timeout);
@@ -480,7 +488,6 @@
        }
        delete socket->clientSocketMap;
    }


    if(socket->dispatch_thread != 0)
        pthread_cancel(socket->dispatch_thread);

 src/util/include/sem_util.h

@@ -9,7 +9,7 @@
    int get(key_t key, unsigned int value);
    int dec(int semId);
    int dec_nowait(int semId);
    int dec_timeout(int semId, struct timespec * timeout);
    int dec_timeout(const int semId, const struct timespec * timeout);
    int inc(int semId);
    void remove(int semid);


 src/util/sem_util.c

@@ -72,6 +72,7 @@

/* Reserve semaphore - decrement it by 1 */
int SemUtil::dec(int semId) {
logger.debug("%d: SemUtil::dec\n", semId);
  struct sembuf sops;

  sops.sem_num = 0;
@@ -103,7 +104,7 @@
  return 0;
}

int SemUtil::dec_timeout(int semId, struct timespec *timeout) {
int SemUtil::dec_timeout(const int semId, const struct timespec *timeout) {
  struct sembuf sops;

  sops.sem_num = 0;
@@ -121,6 +122,7 @@

/* Release semaphore - increment it by 1 */
int SemUtil::inc(int semId) {
logger.debug("%d: SemUtil::inc\n", semId);
  struct sembuf sops;

  sops.sem_num = 0;

 test/Makefile

@@ -14,7 +14,7 @@
include $(ROOT)/Make.defines.$(PLATFORM)

 
PROGS = protocle_parse
PROGS = protocle_parse test

build: $(PROGS)


 test/protocle_parse

Binary files differ

 test/protocle_parse.c

@@ -3,10 +3,10 @@

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

int parse_pubsub_topic(char *str, char **_action, size_t *_action_len,  char **_topic, size_t *_topic_len) {
int parse_pubsub_topic(char *str, char **_action, char **_topic, size_t *head_len ) {
 char *ptr = str;
 char *str_end_ptr = str + strlen(str);
 char *action_start_ptr;
@@ -35,14 +35,20 @@
  ptr += strlen(ACTION_RIDENTIFIER);
// printf("%s\n", ptr);
// printf("%s\n", str_end_ptr-1);
  if( (*ptr == TOPIC_LIDENTIFIER) && (*(str_end_ptr-1) == TOPIC_RIDENTIFIER) ) {
    topic_start_ptr = ptr;
    topic_end_ptr = str_end_ptr;
    topic_len = topic_end_ptr - topic_start_ptr + 1;
    ptr++;
    // while(*(++ptr) != '}') {
    //   length++;
    // }
  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;
  }
@@ -53,13 +59,11 @@
 char *topic = (char *)calloc(1, topic_len+1);
 strncpy(topic, topic_start_ptr, topic_len);
 *_topic = topic;
 *_topic_len = topic_len;

 char *action = (char *)calloc(1, action_len+1);
 strncpy(action, action_start_ptr, action_len);
 *_action = action;
 *_action_len = action_len;

 *head_len = ptr-str;

 return 1;
}
@@ -68,12 +72,13 @@
 char *action;
 size_t action_len;
 char *topic;
 size_t topic_len;
 size_t head_len;
  
 char *str = "<**sub**>{经济}";
 if(parse_pubsub_topic(str, &action, &action_len, &topic, &topic_len)) {
 char *str = "<**pub**>{经济}abcdef";
 if(parse_pubsub_topic(str, &action, &topic, &head_len)) {
  printf("action:%s\n", action);
  printf("topic:%s\n", topic);
  printf("content:%s\n", str+head_len);
  free(action);
  free(topic);
 } else {

 test/test

Binary files differ

 test/test.c

New file
@@ -0,0 +1,12 @@
#include "usg_common.h"
#include "usg_typedef.h"

void test(char *src, int size) {
    char dest[size];
    strncpy(dest, src, size);
    puts(dest);
}
int main() {
    char *str = "hello";
    test(str, strlen(str));
}

 test_socket/Makefile

@@ -14,7 +14,7 @@
include $(ROOT)/Make.defines.$(PLATFORM)


PROGS =    dgram_socket_test dgram_mod_req_rep dgram_mod_survey stream_mod_req_rep
PROGS =    dgram_mod_bus dgram_mod_survey


build: $(PROGS)

 test_socket/dgram_mod_bus

Binary files differ

 test_socket/dgram_mod_bus.c

New file
@@ -0,0 +1,86 @@
#include "dgram_mod_socket.h"
#include "shm_mm.h"
#include "usg_common.h"

void server(int port) {
  void *socket = dgram_mod_open_socket();
  dgram_mod_bind(socket, port);
   
  start_bus(socket);
  
}


void *run_recv(void *socket) {
  pthread_detach(pthread_self());
  void *recvbuf;
  int size;
  int port;
  while (dgram_mod_recvfrom( socket, &recvbuf, &size, &port) == 0) {
    printf("收到订阅消息:%s\n", recvbuf);
    free(recvbuf);
  }
  
}

void client(int port) {
  void *socket = dgram_mod_open_socket();
  pthread_t tid;
  pthread_create(&tid, NULL, run_recv, socket);
  int size;
  
  char action[512];
  char topic[512];
  char content[512];
  long i = 0;
  while (true) {
    //printf("Usage: pub <topic> [content] or sub <topic>\n");
    printf("Can I help you? sub, pub or quit\n");
    scanf("%s",action);
    
    if(strcmp(action, "sub") == 0) {
      printf("Please input topic!\n");
      scanf("%s", topic);
      sub(socket, topic, strlen(topic),  port);
      printf("Sub success!\n");
    }
    else if(strcmp(action, "pub") == 0) {
      // printf("%s %s %s\n", action, topic, content);
      printf("Please input topic and content\n");
      scanf("%s %s", topic, content);
      pub(socket, topic, strlen(topic)+1, content, strlen(content)+1,  port);
      printf("Pub success!\n");
    } else if(strcmp(action, "quit") == 0) {
      break;
    } else {
      printf("error input\n");
      continue;
    }
   
  }
  printf("(%d) quit\n", dgram_mod_get_socket_port(socket));
  dgram_mod_close_socket(socket);
}

 

int main(int argc, char *argv[]) {
  shm_init(512);
  int port;
  if (argc < 3) {
    fprintf(stderr, "Usage: reqrep %s|%s <PORT> ...\n", "server", "client");
    return 1;
  }

  port = atoi(argv[2]);

  if (strcmp("server", argv[1]) == 0) {
    server(port);
  }

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

  
  return 0;
}

 test_socket/dgram_mod_req_rep

Binary files differ

 test_socket/dgram_mod_survey

Binary files differ

 test_socket/dgram_mod_survey.c

@@ -3,7 +3,7 @@
#include "usg_common.h"

void server(int port) {
  void *socket = dgram_mod_open_socket(SURVEY);
  void *socket = dgram_mod_open_socket();
  dgram_mod_bind(socket, port);
  int size;
  void *recvbuf;
@@ -18,16 +18,14 @@
}

void client(int port) {
  void *socket = dgram_mod_open_socket(SURVEY);
  void *socket = dgram_mod_open_socket();
  int size;
  void *recvbuf;
  char sendbuf[512];
  long i = 0;
  while (true) {
    sprintf(sendbuf, "%d", i);
    printf("SEND HEART:%s\n", sendbuf);
    dgram_mod_sendto(socket, sendbuf, strlen(sendbuf) + 1, port);
    free(recvbuf);
    sleep(1);
    i++;
  }

 test_socket/dgram_socket_test

Binary files differ

 test_socket/stream_mod_req_rep

Binary files differ