ASK Online

Today we are going to explore a very important system call in Linux, The Select system call. Select System call is use when we need non blocking executions. The use of the select can be explained via an example code. Lets assume that we use a socket program with a server and few client programs. To accept and read from clients what will be your normal approach?.. Probably you are checking for new clients and the data from the existing clients and probably use a read system call. Here the approach is blocking, the program is blocked for events. But using the select system call, it can be achieved in a non- blocking manner. Here we put the connected clients along with the server into a fd_set type variable. Then the select system call is looking for any read events for each clients and read the data non bloc-kingly.

Lets see an example code for this scenario.

In this example there is a server which is binded to port 9999. and clients can be connected to that port. Then the server identify whether a new client or data from an existing client. Select call will monitor the read_fds and act immediately for a change of a read file descriptor. There is a time out value for select system call.

In this example many clients are connected to the server and heart beats are send for the clients for every 3 seconds. And from the server we can send work(In this scenario messages) for clients. Then the clients work on those messages and send result to the server. We can non- bloc-kingly send data to clients and read data from clients using the select system call. :)

Server.cpp

#include <stdio.h>
#include <stdlib.h>
#include <cstring>
#include <unistd.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/socket.h>
#include <sys/select.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <pthread.h>
#include <time.h> 
#include "Data.h"

#define PORT "9999"   // port we're listening on
#define MAX_MESSAGE_SIZE 50 // Maximum message size

pthread_mutex_t mutexSync = PTHREAD_MUTEX_INITIALIZER;

int signal =0;
char Buffer[1024];

int Initialize(void);
void *Communicate(void *);
void *get_in_addr(struct sockaddr *sa);
void AcceptNewClients(fd_set &master , int &fdmax ,int listener);
void AcceptDataFromClients(int iClient , fd_set & master);
void SendHeartBeats(int iListener , int iClient);
void SendWorkForClients( int fdmax , int iListener , int iClient , int &signal);



//****************************************************************************************************
int main(void)
{
	pthread_t t;
	// char Buffer[256];
	int iListenSockId = Initialize();
	//Communicate(&iListenSockId);
	pthread_create(&t , NULL, Communicate , (void *)&iListenSockId); 
	while(1)
	{
		printf("Enter a message to Send :: ");
		fgets(Buffer , 255 , stdin);
		pthread_mutex_lock(&mutexSync);
		signal++;
		//printf("Message is %s %d \n", Buffer , signal);
		pthread_mutex_unlock(&mutexSync);
		sleep(1);
		//memset(&Buffer[0], 0, sizeof(Buffer));
			
	}
	
    return 0;
}

//****************************************************************************************************
int Initialize()
{
	int listener ;     // listening socket descriptor
	int rv ;
	struct addrinfo hints, *ai, *p;
	int yes=1;        // for setsockopt() 
	 // get us a socket and bind it
    
	memset(&hints, 0, sizeof hints);
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_flags = AI_PASSIVE;
    if ((rv = getaddrinfo(NULL, PORT, &hints, &ai)) != 0) 
	{
        fprintf(stderr, "selectserver: %s\n", gai_strerror(rv));
        exit(1);
    }
    
    for(p = ai; p != NULL; p = p->ai_next) 
	{
        listener = socket(p->ai_family, p->ai_socktype, p->ai_protocol);
        if (listener < 0) 
		{ 
            continue;
        }       
        // lose the pesky "address already in use" error message
        setsockopt(listener, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int));
        if (bind(listener, p->ai_addr, p->ai_addrlen) < 0) 
		{
            close(listener);
            continue;
        }
        break;
    }
    // if we got here, it means we didn't get bound
    if (p == NULL) 
	{
        fprintf(stderr, "selectserver: failed to bind\n");
        exit(2);
    }
    freeaddrinfo(ai); // all done with this
    // listen
    if (listen(listener, 10) == -1) {
        perror("listen");
        exit(3);
    }
	
	return listener;

}

