一个并行server程序

andy007

以下实现了典型的server程序，它以并行的方式来处理文本请求。因为处理请求时大部分时间都消耗在了创建进程，所以下面这个例程像apache那样事先创建几个进程，当连接已经建立以后，客户端的请求由已经创建的那些进程来处理。
希望对大家有所帮助


[code:1]#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/ipc.h>
#include <sys/shm.h>
#include <sys/time.h>
#include <netinet/in.h>
#include <netdb.h>
#include <sys/socket.h>
#include <sys/signal.h>
#include <sys/types.h>

#define PORT 9876
#define FALSE 0
#define TRUE 1
#define MAX 5

int Running = TRUE;

int main(void)
{
        /*
        **  prepool_server.c - Concurrent Server: Subtask Pool
        */
        struct area {        /* flag for every subtask*/
                int flag[MAX];   /* 0:idle, 1:good  -:exit */
        } area, *aptr;
        int flag = (IPC_CREAT | IPC_EXCL | 0660);
        int size = sizeof(struct area);
        key_t key = 0x5a5a5a5a;
        int shmid;  /*shared memory area id */

        void control_C();  /* ctrl-c signal handler */
        struct sockaddr_in sock;
        int socklen = sizeof(sock); /* struct len for accept() */
        FILE *fp;
        fd_set fds;
        int acc;
        int cli;
        int line;
        int opt = 1;
        int ident;
        int pids[MAX];
        char file[32];
        char buf[80];
        int status;
       
        /* initialize ~C handler */
        signal(SIGINT, control_C);

        /* create a (server) socket to accept a client */
        acc = socket(AF_INET, SOCK_STREAM, 0);

        /* bind socket to port */
        sock.sin_family = AF_INET;
        sock.sin_port = htons(PORT);
        sock.sin_addr.s_addr = htonl(INADDR_ANY);
        bind(acc, (struct sockaddr *)&sock, socklen);
        setsockopt(acc, SOL_SOCKET, SO_REUSEADDR, (char *)&opt, sizeof(opt));
        listen(acc, 5);

        /* get the shared memory area */
        shmid = shmget(key, size, flag);
        /* attach subtask's shared memory array */
        [color=red]aptr = (struct area *) shmat(shmid, 0, 0);[/color]

        /* pre-establish subtask pool */
        for (ident = 0; ident < MAX; ident++) {
                aptr->flag[ident] = 0; /* set flags idle */

                if ((pids[ident] = fork()) == 0) {        /* subtask */
                        /* attach parent's shared memory array */
                        aptr = (struct area *) shmat(shmid, 0, 0);
                        /* notice: shmid is still set correctly */

                        /* nullify ~C handler */
                        signal(SIGINT, SIG_DFL);

                        /*
                        **  Pool Memory array for a non-zero flag
                        */
                        while (1) {
                                if (aptr->flag[ident] = 0) {
                                        usleep(1);  /* release processor */
                                        continue;   /* stay in pool loop */
                                } else if (aptr->flag[ident] < 0) {
                                        exit(0);
                                } [color=red][color=#444444]else [/color][/color]{
                                        cli = accept(acc, (struct sockaddr *)&sock, &socklen);
                                        /* receive an incoming query */
                                        recv(cli, buf, sizeof(buf), 0);
                                        /* open the requested text file */
                                        sscanf(buf, "%s %d", file, &line);
                                        fp = fopen(file, "r");
                                        while (line--) {
                                                fgets(buf, sizeof(buf), fp);
                                        }
                                        fclose(fp);

                                        /* send back a response */
                                        buf[strlen(buf) - 1] = '\0';
                                        send(cli, buf, strlen(buf) + 1, 0);
                                        close(cli);

                                        /* set to available */
                                        aptr->flag[ident] = 0;
                                }
                        }
                }
        }
       
        /*
        **  Parent task passes incoming connection to a subtask
        */
        while (Running) {  /* set FALSE in control_C signal handler */
                FD_ZERO(&fds);
                FD_SET(acc,&fds);
                /* block until client is ready for connect */
                if (select(acc + 1, &fds, NULL, NULL, NULL) < 0) {
                        break;
                }
                /*
                **  Assign incoming query to first available subtask
                */
                for (ident = 0; ident < MAX; ident++) {
                        if (aptr->flag[ident] == 0) {
                                aptr->flag[ident] = 1;
                                break;
                        }
                }
        }

        /* wait for each subtask exit */
        for (ident = 0; ident < MAX; ident++) {
                aptr->flag[ident] = -1;
                waitpid(pids[ident], &status, 0);
        }

        /* remove unused shared memory area and exit */
        shmctl(shmid, IPC_RMID, (struct shmid_ds *) 0);
       
        return 0;
}

/*
** ~C signal handler
*/
void control_C()
{
        Running = FALSE; /* reset flag to allow exit */
        exit(0);
}

[/code:1]

sagaeon

个人认为较详细的文档比纯代码更有帮助 要不楼主整理个文档吧

andy007

没太多的时间去整理出一份文档，我还有很多的事情要做。十分抱歉
如果是对于有一点网络编程基础和linux系统基础的人，以上代码很容易看懂的。况且注释写的也算详细
里面牵涉到tcp通讯以及内存共享存储区操作。

sagaeon

[quote:5b5c1afae7="andy007"]没太多的时间去整理出一份文档，我还有很多的事情要做。十分抱歉
如果是对于有一点网络编程基础和linux系统基础的人，以上代码很容易看懂的。况且注释写的也算详细
里面牵涉到tcp通讯以及内存共享存储区操作。[/quote]典型的技术人才风格(不愿写文档    )。
不过如果有人感兴趣，再问andy007吧，楼主一定会尽力回答的。(楼主是吧 ？ )

