Question n°4: Multi-client/multi-thread

Handling of waiting time

We will start from the example presented in the course on the use of select(). But, in the Practical Work subject, in addition to the read() and write() functions, we have a third blocking function, sleep(), to manage in the state machine. We will have to remove the use of the sleep() function and move the handling of waiting time via the timeout parameter of the select to the srv procedure.

In the global variables, we define a third fd_set that we will call fdwas and which will be used to store the set of sockets waiting for a 200000µs pause.

When we exit from the select(), after the pause has elapsed, the returned value is negative or zero. We will trigger the next state for all sockets on pause.

void srv(int fd_srv) {
 
    FD_ZERO(&fdres);
    FD_SET(fd_srv,&fdres);
    FD_ZERO(&fdwrs);
    FD_ZERO(&fdwas);

    while(1) {
        fd_set fds_read,fds_write;
        int n,i,fd_con;
		struct sockaddr_in adr_cli;
        socklen_t adr_cli_len;
		struct timeval timeout;
		
		timeout.tv_sec=0;
		timeout.tv_usec=200000;
		memcpy(&fds_read,&fdres,sizeof(fd_set));
		memcpy(&fds_write,&fdwrs,sizeof(fd_set));
		
		n=select(FD_SETSIZE,&fds_read,&fds_write,NULL,&timeout);
		if (n > 0)
			for (i = 0 ; i < FD_SETSIZE ; i++) {
				if (FD_ISSET(i,&fds_read)) {
					if (i==fd_srv) {
						adr_cli_len = sizeof(adr_cli);
						fd_con = accept(fd_srv, (struct sockaddr *) &adr_cli, &adr_cli_len);
						if (fd_con >= 0) state0(fd_con);
					} else state[i](i);
				}
				if (FD_ISSET(i,&fds_write)) state[i](i);
			}
		else 
			for (i = 0 ; i < FD_SETSIZE ; i++)
				if (FD_ISSET(i,&fdwas))
					state[i](i);
    }	
}

To manage state changes, we therefore slightly modify the will_read(), will_write(), and will_close() procedures and add a will_wait() procedure.

void will_read(int fd, void (*st)(int)) {
    FD_SET(fd,&fdres);
    FD_CLR(fd,&fdwrs);
    FD_CLR(fd,&fdwas);
    state[fd]=st;
}

void will_write(int fd, void (*st)(int)) {
    FD_SET(fd,&fdwrs);
    FD_CLR(fd,&fdres);
    FD_CLR(fd,&fdwas);
    state[fd]=st;
}

void will_wait(int fd, void (*st)(int)) {
    FD_CLR(fd,&fdres);
    FD_CLR(fd,&fdwrs);
    FD_SET(fd,&fdwas);
    state[fd]=st;
}

void will_close(int fd, void (*st)(int)) {
    FD_CLR(fd,&fdres);
    FD_CLR(fd,&fdwrs);
    FD_CLR(fd,&fdwas);
    state[fd]=NULL;
    st(fd);
    close(fd);
}

The state diagram

Now we just need to build the state diagram and program each corresponding state management procedure.

The variables used by the automaton procedures are grouped into a structure composed as follows:

struct tdata {
    int len;
    int pos;
    int end;
    char *input;
}   *data[FD_SETSIZE];
    
void state0(int fd) {
    printf("Connection accepted!\n");
    data[fd]=malloc(sizeof(struct tdata));
    data[fd]->input=NULL;
    data[fd]->len=0;
    data[fd]->pos=0;
    data[fd]->end=0;
    will_read(fd,&state1);
}
    
void state1(int fd) {
    int n,i,j;
    char buf[10],*tmp;
	
    n=read(fd,buf,sizeof(buf));
    if (n==0) {
        will_close(fd,&state4);
        return;
    }
    i=data[fd]->len;
    data[fd]->input=realloc(data[fd]->input,i+n);
    memcpy(data[fd]->input+i,buf,n);
    data[fd]->len+=n;

    for(j=data[fd]->len-1;j>=i;j--)
        if (data[fd]->input[j]=='\r' || data[fd]->input[j]=='\n')
            break;
    if (j<i) {
        will_read(fd,&state1);
        return;
    }
    data[fd]->end=j+1;
    will_write(fd,&state2);
}
    
void state2(int fd) {
    write(fd,data[fd]->input+data[fd]->pos,1);
    data[fd]->pos++;
    if (data[fd]->pos < data[fd]->end) {
        will_wait(fd,&state3);
        return;
    }
    memcpy(data[fd]->input,data[fd]->input+data[fd]->end,data[fd]->len-data[fd]->end);
    data[fd]->len-=data[fd]->end;
    data[fd]->end=0;
    data[fd]->pos=0;
    will_read(fd,&state1);
}
    
void state3(int fd) {
    will_write(fd,&state2);
}
    
void state4(int fd) {
    free(data[fd]->input);
    free(data[fd]);
    close(fd);
    printf("Connection closed!\n");
}

The complete source code

#include <stdio.h>
#include <stdlib.h>
 
#include <sys/types.h>
#include <netdb.h>
#include <sys/socket.h>
 
#include <sys/select.h>
 
#include <string.h>       // for the functions bzero, strcmp, strtok
#include <arpa/inet.h>    // for the function inet_addr
 