//****************************************************************************************************
void * Communicate(void * id)
{
	int *iSockID = (int *) id;
	int listener =  *iSockID;

    fd_set master;    // master file descriptor list
    fd_set read_fds;  // temp file descriptor list for select() read
    int fdmax;        // maximum file descriptor number

    int i, j, rv;

    FD_ZERO(&master);    // clear the master and temp sets
    FD_ZERO(&read_fds);
    // add the listener to the master set
    FD_SET(listener, &master);
	printf("Listener is %d \n" , listener);

    // keep track of the biggest file descriptor
    fdmax = listener; // so far, it's this one
	//accept 3 clients

	
    // main loop
    for(;;) {
		memcpy(&read_fds , &master , sizeof(master));
		struct timeval tv;
		tv.tv_sec = 3;
		tv.tv_usec = 0;
		int iResult = select(fdmax+1, &read_fds, NULL, NULL, &tv) ;
        if (iResult == -1) 
		{
            perror("select");
            exit(4);
        }
	
		// ADD NEW CONNECTIONS READ FROM CONNECTIONS	
		for(i = 0; i <= fdmax; i++)
		{			   
            if (FD_ISSET(i, &read_fds)) 
			{			
				if (i == listener) 
				{                                    
					AcceptNewClients(master , fdmax , listener );	
                } else 
				{
					AcceptDataFromClients(i , master);
                } 
            } 

			
        } 
        for(i = 0; i <= fdmax; i++) 
		{			
			//send work for clients
			SendWorkForClients(fdmax , listener , i , signal);
			//sending heart beats
		   	SendHeartBeats(listener , i );
		}

    } 
	return 0;
}

//****************************************************************************************************
void SendWorkForClients(int fdmax , int iListener , int iClient , int &signal)
{

		if(signal == 1 && iClient != iListener && iClient > iListener)
		{
			//send data to some client;
			//printf("data send to CLIENT DUMMY \n");
			//signal--;
			ServerData * data = new ServerData();
			data->iType = 2;
			strcpy(data->cMessage , Buffer );
			int numbytes  = write( iClient , data ,sizeof(*data));
			printf("Number of bytes written :: %d ID :: %d  \n" , numbytes , iClient);
		
			delete data;
		}
		if(signal == 1 && iClient != iListener && iClient > iListener && iClient == fdmax)
		{
			pthread_mutex_lock(&mutexSync);
			signal--;
			pthread_mutex_unlock(&mutexSync);
		}
	
}

//****************************************************************************************************
void SendHeartBeats(int iListener , int iClient)
{
	ServerData * data = new ServerData();
	time_t rawtime;
	time ( &rawtime );

	data->iType = 1;
	strcpy(data->cMessage , ctime (&rawtime)  );
	//char cHeartBeat []  = "HEARTBEAT";


		if(iClient!= iListener && iClient > iListener){
			int numbytes  = write( iClient , data , sizeof(*data)); 
			//printf("Number of bytes written :: %d ID :: %d  \n" , numbytes , i);
		}					
	
	delete data;
}
//****************************************************************************************************
void AcceptNewClients(fd_set&  master , int& fdmax ,int listener)
{
		socklen_t addrlen;
		struct sockaddr_storage remoteaddr; // client address
		char remoteIP[INET6_ADDRSTRLEN];		
		int newfd;        // newly accept()ed socket descriptor
		addrlen = sizeof(remoteaddr);
		ServerData * data = new ServerData();
		data->iType = 2;
		strcpy(data->cMessage , "SERVER MESSAGE :: ACCEPTED" );
        newfd = accept(listener, (struct sockaddr *)&remoteaddr, &addrlen);
		write(newfd , data , sizeof(*data));
		if (newfd == -1) 
		{
            perror("accept");
        }else 
		{
            FD_SET(newfd, &master); // add to master set
            if (newfd > fdmax) 
			{    // keep track of the max
				fdmax = newfd;
            }
            printf("selectserver: new connection from %s on socket %d\n", inet_ntop(remoteaddr.ss_family,
                                get_in_addr((struct sockaddr*)&remoteaddr), remoteIP, INET6_ADDRSTRLEN), newfd);
        } 
		
		delete data;
}

