update

wangzhengquan

2020-07-23 b6043642f60ef23a7a100418cd4fec1251a98ad9

update

 .gitignore

@@ -38,3 +38,7 @@
*.tar
*.tar.gz
.vscode/

build/
*.tmp
*.txt

 Makefile

@@ -1,5 +1,5 @@
DIRS = src test_socket demo

TAR_NAME = shm_queue.tar.gz
all:
    for i in $(DIRS); do \
        (cd $$i && echo "making $$i" && $(MAKE) ) || exit 1; \
@@ -12,11 +12,6 @@
    rm -rf build
    ipcrm -a

ipcrm:
    -ipcrm -a
    -ipcs

#     -ipcrm -M 0x1234
#     -ipcrm -S 145
#     -ipcrm -S 146
#     -ipcrm -S 8899
tar:
    rm -f $(TAR_NAME)
    git archive --format tar.gz --output $(TAR_NAME) master

 build/include/array_lock_free_queue.h

File was deleted

 build/include/array_lock_free_queue2.h

File was deleted

 build/include/dgram_mod_socket.h

File was deleted

 build/include/hashtable.h

File was deleted

 build/include/linked_lock_free_queue.h

File was deleted

 build/include/lock_free_queue.h

File was deleted

 build/include/logger_factory.h

File was deleted

 build/include/mem_pool.h

File was deleted

 build/include/mm.h

File was deleted

 build/include/mod_socket.h

File was deleted

 build/include/sem_util.h

File was deleted

 build/include/shm_allocator.h

File was deleted

 build/include/shm_mm.h

File was deleted

 build/include/shm_queue.h

File was deleted

 build/include/shm_queue_wrapper.h

File was deleted

 build/include/shm_socket.h

File was deleted

 build/lib/libshm_queue.a

Binary files differ

 demo/dgram_mod_req_rep.c

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

void server(int port) {
  void *socket = dgram_mod_open_socket(REQ_REP);
  dgram_mod_bind(socket, port);
  int size;
  void *recvbuf;
  char sendbuf[512];
  int rv;
  int remote_port;
  while ( (rv = dgram_mod_recvfrom(socket, &recvbuf, &size, &remote_port) ) == 0) {
    sprintf(sendbuf, "SERVER RECEIVED: %s", recvbuf);
    puts(sendbuf);
    dgram_mod_sendto(socket, sendbuf, strlen(sendbuf) + 1, remote_port);
    free(recvbuf);
  }
  dgram_mod_close_socket(socket);
}

void client(int port) {
  void *socket = dgram_mod_open_socket(REQ_REP);
  int size;
  void *recvbuf;
  char sendbuf[512];
  while (true) {
    printf("request: ");
    scanf("%s", sendbuf);
    dgram_mod_sendandrecv(socket, sendbuf, strlen(sendbuf) + 1, port, &recvbuf, &size);
    printf("reply: %s\n", (char *)recvbuf);
    free(recvbuf);
  }
  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;
}

 demo/pub_sub

Binary files differ

 demo/queue

Binary files differ

 demo/req_rep

Binary files differ

 src/libshm_queue.a

Binary files differ

 src/queue/mm.c

@@ -122,6 +122,7 @@
    SemUtil::inc(mutex);
    return aptr;
  } else {
    SemUtil::inc(mutex);
    err_msg(0, "mm_malloc : out of memery\n");
    return NULL;
  }
@@ -143,9 +144,8 @@
   *}
   */

  
  size_t size = GET_SIZE(HDRP(ptr));
  SemUtil::dec(mutex);
  size_t size = GET_SIZE(HDRP(ptr));
  PUT(HDRP(ptr), PACK(size, 0));
  PUT(FTRP(ptr), PACK(size, 0));
  coalesce(ptr);

 src/socket/dgram_mod_socket.c

@@ -10,19 +10,17 @@
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;
  // pthread_t recv_thread;
    // std::map<int, LockFreeQueue<shm_msg_t, DM_Allocator> * > *recv_queue_map;
} dgram_mod_socket_t;


void *dgram_mod_open_socket(int mod) {
    dgram_mod_socket_t * socket = (dgram_mod_socket_t *)calloc(1, sizeof(dgram_mod_socket_t));
    socket->mod = (socket_mod_t)mod;
    socket->recv_thread = 0;
    socket->recv_queue_map = NULL;
    // socket->recv_thread = 0;
    // socket->recv_queue_map = NULL;
    socket->shm_socket = shm_open_socket(SHM_SOCKET_DGRAM);

    printf("socket->shm_socket = %p \n" , socket->shm_socket);
    return (void *)socket;
}