fd_set fdres,fdwrs,fdwas;
void (*state[FD_SETSIZE])(int);

void will_read(int fd, void (*st)(int)) {
    FD_SET(fd,&fdres);
    FD_CLR(fd,&fdwrs);
    FD_CLR(fd,&fdwas);
    state[fd]=st;
}

void will_write(int fd, void (*st)(int)) {
    FD_SET(fd,&fdwrs);
    FD_CLR(fd,&fdres);
    FD_CLR(fd,&fdwas);
    state[fd]=st;
}

void will_wait(int fd, void (*st)(int)) {
    FD_CLR(fd,&fdres);
    FD_CLR(fd,&fdwrs);
    FD_SET(fd,&fdwas);
    state[fd]=st;
}

void will_close(int fd, void (*st)(int)) {
    FD_CLR(fd,&fdres);
    FD_CLR(fd,&fdwrs);
    FD_CLR(fd,&fdwas);
    state[fd]=NULL;
    st(fd);
    close(fd);
}

void state0(int);
void state1(int);
void state2(int);
void state3(int);
void state4(int);

struct tdata {
    int len; 	// input buffer length
    int pos; 	// writing index
    int end; 	// end-of-write index
    char *input;
}   *data[FD_SETSIZE];

void state0(int fd) {
    printf("Connexion acceptée !\n");
    data[fd]=malloc(sizeof(struct tdata));
    data[fd]->input=NULL;
    data[fd]->len=0;
    data[fd]->pos=0;
    data[fd]->end=0;
    will_read(fd,&state1);
}

void state1(int fd) {
    int n,i,j;
    char buf[10],*tmp;
	
    n=read(fd,buf,sizeof(buf));
    if (n==0) {
        will_close(fd,&state4);
        return;
    }
    // Adding the read data to the input buffer
    i=data[fd]->len;
    data[fd]->input=realloc(data[fd]->input,i+n);
    memcpy(data[fd]->input+i,buf,n);
    data[fd]->len+=n;

    for(j=data[fd]->len-1;j>=i;j--)
        if (data[fd]->input[j]=='\r' || data[fd]->input[j]=='\n')
            break;
    if (j<i) {
        will_read(fd,&state1);
        return;
    }
    data[fd]->end=j+1;
    will_write(fd,&state2);
}

void state2(int fd) {
    write(fd,data[fd]->input+data[fd]->pos,1);
    data[fd]->pos++;
    if (data[fd]->pos < data[fd]->end) {
        will_wait(fd,&state3);
        return;
    }
    memcpy(data[fd]->input,data[fd]->input+data[fd]->end,data[fd]->len-data[fd]->end);
    data[fd]->len-=data[fd]->end;
    data[fd]->end=0;
    data[fd]->pos=0;
    will_read(fd,&state1);
}

void state3(int fd) {
    will_write(fd,&state2);
}

void state4(int fd) {
    free(data[fd]->input);
    free(data[fd]);
    close(fd);
    printf("Connection closed!\n");
}

void srv(int fd_srv) {
 
    FD_ZERO(&fdres);
    FD_SET(fd_srv,&fdres);
    FD_ZERO(&fdwrs);
    FD_ZERO(&fdwas);

    while(1) {
        fd_set fds_read,fds_write;
        int n,i,fd_con;
		struct sockaddr_in adr_cli;
        socklen_t adr_cli_len;
		struct timeval timeout;
		
		timeout.tv_sec=0;
		timeout.tv_usec=200000;
		memcpy(&fds_read,&fdres,sizeof(fd_set));
		memcpy(&fds_write,&fdwrs,sizeof(fd_set));
		
		n=select(FD_SETSIZE,&fds_read,&fds_write,NULL,&timeout);
		if (n > 0)
			for (i = 0 ; i < FD_SETSIZE ; i++ ) {
				if (FD_ISSET(i,&fds_read)) {
					if (i==fd_srv) {
						adr_cli_len = sizeof(adr_cli);
						fd_con = accept(fd_srv, (struct sockaddr *) &adr_cli, &adr_cli_len);
						if (fd_con >= 0) state0(fd_con);
					} else state[i](i);
				}
				if (FD_ISSET(i,&fds_write)) state[i](i);
			}
		else 
			for (i = 0 ; i < FD_SETSIZE ; i++ )
				if (FD_ISSET(i,&fdwas))
					state[i](i);
    }	
}

int main()
{
    int fd_srv;
    struct sockaddr_in  adr_srv;
    int err;

    fd_srv = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
    if (fd_srv < 0) {
        printf("Socket creation error!\n");
        exit(1);
    }
    bzero(&adr_srv, sizeof(adr_srv));
    adr_srv.sin_family = AF_INET;
    err = inet_aton("0.0.0.0", &adr_srv.sin_addr);
    if (err == 0) {
        printf("Invalid IPv4 address!\n");
        exit(1);
    }
    adr_srv.sin_port = htons(8080);
    err = bind(fd_srv, (struct sockaddr *) &adr_srv, sizeof(adr_srv));
    if (err != 0) {
        printf("Server port access error!\n");
        exit(1);
    }
    err = listen(fd_srv, 5);
    if (err != 0) {
        printf("Queue creation error!\n");
        exit(1);
    }
    printf("Server launched!\n");
	srv(fd_srv);
    return 0;
}