andy007

呵呵，如果哪位有问题我会尽量解释的
不过这里我还有个问题，下面是相应的客户端程序，运行一直出现段错误，希望哪位有兴趣的帮我一起调试？


[code:1]
#include <math.h>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/time.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netdb.h>

#define PORT 9876
#define NUM 50

int main(int argc, char *argv[])
{
        /*
        ** Client: request a line from a remote file
        */
        struct timeval bef, aft;     /* timer value before/after */
        struct sockaddr_in sock;  /* structure for socket */
        struct hostent *hp;   /* structure for IP address */
        double random;    /* random reals : 0.0->1.0 */
        long timeleft;    /* time remaining in IAT */
        long secs = 0L;   /* inter-arrival time seconds */
        long usec = 0L;   /* inter-arrival microseconds */
        int pids[NUM];    /* subtask process ids */
        int opt = 1;    /* setsockopt parameter */
        int fd;     /* socket file descriptor */
        int ii, jj, kk = 0;
        char buf[80], *p;
        int status;   /* used in waitpid() */
       
        srand((unsigned int) getpid());  /* seed rand() */

        /*
        ** Operands are remote HOST name and IAT (Inter-Arrival Time)
        */
        if (argc != 3) {
                printf("\n\tUsage: %s <HOST> <IAT>\n\n", argv[0]);
                exit(-1);
        }

        /* uSeconds part of inter-arrival */
        if ((p = strchr(argv[2], '.')) != NULL) {
                *p = '\0'; /* end whole number at decimal */
                p++; /* bump pointer to start of mantissa */
                while (strlen(p) < 6) {
                        strcat(p, '0'); /* pad out to 6 digits */
                }
                p[6] = '\0'; /* struncate to 6 digits max */
                usec = atol(p);
        }
        secs = atol(argv[2]); /* seconds part of IAT */
       
        /*
        **        LOOP: send and receive NUM packets
        */
        for (ii = 0; ii < NUM; ii++) {
                /* get time before send */
                gettimeofday(&bef, NULL);

                /* random integer from 1 through 5 */
                random = rand() / (double) RAND_MAX;
                jj = (int) ((double) (5.0) * random) + 1;
                if (jj == 6)
                        jj = 5;
                sprintf(buf, "/tmp/sample.txt %d", jj);

                if ((pids[kk++] = fork()) == 0) {
                        /* set up socket info for connect */
                        fd = socket(AF_INET, SOCK_STREAM, 0);
                        memset((char*) &sock, 0, sizeof(sock));
                        hp = gethostbyname(argv[1]);
                        memcpy(&sock.sin_addr, hp->h_addr, hp->h_length);
                        sock.sin_family = hp->h_addrtype;
                        sock.sin_port = htons(PORT);
                        setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char *) &opt, sizeof(opt));

                        /* connect to server */
                        connect(fd, (struct sockaddr *)&sock, sizeof(sock));
                        send(fd, buf, strlen(buf) + 1, 0);
                        buf[0] = 0; /* clear buffer */
                        recv(fd, buf, sizeof(buf), 0);

                        /* print out response to our query */
                        printf("\tLine %d: '%s' ", jj, buf);
                        /* check for correct line */
                        p = strrchr(buf, ' ') + 2;
                        if (jj != atoi(p))
                                printf("*");
                        printf("\n");
                        close(fd);
                        exit(0);
                }

                /*
                **  Sleep for remainder of IAT
                */
                gettimeofday(&aft, NULL);
                aft.tv_sec -= bef.tv_sec;
                aft.tv_usec -= bef.tv_usec;
                if (aft.tv_usec < 0L) {
                        aft.tv_usec += 1000000L;
                        aft.tv_sec -= 1;
                }
                bef.tv_sec = secs;
                bef.tv_usec = usec;
                bef.tv_sec -= aft.tv_sec;
                bef.tv_usec -= aft.tv_usec;
                if (bef.tv_usec < 0L) {
                        bef.tv_usec += 1000000L;
                        bef.tv_sec -= 1;
                }
                timeleft = (bef.tv_sec * 1000000L) + bef.tv_usec;
                if (timeleft < 0) {
                        printf("\tERROR: A higher IAT value is required - exiting.\n");
                        break;
                }
                usleep(timeleft);
        }
        for (ii = 0; ii < kk; ii++) {
                waitpid(pids[ii], &status, 0);
        }
        puts("Done.");
        return 0;
}
[/code:1]

andy007

为了便于理解，我在变量部分作上注释：
                struct timeval bef, aft;     /* timer value before/after */
        struct sockaddr_in sock;  /* structure for socket */
        struct hostent *hp;   /* structure for IP address */
        double random;    /* random reals : 0.0->1.0 */
        long timeleft;    /* time remaining in IAT */
        long secs = 0L;   /* inter-arrival time seconds */
        long usec = 0L;   /* inter-arrival microseconds */
        int pids[NUM];    /* subtask process ids */
        int opt = 1;    /* setsockopt parameter */
        int fd;     /* socket file descriptor */
        int ii, jj, kk = 0;
        char buf[80], *p;
        int status;   /* used in waitpid() */

llc

大家将自己写的一些代码贴出来，有没有人捧场不要紧，就当作备份或笔记 :-)

lanche

等我的几个项目搞完了，一定找时间看看。

wuhu

else
不明白

wuhu

挺复杂的哦，慢慢看

andy007

[quote:2bc8eadcc5="wuhu"]else
不明白[/quote]

这些东西不履属于程序的，我原本要把某些语句加上颜色,但又没弄好。不好意思