@@ -30,18 +28,18 @@
int dgram_mod_close_socket(void * _socket) {
    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
    shm_close_socket(socket->shm_socket);
    if(socket->recv_queue_map != NULL) {
        for(auto iter = socket->recv_queue_map->begin(); iter != socket->recv_queue_map->end(); iter++) {
            delete iter->second;
            socket->recv_queue_map->erase(iter);
    // if(socket->recv_queue_map != NULL) {
    //     for(auto iter = socket->recv_queue_map->begin(); iter != socket->recv_queue_map->end(); iter++) {
    //         delete iter->second;
    //         socket->recv_queue_map->erase(iter);
            
        }
        delete socket->recv_queue_map;
    }
    //     }
    //     delete socket->recv_queue_map;
    // }


    if(socket->recv_thread != 0)
        pthread_cancel(socket->recv_thread);
    // if(socket->recv_thread != 0)
    //     pthread_cancel(socket->recv_thread);
    free(_socket);
}

@@ -53,7 +51,6 @@

int dgram_mod_sendto(void *_socket, const void *buf, const int size, const int port) {
    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;

    return shm_sendto(socket->shm_socket, buf, size, port);

}
@@ -61,79 +58,90 @@
int dgram_mod_recvfrom(void *_socket, void **buf, int *size, int *port) {

    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
    if(socket->mod == REQ_REP && socket->recv_thread != 0) {
        err_exit(0, "you have used sendandrecv method os you can not use recvfrom method any more. these two method can not be used at the same time.");
        return -1;
    }
    // if(socket->mod == REQ_REP && socket->recv_thread != 0) {
    //     err_exit(0, "you have used sendandrecv method os you can not use recvfrom method any more. these two method can not be used at the same time.");
    //     return -1;
    // }
    return shm_recvfrom(socket->shm_socket, buf, size, port);
}

void *_dgram_mod_run_recv(void * _socket) {
    pthread_detach(pthread_self());
    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
    void *buf;
    int size;
    int port;
    shm_msg_t msg;
    LockFreeQueue<shm_msg_t, DM_Allocator> *queue;
    std::map<int, LockFreeQueue<shm_msg_t, DM_Allocator> * >::iterator iter;
// printf("==============_dgram_mod_run_recv recv before\n");
    while(shm_recvfrom(socket->shm_socket, &buf, &size, &port) == 0) {
        if( (iter = socket->recv_queue_map->find(port) ) != socket->recv_queue_map->end()) {
            queue = iter->second;
        } else {
            queue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
            socket->recv_queue_map->insert({port, queue});
        }
// void *_dgram_mod_run_recv(void * _socket) {
//     pthread_detach(pthread_self());
//     dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
//     void *buf;
//     int size;
//     int port;
//     shm_msg_t msg;
//     LockFreeQueue<shm_msg_t, DM_Allocator> *queue;
//     std::map<int, LockFreeQueue<shm_msg_t, DM_Allocator> * >::iterator iter;
// // printf("==============_dgram_mod_run_recv recv before\n");
//     while(shm_recvfrom(socket->shm_socket, &buf, &size, &port) == 0) {
//         if( (iter = socket->recv_queue_map->find(port) ) != socket->recv_queue_map->end()) {
//             queue = iter->second;
//         } else {
//             queue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
//             socket->recv_queue_map->insert({port, queue});
//         }

        msg.buf = buf;
        msg.size = size;
        msg.port = port;
// printf("==============_dgram_mod_run_recv push before\n");
        queue->push(msg);
// printf("==============_dgram_mod_run_recv push after\n");
//         msg.buf = buf;
//         msg.size = size;
//         msg.port = port;
// // printf("==============_dgram_mod_run_recv push before\n");
//         queue->push(msg);
// // printf("==============_dgram_mod_run_recv push after\n");
     
    }
    return NULL;
//     }
//     return NULL;



}
// }

