req_rep

wangzhengquan

2020-10-12 575f2e424a17737111786227103a428fb5c20396

req_rep

 src/Makefile

@@ -33,11 +33,16 @@
    PREFIX := /usr/local
endif

all: install
all: build
 

.PHONY: build
build: prebuild $(MYLIBS)
    mkdir -p $(DEST)/lib
    cp $(MYLIBS) $(DEST)/lib
    mkdir -p $(DEST)/include/shmqueue
    cp  ./*.h ./queue/*.h ./socket/*.h  $(DEST)/include/shmqueue
    cp $(ROOT)/lib/* $(DEST)/lib

.PHONY: prebuild
prebuild:

 src/libshm_queue.a

Binary files differ

 src/socket/net_mod_server_socket.c

@@ -88,18 +88,21 @@
  int n;
  net_mod_request_head_t request_head;
  net_mod_response_head_t response_head;
  void *buf, *recv_buf;
  void  *recv_buf;
  static void *buf;
  int recv_size;

  size_t max_buf = 8096;

  buf = malloc(max_buf);
  static size_t max_buf = 1024;
  if(buf == NULL) {
    err_exit(errno, "process_client malloc");
    buf = malloc(max_buf);
    if(buf == NULL) {
      err_exit(errno, "process_client malloc");
    }
  }
  
 
  if ((n = rio_readnb(rio, &request_head, sizeof(net_mod_request_head_t))) !=  sizeof(net_mod_request_head_t))
  {
    free(buf);
    return -1;
  }

@@ -112,15 +115,16 @@
  }  

  if ((n = rio_readnb(rio, buf, request_head.content_length)) != request_head.content_length ) {
    free(buf);
    return -1;
  }

  shmModSocket.sendandrecv(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size);
  response_head.content_length = recv_size;
  Rio_writen(connfd, &response_head, sizeof(response_head));
  Rio_writen(connfd, recv_buf, recv_size);
  free(buf);
  if(request_head.mod == REQ_REP) {
    shmModSocket.sendandrecv(buf, request_head.content_length, request_head.key, &recv_buf, &recv_size);
    response_head.content_length = recv_size;
    Rio_writen(connfd, &response_head, sizeof(response_head));
    Rio_writen(connfd, recv_buf, recv_size);
  }

  return 0;
  
}

 src/socket/net_mod_socket.c

@@ -12,7 +12,7 @@


int NetModSocket::sendandrecv(net_node_t *node_arr, int arrlen, void *send_buf, int send_size, 
  net_mod_recv_msg_t ** resp_arr, int *resp_arr_size) {
  net_mod_recv_msg_t ** recv_arr, int *recv_arr_size) {

  int i, n, clientfd;
  char portstr[32];
@@ -77,8 +77,8 @@
    ret_arr[i].content = recv_buf;
    ret_arr[i].content_length = recv_size;
  }
  *resp_arr = ret_arr;
  *resp_arr_size = i;
  *recv_arr = ret_arr;
  *recv_arr_size = i;

  return i;
     

 src/socket/net_mod_socket.h

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


struct net_mod_request_head_t {
    socket_mod_t mod;
    int key;
    uint32_t mod;
    uint32_t key;
    uint32_t content_length;
};

@@ -31,7 +31,7 @@
  int port;
  int key;
  void *content;
  uint32_t content_length;
  int content_length;
  
};


 src/socket/net_mod_socket_io.c

@@ -1,6 +1,6 @@
#include "net_mod_socket_io.h"
#include "socket_io.h"
 
#include <stdint.h> 

ssize_t rio_readpkgb(rio_t *rp, char *usrbuf, size_t maxlen)
{
@@ -48,4 +48,72 @@
    return -1;
  }
  
}



int is_little_endian() {
  int test_num = 0xff;
  unsigned char* byte_start = (unsigned char*) &test_num;

  if (byte_start[0] == 0xff) {
    return 1;
  }
  return 0;
}

void swap_bytes(void *pv, size_t n)
{
    assert(n > 0);

    char *p = (char *)pv;
    size_t lo, hi;
    for(lo=0, hi=n-1; hi>lo; lo++, hi--)
    {
        char tmp=p[lo];
        p[lo] = p[hi];
        p[hi] = tmp;
    }
}


//! Byte swap unsigned short
uint16_t swap_uint16( uint16_t val ) 
{
    return (val << 8) | (val >> 8 );
}

//! Byte swap short
int16_t swap_int16( int16_t val ) 
{
    return (val << 8) | ((val >> 8) & 0xFF);
}

//! Byte swap unsigned int
uint32_t swap_uint32( uint32_t val )
{
    val = ((val << 8) & 0xFF00FF00 ) | ((val >> 8) & 0xFF00FF ); 
    return (val << 16) | (val >> 16);
}

//! Byte swap int
int32_t swap_int32( int32_t val )
{
    val = ((val << 8) & 0xFF00FF00) | ((val >> 8) & 0xFF00FF ); 
    return (val << 16) | ((val >> 16) & 0xFFFF);
}


int64_t swap_int64( int64_t val )
{
    val = ((val << 8) & 0xFF00FF00FF00FF00ULL ) | ((val >> 8) & 0x00FF00FF00FF00FFULL );
    val = ((val << 16) & 0xFFFF0000FFFF0000ULL ) | ((val >> 16) & 0x0000FFFF0000FFFFULL );
    return (val << 32) | ((val >> 32) & 0xFFFFFFFFULL);
}

uint64_t swap_uint64( uint64_t val )
{
    val = ((val << 8) & 0xFF00FF00FF00FF00ULL ) | ((val >> 8) & 0x00FF00FF00FF00FFULL );
    val = ((val << 16) & 0xFFFF0000FFFF0000ULL ) | ((val >> 16) & 0x0000FFFF0000FFFFULL );
    return (val << 32) | (val >> 32);
}

 src/socket/net_mod_socket_io.h

@@ -6,5 +6,46 @@
#define PKG_SEP "\r\n\r\n"

ssize_t rio_readpkgb(rio_t *rp, char *usrbuf, size_t maxlen);
int is_little_endian();
void swap_bytes(void *pv, size_t n);
uint16_t swap_uint16( uint16_t val ) ;
int16_t swap_int16( int16_t val ) ;
uint32_t swap_uint32( uint32_t val );
int32_t swap_int32( int32_t val );
int64_t swap_int64( int64_t val );
uint64_t swap_uint64( uint64_t val );

#define SWAP_BYTES(x) swap_bytes(&x, sizeof(x))




// #include <endian.h>
// uint16_t htobe16(uint16_t host_16bits);
// uint16_t htole16(uint16_t host_16bits);
// uint16_t be16toh(uint16_t big_endian_16bits);
// uint16_t le16toh(uint16_t little_endian_16bits);

// uint32_t htobe32(uint32_t host_32bits);
// uint32_t htole32(uint32_t host_32bits);
// uint32_t be32toh(uint32_t big_endian_32bits);
// uint32_t le32toh(uint32_t little_endian_32bits);

// uint64_t htobe64(uint64_t host_64bits);
// uint64_t htole64(uint64_t host_64bits);
// uint64_t be64toh(uint64_t big_endian_64bits);
// uint64_t le64toh(uint64_t little_endian_64bits);




// #include <arpa/inet.h>

// uint32_t htonl(uint32_t hostlong);

// uint16_t htons(uint16_t hostshort);

// uint32_t ntohl(uint32_t netlong);

// uint16_t ntohs(uint16_t netshort);
#endif

 test/Makefile

@@ -1,26 +1,39 @@
#
# Makefile for common library.
#
ROOT=..
LDLIBS+=-Wl,-rpath=$(ROOT)/lib:$(ROOT)/build/lib
# 开源工具包路径
LDDIR +=  -L$(ROOT)/lib -L$(ROOT)/build/lib
# 开源工具包
LDLIBS += -lshm_queue -lusgcommon -lpthread

INCLUDE += -I$(ROOT)/build/include

ROOT := ..
PLATFORM=$(shell $(ROOT)/systype.sh)
include $(ROOT)/Make.defines.$(PLATFORM)

 
PROGS = protocle_parse strtok test_set test_vector lambda test test_type
#RPATH += -Wl,-rpath=${ROOT}/lib
# 开源工具包路径
LDDIR += -L${DEST}/lib

build: $(PROGS)
# 开源工具包
LDLIBS += -lshm_queue -lusgcommon -lpthread

# test1: $(LIBCOMMON)
INCLUDES += -I${DEST}/include/shmqueue -I$(ROOT)/include/usgcommon

# 如果包A 引用包B， B 要放在 A 后面

PROGS = ${DEST}/is_little_endian

DEPENDENCES = $(patsubst %, %.d, $(PROGS)) 

#LIBCOMMON=${ROOT}/lib/libusgcommon.a

build:     $(PROGS)


# class
#$(DEST)/kucker : kucker.c


clean:
    rm -f $(TEMPFILES) $(PROGS)
    rm -f $(PROGS) $(DEPENDENCES) $(TEMPFILES)

# $(LIBCOMMON):
#     @(cd $(ROOT)/common && $(MAKE))

-include $(DEPENDENCES)
include $(ROOT)/Make.common.inc




 test_net_socket/Makefile

@@ -20,7 +20,7 @@
#LIBCOMMON=${ROOT}/lib/libusgcommon.a

build:     $(PROGS)

    cp -a net_mod_req_rep.sh ${DEST}

# class
#$(DEST)/kucker : kucker.c

 test_net_socket/net_mod_req_rep.c

@@ -4,21 +4,25 @@
#include "dgram_mod_socket.h"
#include "usg_common.h"

void server() {
    NetModServerSocket *serverSocket  = new NetModServerSocket(5000);
void server(int port) {
    NetModServerSocket *serverSocket  = new NetModServerSocket(port);
    serverSocket->start();
}

void client(){
void client(int port ){
    NetModSocket client;
    char send_buf[MAXLINE];
    net_mod_recv_msg_t *recv_arr;
    int recv_arr_size, i;
    net_node_t node_arr[1] = {
        {"localhost", 5000, 8}
    net_node_t node_arr[] = {
        {"localhost", port, 11},
        {"localhost", port, 12},
        {"localhost", port, 13},
        {"localhost", port, 14}
    };

  while (fgets(send_buf, MAXLINE, stdin) != NULL) {
    client.sendandrecv( node_arr, 1, send_buf, strlen(send_buf), &recv_arr, &recv_arr_size);
    client.sendandrecv( node_arr, 4, send_buf, strlen(send_buf), &recv_arr, &recv_arr_size);
    for(i=0; i<recv_arr_size; i++) {
        printf("host:%s, port: %d, key:%d, content: %s\n", 
            recv_arr[i].host,
@@ -27,23 +31,29 @@
            recv_arr[i].content
        );
    }

    NetModSocket::free_recv_msg_arr(recv_arr, recv_arr_size);
  }
}

int main(int argc, char *argv[]) {
    shm_init(512);

    if (argc < 2) {
     fprintf(stderr, "Usage: %s %s|%s\n", argv[0], "server", "client");
     return 1;
    int port;
  if (argc < 3) {
    fprintf(stderr, "Usage: %s %s|%s <PORT> \n", argv[0],  "server", "client");
    return 1;
  }

  port = atoi(argv[2]);
	 

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

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



 test_socket/test_survey.sh

@@ -19,7 +19,7 @@
    done
}

function close_cleints() {
function close_clients() {
    for (( i=0; i<$PROCESSES; i++ ))
    do
        echo "kill ${pid_arr[$i]}" 
@@ -36,14 +36,14 @@
  "clients")
  start_clients
  sleep 5
    close_cleints
    close_clients
  ;;
  "")
    start_server
    sleep 1
    start_clients
    sleep 5
    close_cleints
    close_clients
  ;;
  *)
  echo "error input"