//****************************************************************************************************
void AcceptDataFromClients(int iClient , fd_set& master)
{
	char buf[1024];    // buffer for client data
	int nbytes;
	
	nbytes = recv(iClient, buf, sizeof buf, 0);
	//printf("DISCONNECTING  %d \n", nbytes);
    if (nbytes <= 0) 
	{
    // got error or connection closed by client
		if (nbytes == 0) 
		{
			// connection closed
			printf("selectserver: socket %d hung up\n", iClient);
		} else 
		{
			perror("recv");
		}
        close(iClient); // bye!
        FD_CLR(iClient, &master); // remove from master set
    } else 
	{
    // we got some data from a client
		printf("CLIENT DATA :: %s",buf);
                       
    }

}

//****************************************************************************************************
// get sockaddr, IPv4 or IPv6:
void *get_in_addr(struct sockaddr *sa)
{
    if (sa->sa_family == AF_INET) 
	{
        return &(((struct sockaddr_in*)sa)->sin_addr);
    }

    return &(((struct sockaddr_in6*)sa)->sin6_addr);
}

Client.cpp

#include <stdio.h>
#include <sys/socket.h>
#include <stdlib.h>
#include <sys/types.h>
#include <netinet/in.h>
#include <netdb.h>
#include <string.h>
#include <unistd.h>
#include "Data.h"


int main(int argc , char * argv [])
{
	int SockFD , PortNo , n , fdmax ,nbytes;
	struct sockaddr_in Server_Address;
	struct hostent *Server;
	fd_set master;    // master file descriptor list
    fd_set read_fds;  // temp file descriptor list for select()
	FD_ZERO(&master);    // clear the master and temp sets
    FD_ZERO(&read_fds);

	char Buffer[1024];
	memset(Buffer, 0, sizeof(Buffer));

	if(argc < 3)
	{
		printf("Error hostname port required \n");
		exit(0);
	}

	PortNo = atoi(argv[2]);

	// create a socket	
	SockFD =  socket(AF_INET , SOCK_STREAM , 0);

	if(SockFD < 0 )
	{
		perror("Error Creating Socket");
		exit(1);
	}
	
	Server = (struct hostent *) gethostbyname(argv[1]);
	if(Server == NULL)
	{
		printf("Error :: No Such Host \n");
		exit(0);
	}

	bzero((char *) &Server_Address , sizeof(Server_Address));
	Server_Address.sin_family =  AF_INET;
	bcopy((char *) Server->h_addr , (char *) &Server_Address.sin_addr.s_addr , Server->h_length);
	
	Server_Address.sin_port = htons(PortNo);
	
	//connect to the server
	int listener = connect(SockFD , (struct sockaddr *) &Server_Address , sizeof(Server_Address));
	if( listener < 0)
	{
		perror("Error Connecting \n");
		exit(1);
	}
	

	while(1)
	{	
		n = read(SockFD , Buffer , sizeof(Buffer));
		ServerData * data = new ServerData();
		data = (ServerData *) Buffer;
		
		if(n <= 0)
		{
			perror("Error receiving data \n");
			exit(1);
		}
		//int result = strcmp( Buffer, "HEARTBEAT" );
		if(data->iType == 2)
		{
			//sleep(1);	
			printf("Received :: %s", data->cMessage);
			write(SockFD , "CLIENT MESSAGE-->ACCEPTED" ,26);
		}	
		else if(data->iType == 1 )
		{
			//sleep(2);	
			printf("Heart Beating Server::  %s", data->cMessage);
		}
		memset(Buffer, 0, sizeof(Buffer));
		//
		
	}

	return 0;

}

Data.h

// Header file for data
// type 1 -> Heart Beat Message
// type 2 -> Server Task Message

class ServerData
{
	public:
	int iType;
	char cMessage [50];
};

To compile and run the server...

1
2
3

g++ Select.cpp -lpthread

./a.out

To compile and run the client...

1
2
3

g++ -o client client.cpp

./client localhost 9999

That's it folks. If you have anything to clarify just put a comment. I'll answer for the best of my knowledge.

Here is a nice tutorial for Linux system calls. I recommend you all to read it.
http://www.beej.us/guide/bgnet/output/html/multipage/index.html

Pages

ASK Online

Ads 468x60px

Labels

Saturday, February 9, 2013

Non Blocking I/O using SELECT System Call in Linux

DZone - Most-Valuable-Blogger

Blog Archive

Popular Posts

Followers

Total Pageviews

Hathmaluwa