2017-2018-1 20155304 实验三 实时系统

2017-2018-1 20155304 实验三 实时系统

实验三-并发程序-1

学习使用Linux命令wc(1)

基于Linux Socket程序设计实现wc（1）服务器(端口号是你学号的后6位)和客户端

客户端传一个文本文件给服务器

服务器返加文本文件中的单词数

• 学习wc

统计指定文件中的字节数、字数、行数，并将统计结果显示输出。该命令统计指定文件中的字节数、字数、行数。如果没有给出文件名，则从标准输入读取。wc同时也给出所指定文件的总统计数。

-w统计字数。一个字被定义为由空白、跳格或换行字符分隔的字符串。

常用参数：

• -c：统计字节数
• -l：统计行数
• -m：统计字符数，且不能与-c参数一起使用
• -w：统计字数，一个字被定义为由空白、跳格或换行字符分割的字符串
• -L：打印最长行的长度

代码如下：

客户端：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <string.h>
#include <arpa/inet.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <signal.h>
#include <fcntl.h>
#define MAXLINE 40
#define BUFFERSIZE 4096
#define PORT 13321
ssize_t writen(int fd, const void * vptr, size_t n)
{
    size_t    nleft;
    ssize_t    nwritten;
    const char *    ptr;

    ptr = vptr;
    nleft = n;
    while (nleft > 0) 
    {
        if ((nwritten = write(fd, ptr, nleft)) <= 0) 
        {
            if (nwritten < 0 && errno == EINTR) 
            {
                nwritten = 0;
            }
            else 
            {
                return -1;
            }
        }

        nleft -= nwritten;
        ptr += nwritten;
    }
    return n;
}