int dgram_mod_sendandrecv(void * _socket, const void *send_buf, const int send_size, const int port, void **recv_buf, int *recv_size) {


int dgram_mod_sendandrecv(void * _socket, const void *send_buf, const int send_size, const int send_port, void **recv_buf, int *recv_size) {
    
    dgram_mod_socket_t * socket = (dgram_mod_socket_t *) _socket;
    if(socket->mod != REQ_REP) {
        err_exit(0, "you can't use this method other than REQ_REP mod!");
    }
    if(socket->recv_queue_map == NULL) {
        socket->recv_queue_map = new std::map<int, LockFreeQueue<shm_msg_t, DM_Allocator> * >;
    }
    // if(socket->recv_queue_map == NULL) {
    //     socket->recv_queue_map = new std::map<int, LockFreeQueue<shm_msg_t, DM_Allocator> * >;
    // }

    std::map<int, LockFreeQueue<shm_msg_t, DM_Allocator> * >::iterator iter;
    LockFreeQueue<shm_msg_t, DM_Allocator> *queue;
    if( (iter = socket->recv_queue_map->find(port) ) != socket->recv_queue_map->end()) {
        queue = iter->second;
    } else {
        queue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
        socket->recv_queue_map->insert({port, queue});
    }
    // std::map<int, LockFreeQueue<shm_msg_t, DM_Allocator> * >::iterator iter;
    // LockFreeQueue<shm_msg_t, DM_Allocator> *queue;
    // if( (iter = socket->recv_queue_map->find(port) ) != socket->recv_queue_map->end()) {
    //     queue = iter->second;
    // } else {
    //     queue = new LockFreeQueue<shm_msg_t, DM_Allocator>(16);
    //     socket->recv_queue_map->insert({port, queue});
    // }

    if (socket->recv_thread == 0) {
    // if (socket->recv_thread == 0) {
        
        pthread_create(&(socket->recv_thread ), NULL, _dgram_mod_run_recv , _socket);
    //     pthread_create(&(socket->recv_thread ), NULL, _dgram_mod_run_recv , _socket);
        
    }
    // }

    shm_sendto(socket->shm_socket, send_buf, send_size, port);	
    shm_msg_t msg;
// printf("==============dgram_mod_sendandrecv pop before\n");
    if(queue->pop(msg)) {
        *recv_buf = msg.buf;
        *recv_size = msg.size;
// printf("==============dgram_mod_sendandrecv pop after\n");
        return 0;
    // shm_msg_t msg;
    // if(queue->pop(msg)) {
    //     *recv_buf = msg.buf;
    //     *recv_size = msg.size;
    //     return 0;
    // }

    int recv_port;
    int rv;

    shm_socket_t *shm_socket = shm_open_socket(SHM_SOCKET_DGRAM);
    if (shm_sendto(shm_socket, send_buf, send_size, send_port) == 0) {
        rv = shm_recvfrom(shm_socket, recv_buf, recv_size, &recv_port);
        shm_close_socket(shm_socket);
        return rv;
    }
	
	
    return -1;

}

 src/socket/shm_socket.c

@@ -287,7 +287,7 @@
    }

    shm_msg_t src;
printf("shm_recvfrom pop before");
// printf("shm_recvfrom pop before");
    if (socket->queue->pop(src)) {
        void * _buf = malloc(src.size);
        memcpy(_buf, src.buf, src.size);
@@ -295,7 +295,7 @@
        *size = src.size;
        *port = src.port;
        mm_free(src.buf);
printf("shm_recvfrom pop after");
// printf("shm_recvfrom pop after");
        return 0;
    } else {
        return -1;
@@ -327,7 +327,7 @@
    if( iter !=  socket->clientSocketMap->end() ) {
        socket->clientSocketMap->erase(iter);
    }
    //free((void *)client_socket);
    free((void *)client_socket);

}


 test_socket/dgram_mod_req_rep

Binary files differ

 test_socket/dgram_mod_req_rep.c

@@ -1,20 +1,58 @@
#include "dgram_mod_socket.h"
#include "shm_mm.h"
#include "usg_common.h"
#include "lock_free_queue.h"

#define WORKERS 4

typedef struct task_t {
    void *buf;
    int size;
    int port;

} task_t;


typedef struct Targ {
    int port;
    int id;

}Targ;

LockFreeQueue<task_t, DM_Allocator> task_queue(100);


void *worker(void *socket) {
    pthread_detach(pthread_self());
    char sendbuf[512];
    task_t task;
    while(true) {
        task_queue.pop(task);
        sprintf(sendbuf, "SERVER RECEIVED: %s", task.buf);
   // puts(sendbuf);
    dgram_mod_sendto(socket, sendbuf, strlen(sendbuf) + 1, task.port);
    free(task.buf);
    }
    return NULL;
}

void initThreadPool(void *socket) {

  pthread_t tid;
  for (int i = 0; i < WORKERS; i++)
    pthread_create(&tid, NULL, worker, socket);
}

void server(int port) {
  void *socket = dgram_mod_open_socket(REQ_REP);
  dgram_mod_bind(socket, port);
  int size;
  void *recvbuf;
  char sendbuf[512];
     initThreadPool(socket);

  int rv;
  int remote_port;
  while ( (rv = dgram_mod_recvfrom(socket, &recvbuf, &size, &remote_port) ) == 0) {
    sprintf(sendbuf, "SERVER RECEIVED: %s", recvbuf);
    puts(sendbuf);
    dgram_mod_sendto(socket, sendbuf, strlen(sendbuf) + 1, remote_port);
    free(recvbuf);
  task_t task;
  while ( (rv = dgram_mod_recvfrom(socket, &task.buf, &task.size, &task.port) ) == 0) {
      task_queue.push(task);
   
  }
  dgram_mod_close_socket(socket);
}
@@ -34,6 +72,67 @@
  dgram_mod_close_socket(socket);
}



void *runclient(void *arg) {
  Targ *targ = (Targ *)arg;
  int port = targ->port;
  char sendbuf[512];
  int scale = 100000;
  int i;
  void *socket = dgram_mod_open_socket(REQ_REP);
	
    char filename[512];
    sprintf(filename, "test%d.txt", targ->id);
    FILE *fp = NULL;
    fp = fopen(filename, "w+");

    int recvsize;
    void *recvbuf;
  for (i = 0; i < scale; i++) {
    sprintf(sendbuf, "thread(%d) %d", targ->id, i);
    fprintf(fp, "requst:%s\n", sendbuf);
    dgram_mod_sendandrecv(socket, sendbuf, strlen(sendbuf) + 1, port, &recvbuf, &recvsize);
        fprintf(fp, "reply: %s\n", (char *)recvbuf);
        free(recvbuf);
  }
  fclose(fp);
  dgram_mod_close_socket(socket);
  return (void *)i;
}

void startClients(int port) {

  int status, i = 0, processors = 4;
  void *res[processors];
  Targ *targs = (Targ *)calloc(processors, sizeof(Targ));
  pthread_t tids[processors];
  char sendbuf[512];

  struct timeval start;
  gettimeofday(&start, NULL);
  for (i = 0; i < processors; i++) {
    targs[i].port = port;
    targs[i].id = i;
    pthread_create(&tids[i], NULL, runclient, (void *)&targs[i]);
  }

  for (i = 0; i < processors; i++) {
    if (pthread_join(tids[i], &res[i]) != 0) {
      perror("multyThreadClient pthread_join");
    } else {
      fprintf(stderr, "client(%d) 写入 %ld 条数据\n", i, (long)res[i]);
    }
  }

  struct timeval end;
  gettimeofday(&end, NULL);

  double difftime = end.tv_sec * 1000000 + end.tv_usec - (start.tv_sec * 1000000 + start.tv_usec);
  long diffsec = (long) (difftime/1000000);
  long diffmsec = difftime - diffsec;
  printf("cost: %ld sec: %ld msc\n", diffsec, diffmsec);
}
 

int main(int argc, char *argv[]) {
@@ -51,7 +150,7 @@
  }

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

  
  return 0;

 test_socket/dgram_mod_survey.c

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

void server(int port) {
  void *socket = dgram_mod_open_socket(SURVEY);
  dgram_mod_bind(socket, port);
  int size;
  void *recvbuf;
  char sendbuf[512];
  int rv;
  int remote_port;
  while ( (rv = dgram_mod_recvfrom(socket, &recvbuf, &size, &remote_port) ) == 0) {
    printf( "收到来自%d注册信息: %s", remote_port, recvbuf);
    free(recvbuf);
  }
  dgram_mod_close_socket(socket);
}

void client(int port) {
  void *socket = dgram_mod_open_socket(SURVEY);
  int size;
  void *recvbuf;
  char sendbuf[512];
  while (true) {
    printf("request: ");
    scanf("%s", sendbuf);
    dgram_mod_send(socket, sendbuf, strlen(sendbuf) + 1, port, &recvbuf, &size);
    printf("reply: %s\n", (char *)recvbuf);
    free(recvbuf);
  }
  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_socket_test

Binary files differ