kolan/dev
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

ServerNBTHR

Browse Source
This commit is contained in:
Kolan Sh 2011-05-02 12:11:12 +04:00
parent a5d64c0019
commit 3a281d5581
13 changed files with 617 additions and 515 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
c/ServerNBTHR/ClientSideProc.c
View File

@ -15,8 +15,8 @@
int main(int argc, char **argv) int main(int argc, char **argv)
{ {
CLIENT *cl; CLIENT *cl;
struct square_in in; square_in in;
struct square_out out; square_out out;
if (argc != 3) { if (argc != 3) {
printf("Usage : client <hostname> <integer_valus=e>\n"); printf("Usage : client <hostname> <integer_valus=e>\n");
exit(1); exit(1);

376
c/ServerNBTHR/ServerNBTHR.c
View File

@ -16,72 +16,65 @@
pthread_mutex_t request_mutex = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t request_mutex = PTHREAD_MUTEX_INITIALIZER;
void void die(const char *func, int err)
die(const char *func, int err)
{ {
fprintf(stderr,"%s: %s\n",func, strerror(err)); fprintf(stderr, "%s: %s\n", func, strerror(err));
abort(); abort();
} }
void void bark(const char *func, int err)
bark(const char *func, int err)
{ {
fprintf(stderr,"%s: %s\n",func, strerror(err)); fprintf(stderr, "%s: %s\n", func, strerror(err));
} }
/* /*
Описание поцедуры ведущего потока , которая возвращает Описание поцедуры ведущего потока , которая возвращает
дескрипторов пассивного сокета, привязанного к адресу дескрипторов пассивного сокета, привязанного к адресу
сервера. сервера.
*/ */
int int getServerSocket(unsigned short int port)
getServerSocket(unsigned short int port)
{ {
int listenSocket; int listenSocket;
struct sockaddr_in listenSockaddr; struct sockaddr_in listenSockaddr;
if((listenSocket=socket(PF_INET,SOCK_STREAM,0))<0) if ((listenSocket = socket(PF_INET, SOCK_STREAM, 0)) < 0)
die("socket()",errno); die("socket()", errno);
memset(&listenSockaddr, 0, sizeof(listenSockaddr)); memset(&listenSockaddr, 0, sizeof(listenSockaddr));
listenSockaddr.sin_family = PF_INET; listenSockaddr.sin_family = PF_INET;
listenSockaddr.sin_port = htons(port); listenSockaddr.sin_port = htons(port);
listenSockaddr.sin_addr.s_addr = INADDR_ANY; listenSockaddr.sin_addr.s_addr = INADDR_ANY;
if(bind(listenSocket,(struct sockaddr*)&listenSockaddr, if (bind(listenSocket, (struct sockaddr *)&listenSockaddr, sizeof(listenSockaddr)) < 0)
sizeof(listenSockaddr)) < 0) die("bind()", errno);
die("bind()",errno);
if(listen(listenSocket,5)<0) if (listen(listenSocket, 5) < 0)
die("listen()",errno); die("listen()", errno);
return listenSocket; return listenSocket;
} }
/* /*
Описание процедуры выполняемой всеми ведомыми потоками Описание процедуры выполняемой всеми ведомыми потоками
*/ */
void *serv_request(void *data)
void *
serv_request(void *data)
{ {
struct connection_cb struct connection_cb {
{ int dataSocket;
int dataSocket; char data[256];
char data[256]; int dataSent;
int dataSent; int dataToSend;
int dataToSend; int isReading;
int isReading; struct connection_cb *next;
struct connection_cb *next; };
};
struct connection_cb *connections = NULL; struct connection_cb *connections = NULL;
int listenSocket = (int)data; int listenSocket = (int)data;
if(fcntl(listenSocket,F_SETFL,O_NONBLOCK)<0) if (fcntl(listenSocket, F_SETFL, O_NONBLOCK) < 0)
die("fcntl()",errno); die("fcntl()", errno);
while(1) while (1) {
{
fd_set readFdSet; fd_set readFdSet;
fd_set writeFdSet; fd_set writeFdSet;
struct connection_cb *currentConn, **currentConnPtr, *tempConn; struct connection_cb *currentConn, **currentConnPtr, *tempConn;
@ -90,204 +83,179 @@ int listenSocket = (int)data;
FD_ZERO(&readFdSet); FD_ZERO(&readFdSet);
FD_ZERO(&writeFdSet); FD_ZERO(&writeFdSet);
/* /*
Добавление дескриптора к множеству readFdSet Добавление дескриптора к множеству readFdSet
*/ */
FD_SET(listenSocket,&readFdSet); FD_SET(listenSocket, &readFdSet);
maxFdNum = listenSocket; maxFdNum = listenSocket;
for(currentConn = connections;currentConn!=NULL;currentConn = for (currentConn = connections; currentConn != NULL; currentConn = currentConn->next) {
currentConn->next) if (currentConn->isReading)
{ FD_SET(currentConn->dataSocket, &readFdSet);
if(currentConn->isReading) else
FD_SET(currentConn->dataSocket,&readFdSet); FD_SET(currentConn->dataSocket, &writeFdSet);
else maxFdNum = currentConn->dataSocket > maxFdNum ? currentConn - >dataSocket : maxFdNum;
FD_SET(currentConn->dataSocket,&writeFdSet);
maxFdNum = currentConn->dataSocket > maxFdNum ?currentConn-
>dataSocket : maxFdNum;
}
/*
Получение множества дескрипторов сокетов для обработки
*/
if(select(maxFdNum+1,&readFdSet,&writeFdSet,NULL,NULL) < 0)
{
if(errno == EINTR)
continue;
die("select()",errno);
}
currentConnPtr=&connections;
while(*currentConnPtr!=NULL)
{
/*
Проверка принадлежности дескриптора
(*currentConnPtr)->dataSocket к множеству readFdSet
*/
if((*currentConnPtr)->isReading &&
FD_ISSET((*currentConnPtr)->dataSocket,&readFdSet))
{
int result = recv((*currentConnPtr)->dataSocket, (*currentConnPtr)->data,
sizeof((*currentConnPtr)->data),0);
if(result < 0)
{
if(errno!=EINTR && errno!=EAGAIN && errno!=EWOULDBLOCK)
{
bark("recv()",errno);
close((*currentConnPtr)->dataSocket);
tempConn = *currentConnPtr;
*currentConnPtr = (*currentConnPtr)->next;
free(tempConn);
continue;
}
} }
else /*
if(result==0) Получение множества дескрипторов сокетов для обработки
{ */
close((*currentConnPtr)->dataSocket); if (select(maxFdNum + 1, &readFdSet, &writeFdSet, NULL, NULL) < 0) {
tempConn = *currentConnPtr; if (errno == EINTR)
*currentConnPtr = (*currentConnPtr)->next; continue;
free(tempConn); die("select()", errno);
continue;
} }
else
{
(*currentConnPtr)->dataToSend = result;
(*currentConnPtr)->dataSent = 0;
(*currentConnPtr)->isReading = 0;
printf("Recieving as Slave Thread id = '%d' \n",pthread_self());
}
}
else
/*
Проверка принадлежности дескриптора
(*currentConnPtr)->dataSocket к множеству writedFdSet
*/
if(FD_ISSET((*currentConnPtr)->dataSocket,&writeFdSet))
{
int result = send((*currentConnPtr)->dataSocket,
(*currentConnPtr)->data+(*currentConnPtr)->dataSent,
(*currentConnPtr) ->dataToSend-(*currentConnPtr)->dataSent, 0);
if(result < 0) currentConnPtr = &connections;
{
if(errno!=EINTR && errno!=EAGAIN)
{
bark("write()",errno);
close((*currentConnPtr)->dataSocket);
tempConn = *currentConnPtr;
*currentConnPtr = (*currentConnPtr)->next;
free(tempConn);
continue;
}
}
else
{
(*currentConnPtr)->dataSent +=result;
if((*currentConnPtr)->dataSent >= (*currentConnPtr)->dataToSend) while (*currentConnPtr != NULL) {
(*currentConnPtr)->isReading = 1;
/*
Проверка принадлежности дескриптора
(*currentConnPtr)->dataSocket к множеству readFdSet
*/
if ((*currentConnPtr)->isReading && FD_ISSET((*currentConnPtr)->dataSocket, &readFdSet)) {
int result = recv((*currentConnPtr)->dataSocket, (*currentConnPtr)->data,
sizeof((*currentConnPtr)->data), 0);
if (result < 0) {
if (errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK) {
bark("recv()", errno);
close((*currentConnPtr)->dataSocket);
tempConn = *currentConnPtr;
*currentConnPtr = (*currentConnPtr)->next;
free(tempConn);
continue;
}
} else if (result == 0) {
close((*currentConnPtr)->dataSocket);
tempConn = *currentConnPtr;
*currentConnPtr = (*currentConnPtr)->next;
free(tempConn);
continue;
} else {
(*currentConnPtr)->dataToSend = result;
(*currentConnPtr)->dataSent = 0;
(*currentConnPtr)->isReading = 0;
printf("Recieving as Slave Thread id = '%d' \n", pthread_self());
}
} else
/*
Проверка принадлежности дескриптора
(*currentConnPtr)->dataSocket к множеству writedFdSet
*/
if (FD_ISSET((*currentConnPtr)->dataSocket, &writeFdSet)) {
int result = send((*currentConnPtr)->dataSocket,
(*currentConnPtr)->data + (*currentConnPtr)->dataSent,
(*currentConnPtr)->dataToSend - (*currentConnPtr)->dataSent, 0);
if (result < 0) {
if (errno != EINTR && errno != EAGAIN) {
bark("write()", errno);
close((*currentConnPtr)->dataSocket);
tempConn = *currentConnPtr;
*currentConnPtr = (*currentConnPtr)->next;
free(tempConn);
continue;
}
} else {
(*currentConnPtr)->dataSent += result;
if ((*currentConnPtr)->dataSent >= (*currentConnPtr)->dataToSend)
(*currentConnPtr)->isReading = 1;
} }
} }
currentConnPtr = &((*currentConnPtr)->next); currentConnPtr = &((*currentConnPtr)->next);
printf("Sending as Slave Thread id = '%d' \n",pthread_self()); printf("Sending as Slave Thread id = '%d' \n", pthread_self());
} }
/* /*
Проверка принадлежности дескриптора listenSocket Проверка принадлежности дескриптора listenSocket
к множеству readFdSet,т.е. необходимости обработать к множеству readFdSet,т.е. необходимости обработать
вызов connect( ) от нового клиента. вызов connect( ) от нового клиента.
*/ */
if(FD_ISSET(listenSocket,&readFdSet)) if (FD_ISSET(listenSocket, &readFdSet)) {
{
while(1) while (1) {
{
/* /*
Вызовы pthread_mutex_lock, pthread_mutex_unlock Вызовы pthread_mutex_lock, pthread_mutex_unlock
Не нужны в среде Linux Не нужны в среде Linux
*/ */
pthread_mutex_lock(&request_mutex); pthread_mutex_lock(&request_mutex);
int result = accept(listenSocket,(struct sockaddr*)NULL,NULL); int result = accept(listenSocket, (struct sockaddr *)NULL, NULL);
pthread_mutex_unlock(&request_mutex); pthread_mutex_unlock(&request_mutex);
if(result < 0) if (result < 0) {
{ if (errno == EAGAIN || errno == EWOULDBLOCK)
if(errno==EAGAIN || errno == EWOULDBLOCK) break;
break; die("accept()", errno);
die("accept()",errno); } else {
} *currentConnPtr = malloc(sizeof(struct connection_cb));
else if (*currentConnPtr == NULL)
{ die("malloc()", 0);
*currentConnPtr = malloc(sizeof(struct connection_cb));
if(*currentConnPtr==NULL)
die("malloc()",0);
if(fcntl(result,F_SETFL,O_NONBLOCK)<0) if (fcntl(result, F_SETFL, O_NONBLOCK) < 0)
die("fcntl()",errno); die("fcntl()", errno);
(*currentConnPtr)->dataSocket = result; (*currentConnPtr)->dataSocket = result;
(*currentConnPtr)->isReading = 1; (*currentConnPtr)->isReading = 1;
(*currentConnPtr)->next = 0; (*currentConnPtr)->next = 0;
currentConnPtr = &((*currentConnPtr)->next); currentConnPtr = &((*currentConnPtr)->next);
printf("Accepting as Master Thread id = '%d' \n",pthread_self()); printf("Accepting as Master Thread id = '%d' \n", pthread_self());
} }
} }
} }
} }
} }
int
main(int argc,char *argv[])
{
int k;
int descSock;
char *service="1500";
switch(argc) {
case 1:
break;
case 2:
service = argv[1];
break;
default:
printf ("Usage: ./ServerBNTH [port]\n");
exit(1);
}
size_t stacksize; int main(int argc, char *argv[])
pthread_t p_thread[NUM_THREADS]; {
int k;
int descSock;
char *service = "1500";
switch (argc) {
case 1:
break;
case 2:
service = argv[1];
break;
default:
printf("Usage: ./ServerBNTH [port]\n");
exit(1);
}
size_t stacksize;
pthread_t p_thread[NUM_THREADS];
/* /*
Установка размера стека для ведомых потоков Установка размера стека для ведомых потоков
*/ */
pthread_attr_t attr; pthread_attr_t attr;
pthread_attr_init(&attr); pthread_attr_init(&attr);
stacksize = 500000; stacksize = 500000;
pthread_attr_setstacksize (&attr, stacksize); pthread_attr_setstacksize(&attr, stacksize);
pthread_attr_getstacksize (&attr, &stacksize); pthread_attr_getstacksize(&attr, &stacksize);
/* /*
Получение значения дескриптора пассивного сокета Получение значения дескриптора пассивного сокета
*/ */
descSock = getServerSocket(atoi(service)); descSock = getServerSocket(atoi(service));
/* /*
Запуск ведомых потоков Запуск ведомых потоков
*/ */
for(k=0; k<NUM_THREADS; k++) { for (k = 0; k < NUM_THREADS; k++) {
pthread_create(&p_thread[k],&attr,serv_request,(void*)descSock); pthread_create(&p_thread[k], &attr, serv_request, (void *)descSock);
printf("Thread %d started \n",k); printf("Thread %d started \n", k);
} }
pthread_attr_destroy(&attr); pthread_attr_destroy(&attr);
for(k=0;k<NUM_THREADS;k++) { for (k = 0; k < NUM_THREADS; k++) {
pthread_join(p_thread[k], NULL); pthread_join(p_thread[k], NULL);
printf("Completed join with thread %d\n",k); printf("Completed join with thread %d\n", k);
} }
} }

33
c/ServerNBTHR/ServerSideProc.c
View File

@ -1,7 +1,7 @@
/* /*
ServerSideProc.c ServerSideProc.c
*/ */
#include "square.h" #include "square.h"
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <rpc/pmap_clnt.h> #include <rpc/pmap_clnt.h>
@ -10,21 +10,22 @@
#include <sys/socket.h> #include <sys/socket.h>
#include <netinet/in.h> #include <netinet/in.h>
int request=0; int request = 0;
bool_t squareproc_2_svc(square_in *inp,square_out *outp,struct svc_req *rqstp) bool_t squareproc_2_svc(square_in * inp, square_out * outp, struct svc_req *rqstp)
{ {
printf("Thread id = '%ld' started, arg = %d\n",pthread_self(),inp->arg1); printf("Thread id = '%ld' started, arg = %d\n", pthread_self(), inp->arg1);
/* /*
Имитация работы процедуры , выполняемой потоками сервера Имитация работы процедуры , выполняемой потоками сервера
*/ */
sleep(5); sleep(5);
outp->res1=inp->arg1*inp->arg1; outp->res1 = inp->arg1 * inp->arg1;
printf("Thread id = '%ld' is done %d \n",pthread_self(),outp->res1); printf("Thread id = '%ld' is done %d \n", pthread_self(), outp->res1);
return(TRUE); return (TRUE);
} }
int square_prog_2_freeresult(SVCXPRT *transp,xdrproc_t xdr_result, caddr_t result)
int square_prog_2_freeresult(SVCXPRT * transp, xdrproc_t xdr_result, caddr_t result)
{ {
xdr_free(xdr_result,result); xdr_free(xdr_result, result);
return(1); return (1);
} }

11
c/ServerNBTHR/square.h
View File

@ -1,11 +0,0 @@
struct square_in {
long arg1;
};
struct square_out {
long res1;
};
program SQUARE_PROG {
version SQUARE_VERS {
square_out SQUAREPROC(square_in) = 1;
} = 2 ;
} = 0x31230000;

48
c/ServerNBTHR/square_client.c
View File

@ -1,48 +0,0 @@
/*
* This is sample code generated by rpcgen.
* These are only templates and you can use them
* as a guideline for developing your own functions.
*/
#include "square.h"
void
square_prog_2(char *host)
{
CLIENT *clnt;
enum clnt_stat retval_1;
square_out result_1;
square_in squareproc_2_arg;
#ifndef DEBUG
clnt = clnt_create (host, SQUARE_PROG, SQUARE_VERS, "udp");
if (clnt == NULL) {
clnt_pcreateerror (host);
exit (1);
}
#endif /* DEBUG */
retval_1 = squareproc_2(&squareproc_2_arg, &result_1, clnt);
if (retval_1 != RPC_SUCCESS) {
clnt_perror (clnt, "call failed");
}
#ifndef DEBUG
clnt_destroy (clnt);
#endif /* DEBUG */
}
int
main (int argc, char *argv[])
{
char *host;
if (argc < 2) {
printf ("usage: %s server_host\n", argv[0]);
exit (1);
}
host = argv[1];
square_prog_2 (host);
exit (0);
}

31
c/ServerNBTHR/square_server.c
View File

@ -1,31 +0,0 @@
/*
* This is sample code generated by rpcgen.
* These are only templates and you can use them
* as a guideline for developing your own functions.
*/
#include "square.h"
bool_t
squareproc_2_svc(square_in *argp, square_out *result, struct svc_req *rqstp)
{
bool_t retval;
/*
* insert server code here
*/
return retval;
}
int
square_prog_2_freeresult (SVCXPRT *transp, xdrproc_t xdr_result, caddr_t result)
{
xdr_free (xdr_result, result);
/*
* Insert additional freeing code here, if needed
*/
return 1;
}

179
c/ServerNBTHR/square_svc.c
View File

@ -25,26 +25,24 @@ pthread_attr_t attr;
*/ */
void * void *serv_request(void *data)
serv_request(void *data)
{ {
struct thr_data struct thr_data {
{ struct svc_req *rqstp;
struct svc_req *rqstp; SVCXPRT *transp;
SVCXPRT *transp; } *ptr_data;
} *ptr_data;
{ {
union { union {
square_in squareproc_2_arg; square_in squareproc_2_arg;
} argument; } argument;
union { union {
square_out squareproc_2_res; square_out squareproc_2_res;
} result; } result;
bool_t retval; bool_t retval;
xdrproc_t _xdr_argument, _xdr_result; xdrproc_t _xdr_argument, _xdr_result;
bool_t (*local)(char *, void *, struct svc_req *); bool_t(*local) (char *, void *, struct svc_req *);
/* /*
@ -52,53 +50,51 @@ SVCXPRT *transp;
*/ */
ptr_data = (struct thr_data *)data; ptr_data = (struct thr_data *)data;
struct svc_req *rqstp = ptr_data->rqstp; struct svc_req *rqstp = ptr_data->rqstp;
register SVCXPRT *transp = ptr_data->transp; register SVCXPRT *transp = ptr_data->transp;
switch (rqstp->rq_proc) {
case NULLPROC:
(void)svc_sendreply(transp, (xdrproc_t) xdr_void, (char *)NULL);
return;
switch (rqstp->rq_proc) { case SQUAREPROC:
case NULLPROC: _xdr_argument = (xdrproc_t) xdr_square_in;
(void) svc_sendreply (transp, (xdrproc_t) xdr_void, (char *)NULL); _xdr_result = (xdrproc_t) xdr_square_out;
return; local = (bool_t(*)(char *, void *, struct svc_req *))squareproc_2_svc;
break;
case SQUAREPROC: default:
_xdr_argument = (xdrproc_t) xdr_square_in; svcerr_noproc(transp);
_xdr_result = (xdrproc_t) xdr_square_out; return;
local = (bool_t (*) (char *, void *, struct svc_req *))squareproc_2_svc; }
break; memset((char *)&argument, 0, sizeof(argument));
if (!svc_getargs(transp, (xdrproc_t) _xdr_argument, (caddr_t) & argument)) {
svcerr_decode(transp);
return;
}
default: /*
svcerr_noproc (transp);
return;
}
memset ((char *)&argument, 0, sizeof (argument));
if (!svc_getargs (transp, (xdrproc_t) _xdr_argument, (caddr_t) &argument)) {
svcerr_decode (transp);
return;
}
/* Стандартный вызов функции сервера.
Данные для вызова уже приведены к стандарту.
Стандартный вызов функции сервера. */
Данные для вызова уже приведены к стандарту.
*/ retval = (bool_t) (*local) ((char *)&argument, (void *)&result, rqstp);
retval = (bool_t) (*local)((char *)&argument, (void *)&result, rqstp); if (retval > 0 && !svc_sendreply(transp, (xdrproc_t) _xdr_result, (char *)&result)) {
svcerr_systemerr(transp);
if (retval > 0 && !svc_sendreply(transp, (xdrproc_t) _xdr_result, (char *)&result)) }
{ if (!svc_freeargs(transp, (xdrproc_t) _xdr_argument, (caddr_t) & argument)) {
svcerr_systemerr (transp); fprintf(stderr, "%s", "unable to free arguments");
} exit(1);
if (!svc_freeargs (transp, (xdrproc_t) _xdr_argument, (caddr_t) &argument)) { }
fprintf (stderr, "%s", "unable to free arguments"); if (!square_prog_2_freeresult(transp, _xdr_result, (caddr_t) & result))
exit (1); fprintf(stderr, "%s", "unable to free results");
} return;
if (!square_prog_2_freeresult (transp, _xdr_result, (caddr_t) &result)) }
fprintf (stderr, "%s", "unable to free results");
return;
}
} }
/* /*
@ -109,58 +105,39 @@ register SVCXPRT *transp = ptr_data->transp;
*/ */
static void static void square_prog_2(struct svc_req *rqstp, register SVCXPRT * transp)
square_prog_2(struct svc_req *rqstp, register SVCXPRT *transp)
{
struct data_str
{
struct svc_req *rqstp;
SVCXPRT *transp;
} *data_ptr =(struct data_str*)malloc(sizeof(struct data_str);
{ {
struct data_str {
struct svc_req *rqstp;
SVCXPRT *transp;
} *data_ptr = (struct data_str *)malloc(sizeof(struct data_str); {
/* /*
Упаковка данных в структуру для передачи ссылки на нее, Упаковка данных в структуру для передачи ссылки на нее,
как параметра запускаемому потоку как параметра запускаемому потоку
*/ */
data_ptr->rqstp = rqstp; data_ptr->rqstp = rqstp;
data_ptr->transp = transp; data_ptr->transp = transp;
pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED); pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
pthread_create(&p_thread,&attr,serv_request,(void *)data_ptr); pthread_create(&p_thread, &attr, serv_request, (void *)data_ptr);}
} }
}
int int main(int argc, char **argv) {
main (int argc, char **argv) register SVCXPRT * transp;
{ pmap_unset(SQUARE_PROG, SQUARE_VERS);
register SVCXPRT *transp; transp = svcudp_create(RPC_ANYSOCK); if (transp == NULL) {
fprintf(stderr, "%s", "cannot create udp service."); exit(1);}
if (!svc_register(transp, SQUARE_PROG, SQUARE_VERS, square_prog_2, IPPROTO_UDP)) {
fprintf(stderr, "%s", "unable to register (SQUARE_PROG, SQUARE_VERS, udp).");
exit(1);}
pmap_unset (SQUARE_PROG, SQUARE_VERS); transp = svctcp_create(RPC_ANYSOCK, 0, 0); if (transp == NULL) {
fprintf(stderr, "%s", "cannot create tcp service."); exit(1);}
if (!svc_register(transp, SQUARE_PROG, SQUARE_VERS, square_prog_2, IPPROTO_TCP)) {
fprintf(stderr, "%s", "unable to register (SQUARE_PROG, SQUARE_VERS, tcp).");
exit(1);}
transp = svcudp_create(RPC_ANYSOCK); svc_run(); fprintf(stderr, "%s", "svc_run returned"); exit(1);
if (transp == NULL) { /* NOTREACHED */
fprintf (stderr, "%s", "cannot create udp service."); }
exit(1);
}
if (!svc_register(transp, SQUARE_PROG, SQUARE_VERS, square_prog_2, IPPROTO_UDP)) {
fprintf (stderr, "%s", "unable to register (SQUARE_PROG, SQUARE_VERS, udp).");
exit(1);
}
transp = svctcp_create(RPC_ANYSOCK, 0, 0);
if (transp == NULL) {
fprintf (stderr, "%s", "cannot create tcp service.");
exit(1);
}
if (!svc_register(transp, SQUARE_PROG, SQUARE_VERS, square_prog_2, IPPROTO_TCP)) {
fprintf (stderr, "%s", "unable to register (SQUARE_PROG, SQUARE_VERS, tcp).");
exit(1);
}
svc_run ();
fprintf (stderr, "%s", "svc_run returned");
exit (1);
/* NOTREACHED */
}

66
c/masync_server/echo-client.c
View File

@ -7,25 +7,27 @@
#include <sys/socket.h> #include <sys/socket.h>
#include <netinet/in.h> #include <netinet/in.h>
unsigned int ip2int(const char * s) { unsigned int ip2int(const char *s)
{
int ip[4]; int ip[4];
sscanf(s,"%d.%d.%d.%d",ip,ip+1,ip+2,ip+3); sscanf(s, "%d.%d.%d.%d", ip, ip + 1, ip + 2, ip + 3);
return ip[0]<<24|ip[1]<<16|ip[2]<<8|ip[3]; return ip[0] << 24 | ip[1] << 16 | ip[2] << 8 | ip[3];
} }
int sendall(int s, char *buf, int len, int flags) int sendall(int s, char *buf, int len, int flags)
{ {
int total = 0; int total = 0;
int n; int n;
while(total < len) while (total < len) {
{ n = send(s, buf + total, len - total, flags);
n = send(s, buf+total, len-total, flags); if (n == -1) {
if(n == -1) { break; } break;
total += n; }
} total += n;
}
return (n==-1 ? -1 : total); return (n == -1 ? -1 : total);
} }
char message[] = "Hello there!\n"; char message[] = "Hello there!\n";
@ -33,30 +35,28 @@ char buf[sizeof(message)];
int main() int main()
{ {
int sock; int sock;
struct sockaddr_in addr; struct sockaddr_in addr;
sock = socket(AF_INET, SOCK_STREAM, 0); sock = socket(AF_INET, SOCK_STREAM, 0);
if(sock < 0) if (sock < 0) {
{ perror("socket");
perror("socket"); exit(1);
exit(1); }
}
addr.sin_family = AF_INET; addr.sin_family = AF_INET;
addr.sin_port = htons(3425); // или любой другой порт... addr.sin_port = htons(3425); // или любой другой порт...
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); // (ip2int("192.168.2.1")); addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); // (ip2int("192.168.2.1"));
if(connect(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) if (connect(sock, (struct sockaddr *)&addr, sizeof(addr)) < 0) {
{ perror("connect");
perror("connect"); exit(2);
exit(2); }
}
send(sock, message, sizeof(message), 0); send(sock, message, sizeof(message), 0);
recv(sock, buf, sizeof(message), 0); recv(sock, buf, sizeof(message), 0);
printf("%s\n",buf); printf("%s\n", buf);
close(sock); close(sock);
return 0; return 0;
} }

332
c/masync_server/echo-server.c
View File

@ -7,59 +7,303 @@
#include <sys/socket.h> #include <sys/socket.h>
#include <netinet/in.h> #include <netinet/in.h>
static const char * int2ip(unsigned int ip) { #include <pthread.h>
#include <fcntl.h>
#include "xerror.h"
#define NUM_THREADS 8
pthread_mutex_t request_mutex = PTHREAD_MUTEX_INITIALIZER;
static const char *int2ip(unsigned int ip)
{
static char s[16]; static char s[16];
sprintf(s,"%d.%d.%d.%d",ip>>24,ip<<8>>24,ip<<16>>24,ip<<24>>24); sprintf(s, "%d.%d.%d.%d", ip >> 24, ip << 8 >> 24, ip << 16 >> 24, ip << 24 >> 24);
return s; return s;
} }
int main() void *serv_request(void *data)
{ {
int sock, listener; struct connection_cb {
struct sockaddr_in addr; int dataSocket;
char buf[1024]; char data[256];
int bytes_read; int dataSent;
int dataToSend;
int isReading;
struct connection_cb *next;
};
listener = socket(AF_INET, SOCK_STREAM, 0); struct connection_cb *connections = NULL;
if(listener < 0)
{
perror("socket");
exit(1);
}
addr.sin_family = AF_INET; int listenSocket = (int)data;
addr.sin_port = htons(3425); if (fcntl(listenSocket, F_SETFL, O_NONBLOCK) < 0)
addr.sin_addr.s_addr = htonl(INADDR_ANY); die("fcntl()", errno);
if(bind(listener, (struct sockaddr *)&addr, sizeof(addr)) < 0)
{
perror("bind");
exit(2);
}
listen(listener, 1); while (1) {
//~ printf("%X\n",addr.sin_addr.s_addr); fd_set readFdSet;
while(1) fd_set writeFdSet;
{ struct connection_cb *currentConn, **currentConnPtr, *tempConn;
socklen_t n = sizeof(struct sockaddr); int maxFdNum;
sock = accept(listener, (struct sockaddr *)&addr, &n);
printf("incoming request from %s\n",int2ip(ntohl(addr.sin_addr.s_addr)));
//~ printf("%X\n",addr.sin_addr.s_addr);
if(sock < 0)
{
perror("accept");
exit(3);
}
while(1) FD_ZERO(&readFdSet);
{ FD_ZERO(&writeFdSet);
bytes_read = recv(sock, buf, 1024, 0);
if(bytes_read <= 0) break;
send(sock, buf, bytes_read, 0);
printf("%s", buf);
}
close(sock); /*
} Добавление дескриптора к множеству readFdSet
*/
FD_SET(listenSocket, &readFdSet);
maxFdNum = listenSocket;
return 0; for (currentConn = connections; currentConn != NULL; currentConn = currentConn->next) {
if (currentConn->isReading)
FD_SET(currentConn->dataSocket, &readFdSet);
else
FD_SET(currentConn->dataSocket, &writeFdSet);
maxFdNum = currentConn->dataSocket > maxFdNum ? currentConn->dataSocket : maxFdNum;
}
/*
Получение множества дескрипторов сокетов для обработки
*/
if (select(maxFdNum + 1, &readFdSet, &writeFdSet, NULL, NULL) < 0) {
if (errno == EINTR)
continue;
die("select()", errno);
}
currentConnPtr = &connections;
while (*currentConnPtr != NULL) {
/*
Проверка принадлежности дескриптора
(*currentConnPtr)->dataSocket к множеству readFdSet
*/
if ((*currentConnPtr)->isReading && FD_ISSET((*currentConnPtr)->dataSocket, &readFdSet)) {
int result = recv((*currentConnPtr)->dataSocket, (*currentConnPtr)->data,
sizeof((*currentConnPtr)->data), 0);
if (result < 0) {
if (errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK) {
bark("recv()", errno);
close((*currentConnPtr)->dataSocket);
tempConn = *currentConnPtr;
*currentConnPtr = (*currentConnPtr)->next;
free(tempConn);
continue;
}
} else if (result == 0) {
close((*currentConnPtr)->dataSocket);
tempConn = *currentConnPtr;
*currentConnPtr = (*currentConnPtr)->next;
free(tempConn);
continue;
} else {
(*currentConnPtr)->dataToSend = result;
(*currentConnPtr)->dataSent = 0;
(*currentConnPtr)->isReading = 0;
printf("Recieving as Slave Thread id = '%d' \n", pthread_self());
}
} else
/*
Проверка принадлежности дескриптора
(*currentConnPtr)->dataSocket к множеству writedFdSet
*/
if (FD_ISSET((*currentConnPtr)->dataSocket, &writeFdSet)) {
int result = send((*currentConnPtr)->dataSocket,
(*currentConnPtr)->data + (*currentConnPtr)->dataSent,
(*currentConnPtr)->dataToSend - (*currentConnPtr)->dataSent, 0);
if (result < 0) {
if (errno != EINTR && errno != EAGAIN) {
bark("write()", errno);
close((*currentConnPtr)->dataSocket);
tempConn = *currentConnPtr;
*currentConnPtr = (*currentConnPtr)->next;
free(tempConn);
continue;
}
} else {
(*currentConnPtr)->dataSent += result;
if ((*currentConnPtr)->dataSent >= (*currentConnPtr)->dataToSend)
(*currentConnPtr)->isReading = 1;
}
}
currentConnPtr = &((*currentConnPtr)->next);
printf("Sending as Slave Thread id = '%d' \n", pthread_self());
}
/*
Проверка принадлежности дескриптора listenSocket
к множеству readFdSet,т.е. необходимости обработать
вызов connect( ) от нового клиента.
*/
if (FD_ISSET(listenSocket, &readFdSet)) {
while (1) {
/*
Вызовы pthread_mutex_lock, pthread_mutex_unlock
Не нужны в среде Linux
*/
pthread_mutex_lock(&request_mutex);
int result = accept(listenSocket, (struct sockaddr *)NULL, NULL);
pthread_mutex_unlock(&request_mutex);
if (result < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK)
break;
die("accept()", errno);
} else {
*currentConnPtr = malloc(sizeof(struct connection_cb));
if (*currentConnPtr == NULL)
die("malloc()", 0);
if (fcntl(result, F_SETFL, O_NONBLOCK) < 0)
die("fcntl()", errno);
(*currentConnPtr)->dataSocket = result;
(*currentConnPtr)->isReading = 1;
(*currentConnPtr)->next = 0;
currentConnPtr = &((*currentConnPtr)->next);
printf("Accepting as Master Thread id = '%d' \n", pthread_self());
}
}
}
}
} }
//~ void *serv_request(void *data)
//~ {
//~ struct connection_cb
//~ {
//~ int dataSocket;
//~ char data[256];
//~ int dataSent;
//~ int dataToSend;
//~ int isRading;
//~ struct connection_cb *next;
//~ };
//~
//~ struct connection_cb *connections = NULL;
//~
//~ int listenSocket = (int)data;
//~
//~ if(fcntl(listenSocket, F_SETFL, O_NONBLOCK) < 0)
//~ die("fcntl()", errno);
//~
//~
//~
//~ }
int getServerSocket(unsigned short int port)
{
int listenSocket;
struct sockaddr_in listenSockaddr;
if ((listenSocket = socket(PF_INET, SOCK_STREAM, 0)) < 0)
die("socket()", errno);
memset(&listenSockaddr, 0, sizeof(listenSockaddr));
listenSockaddr.sin_family = PF_INET;
listenSockaddr.sin_port = htons(port);
listenSockaddr.sin_addr.s_addr = INADDR_ANY;
if (bind(listenSocket, (struct sockaddr *)&listenSockaddr, sizeof(listenSockaddr)) < 0)
die("bind()", errno);
if (listen(listenSocket, 5) < 0)
die("listen()", errno);
return listenSocket;
}
int main(int argc, char *argv[])
{
int k;
int descSock;
char *service = "1500";
switch (argc) {
case 1:
break;
case 2:
service = argv[1];
break;
default:
printf("Usage: ./echo-server [port]\n");
exit(1);
}
pthread_t p_thread[NUM_THREADS];
pthread_attr_t attr;
pthread_attr_init(&attr);
size_t stacksize = 512;
pthread_attr_setstacksize(&attr, stacksize);
pthread_attr_getstacksize(&attr, &stacksize);
descSock = getServerSocket(atoi(service));
for (k = 0; k < NUM_THREADS; k++) {
pthread_create(&p_thread[k], &attr, serv_request, (void *)descSock);
printf("Thread %d started\n", k);
}
pthread_attr_destroy(&attr);
for (k = 0; k < NUM_THREADS; k++) {
pthread_join(p_thread[k], NULL);
printf("Completed join with thread %d\n", k);
}
}
//~ int tmp()
//~ {
//~ int sock, listener;
//~ struct sockaddr_in addr;
//~ char buf[1024];
//~ int bytes_read;
//~
//~ listener = socket(AF_INET, SOCK_STREAM, 0);
//~ if(listener < 0)
//~ {
//~ perror("socket");
//~ exit(1);
//~ }
//~
//~ addr.sin_family = AF_INET;
//~ addr.sin_port = htons(3425);
//~ addr.sin_addr.s_addr = htonl(INADDR_ANY);
//~ if(bind(listener, (struct sockaddr *)&addr, sizeof(addr)) < 0)
//~ {
//~ perror("bind");
//~ exit(2);
//~ }
//~
//~ listen(listener, 1);
//~ //printf("%X\n",addr.sin_addr.s_addr);
//~ while(1)
//~ {
//~ socklen_t n = sizeof(struct sockaddr);
//~ sock = accept(listener, (struct sockaddr *)&addr, &n);
//~ printf("incoming request from %s\n",int2ip(ntohl(addr.sin_addr.s_addr)));
//~ //printf("%X\n",addr.sin_addr.s_addr);
//~ if(sock < 0)
//~ {
//~ perror("accept");
//~ exit(3);
//~ }
//~
//~ while(1)
//~ {
//~ bytes_read = recv(sock, buf, 1024, 0);
//~ if(bytes_read <= 0) break;
//~ send(sock, buf, bytes_read, 0);
//~ printf("%s", buf);
//~ }
//~
//~ close(sock);
//~ }
//~
//~ return 0;
//~ }

14
c/masync_server/int2ip.c
View File

@ -12,22 +12,24 @@
#include <netdb.h> #include <netdb.h>
static const char * int2ip(unsigned int ip) { static const char *int2ip(unsigned int ip)
{
static char s[16]; static char s[16];
sprintf(s,"%d.%d.%d.%d",ip>>24,ip<<8>>24,ip<<16>>24,ip<<24>>24); sprintf(s, "%d.%d.%d.%d", ip >> 24, ip << 8 >> 24, ip << 16 >> 24, ip << 24 >> 24);
return s; return s;
} }
int main(int argc, char * argv[]) { int main(int argc, char *argv[])
if(argc < 2) { {
if (argc < 2) {
perror("need hex ip address"); perror("need hex ip address");
exit(1); exit(1);
} }
unsigned int ip = 0; unsigned int ip = 0;
sscanf(argv[1],"%X",&ip); sscanf(argv[1], "%X", &ip);
printf("%s\n",int2ip(ip)); printf("%s\n", int2ip(ip));
return 0; return 0;
} }

14
c/masync_server/ip2int.c
View File

@ -12,19 +12,21 @@
#include <netdb.h> #include <netdb.h>
unsigned int ip2int(const char * s) { unsigned int ip2int(const char *s)
{
int ip[4]; int ip[4];
sscanf(s,"%d.%d.%d.%d",ip,ip+1,ip+2,ip+3); sscanf(s, "%d.%d.%d.%d", ip, ip + 1, ip + 2, ip + 3);
return ip[0]<<24|ip[1]<<16|ip[2]<<8|ip[3]; return ip[0] << 24 | ip[1] << 16 | ip[2] << 8 | ip[3];
} }
int main(int argc, char * argv[]) { int main(int argc, char *argv[])
if(argc < 2) { {
if (argc < 2) {
perror("need ip address string"); perror("need ip address string");
exit(1); exit(1);
} }
printf("%X\n",ip2int(argv[1])); printf("%X\n", ip2int(argv[1]));
return 0; return 0;
} }

12
c/masync_server/sock1.c
View File

@ -38,9 +38,9 @@
//~ //~
//~ while(total < len) //~ while(total < len)
//~ { //~ {
//~ n = send(s, buf+total, len-total, flags); //~ n = send(s, buf+total, len-total, flags);
//~ if(n == -1) { break; } //~ if(n == -1) { break; }
//~ total += n; //~ total += n;
//~ } //~ }
//~ //~
//~ return (n==-1 ? -1 : total); //~ return (n==-1 ? -1 : total);
@ -54,11 +54,9 @@
// запрет передачи 0 - чтения, 1 - записи, 2 - и того, и др. // запрет передачи 0 - чтения, 1 - записи, 2 - и того, и др.
//~ int shutdown(int sockfd, int how); //~ int shutdown(int sockfd, int how);
int main(int argc, char *argv[])
int main(int argc, char * argv[])
{ {
struct a struct a {
{
char aaa[256]; char aaa[256];
}; };

12
c/masync_server/tmp1.c
View File

@ -3,21 +3,21 @@
void f() void f()
{ {
int c = 0; int c = 0;
printf("c=%lu\n",(unsigned long)&c); printf("c=%lu\n", (unsigned long)&c);
} }
int main(int argc, char *argv[])
int main(int argc, char * argv[]) { {
//~ printf("%lu\n", sizeof(long)); //~ printf("%lu\n", sizeof(long));
int a = 0; int a = 0;
int pid = fork(); int pid = fork();
printf("pid=%d\n",pid); printf("pid=%d\n", pid);
printf("a=%lu\n",(unsigned long)&a); printf("a=%lu\n", (unsigned long)&a);
int b = 0; int b = 0;
printf("b=%lu\n",(unsigned long)&b); printf("b=%lu\n", (unsigned long)&b);
f(); f();