int Socket(int domain, int type, int protocol)
{
    int sockfd;
    if ((sockfd = socket(domain, type, protocol)) < 0) 
    {
        fprintf(stderr, "socket error
");
        exit(1);
    }

    return sockfd;
}

int Connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen)
{
    int ret;
    if ((ret = connect(sockfd, addr, addrlen)) < 0) 
    {
        fprintf(stderr, "connect error
");
        exit(1);
    }

    return ret;
}


int Close(int fd)
{
    int ret;
    if ((ret = close(fd)) < 0) 
    {
        fprintf(stderr, "close error
");
        exit(1);
    }
    return ret;
}


int main(int argc, char *argv[])
{
    if (argc != 3) 
    {
        fprintf(stderr, "Usage: ./fileclient <file> <serverIP>
");
        exit(1);
    }

    int sockfd;
    char buff[BUFFERSIZE + 1];
    char filenameheader[BUFFERSIZE + 1];
    struct sockaddr_in servaddr;
    int filefd;    /* file descriptor */
    int count;
    sockfd = Socket(AF_INET, SOCK_STREAM, 0);
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr=INADDR_ANY;
    servaddr.sin_port = htons(PORT);
    
    Connect(sockfd, (struct sockaddr *)&servaddr, sizeof(servaddr));
    printf("已连接服务器
");
    printf("需统计的文件名为: %s........
", argv[1]);
    memcpy(filenameheader, argv[1], strlen(argv[1]));
    filenameheader[strlen(argv[1])] = '
';
    filenameheader[strlen(argv[1]) + 1] = 0;
    writen(sockfd, filenameheader, strlen(filenameheader));

    printf("正上传文件%s至服务器
", argv[1]);

    filefd = open(argv[1], O_RDONLY);
    if (filefd < 0) 
    {
        fprintf(stderr, "can't open the file: %s
", argv[1]);
        exit(1);
    }
    while(count = read(filefd, buff, BUFFERSIZE)) 
    {
        if (count < 0) 
        {
            fprintf(stderr, "filefd read error
");
            exit(1);
        }
        if (writen(sockfd, buff, count) < 0) 
        {
            fprintf(stderr, "writing to sockfd error
");
            exit(1);
        }
    }   
    Close(filefd);
    Close(sockfd);
    printf("文件%s已上传至服务器!
", argv[1]);
    return 0;
}

服务器：

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <dirent.h>
#include <string.h>
#include <arpa/inet.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <signal.h>
#include <fcntl.h>
#define LEN 4096  
#define MAXLINE 40
#define BUFFERSIZE 4096
#define PORT 13321

ssize_t writen(int fd, const void * vptr, size_t n)
{
    size_t nleft;
    ssize_t nwritten;
    const char * ptr;

    ptr = vptr;
    nleft = n;
    while ( nleft > 0)
     {
        if ((nwritten = write(fd, ptr, nleft)) <= 0)
        {
            if (nwritten < 0 && errno == EINTR) 
            {
                nwritten = 0;
            }
            else
            {
                return -1;
            }
        }
        nleft -= nwritten;
        ptr += nwritten; 
    }
    return n;
}


ssize_t readline(int fd, void * vptr, size_t maxlen)
{
    ssize_t    n, rc;
    char    c, *ptr;

    ptr = vptr;
    for (n = 1; n < maxlen; n++)
    {
        again:
        if ((rc = read(fd, &c, 1)) == 1)
        {
            *ptr++ = c;
            if (c == '
')
            {
                break;
            }
        }
        else if (rc == 0)
        {   
            *ptr = 0;
            return (n - 1);
        }
        else
        {
            if (errno == EINTR)
            {
                goto again;
            }
            return (-1);
        }
    }
    *ptr = 0;
    return (n);
}


int Socket(int domain, int type, int protocol)
{
    int sockfd;
    if ((sockfd = socket(domain, type, protocol)) < 0)
    {
    fprintf(stderr, "socket error
");
    exit(1);
    }

    return sockfd;
}

int Accept(int sockfd, struct sockaddr * addr, socklen_t * addrlen)
{
    int ret;
    if ((ret = accept(sockfd, addr, addrlen)) < 0)
    {
        fprintf(stderr, "accept error
");
        exit(1);
    }
    return ret;
}

int Bind(int sockfd, const struct sockaddr *addr, socklen_t addrlen)
{
    int ret;
    if ((ret = bind(sockfd, addr, addrlen)) < 0)
    {
        fprintf(stderr, "bind error
");
        exit(1);
    }
    return ret;
}

int Listen(int sockfd, int backlog)
{
    int ret;
    if ((ret = listen(sockfd, backlog)) < 0)
    {
        fprintf(stderr, "listen error
");
        exit(1);
    }
    return ret;
}

int Close(int fd)
{
    int ret;
    if ((ret = close(fd)) < 0)
    {
        fprintf(stderr, "close error
");
        exit(1);
    }
    return ret;
}

int main(void)
{
    int listenfd, connfd;
    char    buff[BUFFERSIZE + 1];
    char    filename[BUFFERSIZE + 1];
    char    cd[BUFFERSIZE+1];
    char    choose[10];
    struct sockaddr_in  servaddr, cliaddr;
    int cliaddrlen;
    int filefd;   
    int count;
    DIR *dir;
    struct dirent   *ptr;
    listenfd = Socket(AF_INET, SOCK_STREAM, 0);
    bzero(&servaddr, sizeof(servaddr));
    servaddr.sin_family = AF_INET;
    servaddr.sin_addr.s_addr = INADDR_ANY;
    servaddr.sin_port = htons(PORT);
    Bind(listenfd, (struct sockaddr *)&servaddr, sizeof(servaddr));
    Listen(listenfd, 5);
    while(1)
    {
        printf("开始监听
");
        cliaddrlen = sizeof(cliaddr);
        connfd = Accept(listenfd, (struct sockaddr *)&cliaddr, &cliaddrlen);
        if (readline(connfd, buff, BUFFERSIZE) < 0)
        {
            fprintf(stderr, "readline error
");
            exit(1);
        }
        buff[strlen(buff) - 1] = 0;    /* change '
' to NUL */
        memcpy(filename, buff, BUFFERSIZE + 1);
        printf("统计的文件名: %s
", buff);
        printf("Input the direct you want to store %s:
", buff);
        scanf("%s", cd);
        if(chdir(cd) < 0)
        {
            fprintf(stderr, "direct error
");
            exit(1);
        }
        dir = opendir(cd);
        while((ptr = readdir(dir)) != NULL)
        {
            if(strcmp(buff, ptr->d_name) == 0)
            {
                printf("已存在文件：%s
", buff);
                printf("若想重命名请输入yes,否则请输入no
");    
                scanf("%s", choose);
                if(strcmp(choose, "yes") == 0)
                {           
                    printf("重命名为:	");
                    scanf("%s", buff);
                }
                else
                {
                    printf("已存在该文件, 已退出！
");
                    exit (1);
                }
                
            }
        }

        filefd = open(buff, O_WRONLY | O_CREAT);
        if (filefd < 0)
        {
            fprintf(stderr, "can't open the file: %s
", buff);
            exit(1);
        }

        while(count = read(connfd, buff, BUFFERSIZE))
        {
            if (count < 0)
            {
                fprintf(stderr, "connfd read error
");
                exit(1);
            }
            if (writen(filefd, buff, count) < 0) 
            {
                fprintf(stderr, "writing to filefd error
");
                exit(1);
            }
        }
        closedir(dir);
        Close(filefd);
        Close(connfd);
        printf("file %s received!
", filename);

        FILE *fp;  // 指向文件的指针
        char buffer[1003];  //缓冲区，存储读取到的每行的内容
        int bufferLen;  // 缓冲区中实际存储的内容的长度
        int i;  // 当前读到缓冲区的第i个字符
        char c;  // 读取到的字符
        int isLastBlank = 0;  // 上个字符是否是空格
        int charNum = 0;  // 当前行的字符数
        int wordNum = 0; // 当前行的单词数
        int totalNum[1] = {0};

        if( (fp = fopen("/home/rafel/shiyan3/1/save/text1.txt","r+")) == NULL ){
            perror(filename);
            exit(1);
        }
        while(fgets(buffer, 1003, fp) != NULL){
        bufferLen = strlen(buffer);
        // 遍历缓冲区的内容
        for(i=0; i<bufferLen; i++){
            c = buffer[i];
            if( c==' ' || c=='	'){  // 遇到空格
                !isLastBlank && wordNum++;  // 如果上个字符不是空格，那么单词数加1
                isLastBlank = 1;
            }else if(c!='
'&&c!='
'){  // 忽略换行符
                charNum++;  // 如果既不是换行符也不是空格，字符数加1
                isLastBlank = 0;
            }
        }
        !isLastBlank && wordNum++;  // 如果最后一个字符不是空格，那么单词数加1
        isLastBlank = 1;  // 每次换行重置为1
        // 一行结束，计算总单词数
        totalNum[0] += wordNum;  // 总单词数
        // 置零，重新统计下一行
        charNum = 0;
        wordNum = 0;
    }
    printf("Total: %d words
", totalNum[0]);
    }
    Close(listenfd);
    return 0;
}

• 完成截图：
  

实验三-并发程序-2

使用多线程实现wc服务器并使用同步互斥机制保证计数正确

对比单线程版本的性能，并分析原因

多线程实现wc服务器时，会出现多个客户端同时像服务器传送文件的情况，所以仅仅在服务器端创建一个recvfile.txt用于接收文件是远远不够的，需要根据发送的不同文件名创建新的接收文件。

• 代码如下:

服务器：

#include "csapp.h"
#include<stdio.h>
#include<stdlib.h>
static int byte_cnt;  /* byte counter */
static sem_t mutex;
#define NTHREADS  4
#define SBUFSIZE  16
typedef struct {
    int *buf;          /* Buffer array */        
    int n;             /* Maximum number of slots */
    int front;         /* buf[(front+1)%n] is first item */
    int rear;          /* buf[rear%n] is last item */
    sem_t mutex;       /* Protects accesses to buf */
    sem_t slots;       /* Counts available slots */
    sem_t items;       /* Counts available items */
} sbuf_t;
void echo_cnt(int connfd);
void *thread(void *vargp);
int wc(char *name)
{
 char ch;
 FILE *fp;
 long count=0;
 char s[21];
  
  
 if ((fp=fopen(name,"r+"))==NULL)
 {
  fprintf(stderr,"不能打开文件
");
  exit(EXIT_FAILURE);
 }
while(fscanf(fp,"%s",s)!=EOF)
        count++;
  
 fclose(fp);
 printf("File %s has %ld characters
",name,count);
 return 0;
}
 
sbuf_t sbuf; /* shared buffer of connected descriptors */
 
int main(int argc, char **argv)
{
    int i, listenfd, connfd, port, clientlen=sizeof(struct sockaddr_in);
    struct sockaddr_in clientaddr;
    pthread_t tid;
 
    if (argc != 2) {
    fprintf(stderr, "usage: %s <port>
", argv[0]);
    exit(0);
    }
    port = atoi(argv[1]);
    sbuf_init(&sbuf, SBUFSIZE);
    listenfd = Open_listenfd(port);
 
    for (i = 0; i < NTHREADS; i++)  /* Create worker threads */
    Pthread_create(&tid, NULL, thread, NULL);
 
    while (1) {
    connfd = Accept(listenfd, (SA *) &clientaddr, &clientlen);
    sbuf_insert(&sbuf, connfd); /* Insert connfd in buffer */
    }
}
 
static void init_echo_cnt(void)
{
    Sem_init(&mutex, 0, 1);
    byte_cnt = 0;
}
 
void echo_cnt(int connfd)
{
    int n,x;
    long int count;
    char buf[MAXLINE];
    char name[MAXLINE]
    rio_t rio;
    static pthread_once_t once = PTHREAD_ONCE_INIT;
 
    Pthread_once(&once, init_echo_cnt);
    Rio_readinitb(&rio, connfd);
    while((n = Rio_readlineb(&rio, buf, MAXLINE)) != 0) {
    P(&mutex);
    byte_cnt += n;
        /*x = sizeof(buf);
        buf[x] = 0;
        count = wc(buf);*/
    printf("thread %d received %d (%d total) bytes on fd %d
",
           (int) pthread_self(), n, byte_cnt, connfd);
        //name = buf;
    V(&mutex);
        //sprint(buf,"%s:%ld characters".count);
 
    Rio_writen(connfd, buf, n);
    }
}
 
void sbuf_init(sbuf_t *sp, int n)
{
    sp->buf = Calloc(n, sizeof(int));
    sp->n = n;                       /* Buffer holds max of n items */
    sp->front = sp->rear = 0;        /* Empty buffer iff front == rear */
    Sem_init(&sp->mutex, 0, 1);      /* Binary semaphore for locking */
    Sem_init(&sp->slots, 0, n);      /* Initially, buf has n empty slots */
    Sem_init(&sp->items, 0, 0);      /* Initially, buf has zero data items */
}
/* $end sbuf_init */
 
/* Clean up buffer sp */
/* $begin sbuf_deinit */
void sbuf_deinit(sbuf_t *sp)
{
    Free(sp->buf);
}
/* $end sbuf_deinit */
 
/* Insert item onto the rear of shared buffer sp */
/* $begin sbuf_insert */
void sbuf_insert(sbuf_t *sp, int item)
{
    P(&sp->slots);                          /* Wait for available slot */
    P(&sp->mutex);                          /* Lock the buffer */
    sp->buf[(++sp->rear)%(sp->n)] = item;   /* Insert the item */
    V(&sp->mutex);                          /* Unlock the buffer */
    V(&sp->items);                          /* Announce available item */
}
/* $end sbuf_insert */
 
/* Remove and return the first item from buffer sp */
/* $begin sbuf_remove */
int sbuf_remove(sbuf_t *sp)
{
    int item;
    P(&sp->items);                          /* Wait for available item */
    P(&sp->mutex);                          /* Lock the buffer */
    item = sp->buf[(++sp->front)%(sp->n)];  /* Remove the item */
    V(&sp->mutex);                          /* Unlock the buffer */
    V(&sp->slots);                          /* Announce available slot */
    return item;
}
 
void *thread(void *vargp)
{ 
    Pthread_detach(pthread_self());
    while (1) {
    int connfd = sbuf_remove(&sbuf); /* Remove connfd from buffer */
    echo_cnt(connfd);                /* Service client */
    Close(connfd);
    }
}

客户端：

#include "csapp.h"
#include<stdio.h>
#include<stdlib.h>
int wc(char *name)
{
 char ch;
 FILE *fp;
 long count=0;
 char s[21];
  
  
 if ((fp=fopen("test1.txt","r+"))==NULL)
 {
  fprintf(stderr,"不能打开文件%s
",name);
  exit(EXIT_FAILURE);
 }
while(fscanf(fp,"%s",s)!=EOF)
        count++;
  
 fclose(fp);
 printf("File %s has %ld characters
",name,count);
 return 0;
}
int main(int argc, char **argv)
{
    int clientfd, port,n,count;
    char *host, buf[MAXLINE];
    rio_t rio;
 
    if (argc != 3) {
    fprintf(stderr, "usage: %s <host> <port>
", argv[0]);
    exit(0);
    }
    host = argv[1];
    port = atoi(argv[2]);
 
    clientfd = Open_clientfd(host, port);
    Rio_readinitb(&rio, clientfd);
 
    while (Fgets(buf, MAXLINE, stdin) != NULL) {
        if((num=recv(sockfd,buf,MAXDATASIZE,0))==-1)
    {
        printf("recv() error
");
        exit(1);
    }
    buf[num-1]='';
     
         
    Rio_writen(clientfd, buf, strlen(buf));
    Rio_readlineb(&rio, buf, MAXLINE);
    Fputs(buf, stdout);
    }
    Close(clientfd);
    exit(0);
}
/* $end echoclientmain */
/* $begin echoclientmain */
//#include "csapp.h"
/*int main(int argc, char **argv)
{
    int clientfd, port;
    char *host, buf[MAXLINE];
    char *name;
    rio_t rio;
    FILE *fp;
    if (argc != 4) {
    fprintf(stderr, "usage: %s <host> <port> <filename>
", argv[0]);
    exit(0);
    }
    host = argv[1];
    port = atoi(argv[2]);
    name = argv[3];
    clientfd = Open_clientfd(host, port);
    Rio_readinitb(&rio, clientfd);
 
    fp=fopen(name,"r+");
    while (Fgets(buf, MAXLINE,fp) != NULL) {
    Rio_writen(clientfd, buf, strlen(buf));
    Rio_readlineb(&rio, buf, MAXLINE);
    Fputs(buf, stdout);
    }
    Close(clientfd);
    exit(0);
}

• 完成截图：

实验三-并发程序-3

交叉编译多线程版本服务器并部署到实验箱中

PC机作客户端测试wc服务器

未完成…………

新学到的知识点

• 进一步的了解和学习了单线程和多线程
• 学会进一步使用pthread
• 更深入学习了服务器与客户端
• 掌握使用man命令查询