src/parsec/disk-image/parsec/parsec-benchmark/pkgs/netapps/netferret/src/client/client.c - public/gem5-resources - Git at Google

 /* AUTORIGHTS
 Copyright (C) 2007 Princeton University

 This file is part of Ferret Toolkit.

 Ferret Toolkit is free software; you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation; either version 2, or (at your option)
 any later version.

 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.

 You should have received a copy of the GNU General Public License
 along with this program; if not, write to the Free Software Foundation,
 Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
 */
 #include <stdio.h>
 #include <stdlib.h>
 #include <assert.h>
 #include <dirent.h>
 #include <string.h>
 #include <sys/types.h>
 #include <sys/stat.h>
 #include <unistd.h>
 #include <pthread.h>

 #include "tpool.h"
 #include "queue.h"

 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <netdb.h>
 #include <errno.h>
 #include <fcntl.h>
 #ifdef ENABLE_PARSEC_UPTCPIP
 #include <uptcp_socket.h>
 #endif

 #define PORT       42285
 /* REPLACE with your server machine name*/
 #define SERVER_IP       0x0a0a0a0c //"10.10.10.12"
 //#define SERVER_IP       0x7f000001 //"127.0.0.1"

 #define MAX_THREADS		16
 #define MAX_SOCKET_BUF		0x4000

 #define DEFAULT_DEPTH	25

 const char *query_dir = NULL;
 const char *output_path = NULL;

 FILE *fout;
 pthread_mutex_t fout_lock = PTHREAD_MUTEX_INITIALIZER;


 int NTHREAD_LOAD = 1;
 int NTHREAD_PROCESS	= 1;
 int DEPTH = DEFAULT_DEPTH;

 struct load_data
 {
 	char *name;
 };

 struct queue q_load_process;


 struct packet{
 	int	len;
 	char	buf[1];
 };

 /* ------- The Helper Functions ------- */
 int cnt_enqueue;
 int cnt_dequeue;
 char path[BUFSIZ];


 int scan_dir (const char *, char *head);

 int dir_helper (char *dir, char *head)
 {
 	DIR *pd = NULL;
 	struct dirent *ent = NULL;
 	int result = 0;
 	pd = opendir(dir);
 	if (pd == NULL) goto except;
 	for (;;)
 	{
 		ent = readdir(pd);
 		if (ent == NULL) break;
 		if (scan_dir(ent->d_name, head) != 0) return -1;
 	}
 	goto final;

 except:
 	result = -1;
 	perror("Error:");
 final:
 	if (pd != NULL) closedir(pd);
 	return result;
 }

 /* the whole path to the file */
 int file_helper (const char *file)
 {
 	int r;
 	struct load_data *data;

 	data = (struct load_data *)malloc(sizeof(struct load_data));
 	assert(data != NULL);

 	data->name = strdup(file);

 	cnt_enqueue++;
 	enqueue(&q_load_process, data);

 	return 0;
 }

 int scan_dir (const char *dir, char *head)
 {
 	struct stat st;
 	int ret;
 	/* test for . and .. */
 	if (dir[0] == '.')
 	{
 		if (dir[1] == 0) return 0;
 		else if (dir[1] == '.')
 		{
 			if (dir[2] == 0) return 0;
 		}
 	}

 	/* append the name to the path */
 	strcat(head, dir);
 	ret = stat(path, &st);
 	if (ret != 0)
 	{
 		perror("Error:");
 		return -1;
 	}
 	if (S_ISREG(st.st_mode)) file_helper(path);
 	else if (S_ISDIR(st.st_mode))
 	{
 		strcat(head, "/");
 		dir_helper(path, head + strlen(head));
 	}
 	/* removed the appended part */
 	head[0] = 0;
 	return 0;
 }


 /* ------ The Stages ------ */
 void *t_load (void *dummy)
 {
 	const char *dir = (const char *)dummy;

 	path[0] = 0;

 	if (strcmp(dir, ".") == 0)
 	{
 		dir_helper(".", path);
 	}
 	else
 	{
 		scan_dir(dir, path);
 	}

 	queue_signal_terminate(&q_load_process);
 	return NULL;
 }

 void send_end_tag()
 {
 	int		 fd;
         struct sockaddr_in      sin;
         struct sockaddr_in      pin;
         struct hostent          *hp;
 	int		end_tag;;

         memset(&pin, 0, sizeof(pin));
        	pin.sin_family = AF_INET;
         pin.sin_addr.s_addr = htonl(SERVER_IP);
        	pin.sin_port = htons(PORT);

         /* grab an Internet domain socket */
 #ifdef ENABLE_PARSEC_UPTCPIP
         if ((fd = uptcp_socket(AF_INET, SOCK_STREAM, 0)) == -1) {
 #else
         if ((fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
 #endif
        	        fprintf(stderr, "Socket error: socket()\n");
         }

         /* connect to PORT on HOST */
 #ifdef ENABLE_PARSEC_UPTCPIP
         if (uptcp_connect(fd,(struct sockaddr *)  &pin, sizeof(pin)) < 0) {
 #else
         if (connect(fd,(struct sockaddr *)  &pin, sizeof(pin)) < 0) {
 #endif
        	        fprintf(stderr, "Socket error: connect()\n");
         }

 	end_tag = -1;
 #ifdef ENABLE_PARSEC_UPTCPIP
         if (uptcp_send(fd, &end_tag, sizeof(int), 0) == -1) {
 #else
         if (send(fd, &end_tag, sizeof(int), 0) == -1) {
 #endif
 		fprintf(stderr, "Socket error: send input file data error\n");
         }

 }


 void *t_process (void *dummy)
 {
 	struct load_data *load;
 	struct packet    *pkt_ptr;
 	char		 *buf_ptr;
 	int		 fd;
         struct sockaddr_in      sin;
         struct sockaddr_in      pin;
         struct hostent          *hp;
         struct stat filestat;


 	pkt_ptr = (struct packet*)malloc(MAX_SOCKET_BUF);
 	if(pkt_ptr == NULL)
 		fprintf(stderr, "allocate packet buffer error\n");

 	while(1){
 		/* 1. dequeue */
 		if(dequeue(&q_load_process, &load) < 0)
 			break;

 		assert(load != NULL);

 		/* 2. read data */

 	        /* src file stat */
 	        if (stat(load->name, &filestat) < 0)
 	       	      fprintf(stderr, "stat() %s failed: %s\n",load->name, strerror(errno));

 	        if (!S_ISREG(filestat.st_mode))
        		    fprintf(stderr, "not a normal file: %s\n", load->name);

 		pkt_ptr->len = filestat.st_size;

         	/* src file open */
 	        if((fd = open(load->name, O_RDONLY)) < 0)
         	    fprintf(stderr, "%s file open error %s\n", load->name, strerror(errno));

 	        size_t bytes_read = 0;
         	int r;
 		buf_ptr = pkt_ptr->buf;
 	        /* Read data until buffer full */
 	        while(bytes_read < pkt_ptr->len) {
         		r = read(fd, buf_ptr, pkt_ptr->len - bytes_read);
 	                if(r<0)
 				fprintf(stderr, "read data from file error\n");

 			if(r==0)
 				break;

 			bytes_read += r;
 			buf_ptr += r;
 	        }
 		close(fd);

 		/* 3. send request to server */
 	        memset(&pin, 0, sizeof(pin));
         	pin.sin_family = AF_INET;
 	        pin.sin_addr.s_addr = htonl(SERVER_IP);
         	pin.sin_port = htons(PORT);

 	        /* grab an Internet domain socket */
 #ifdef ENABLE_PARSEC_UPTCPIP
 	        if ((fd = uptcp_socket(AF_INET, SOCK_STREAM, 0)) == -1) {
 #else
 	        if ((fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
 #endif
         	        fprintf(stderr, "Socket error: socket()\n");
 	        }

 	        /* connect to PORT on HOST */
 #ifdef ENABLE_PARSEC_UPTCPIP
 	        if (uptcp_connect(fd,(struct sockaddr *)  &pin, sizeof(pin)) < 0) {
 #else
 	        if (connect(fd,(struct sockaddr *)  &pin, sizeof(pin)) < 0) {
 #endif
         	        fprintf(stderr, "Socket error: connect()\n");
 	        }

 	        /* Send data to server*/
 	        int bytes_left = pkt_ptr->len + sizeof(int);
 	        int bytes_tosend = 0;
         	buf_ptr = (char*)pkt_ptr;
 	        while(bytes_left >0){
         	        bytes_tosend = bytes_left;
 #ifdef ENABLE_PARSEC_UPTCPIP
 	                if (uptcp_send(fd, buf_ptr, bytes_tosend, 0) == -1) {
 #else
 	                if (send(fd, buf_ptr, bytes_tosend, 0) == -1) {
 #endif
 	                        fprintf(stderr, "Socket error: send input file data error\n");
 	                }
         	        bytes_left -= bytes_tosend;
 	                buf_ptr += bytes_tosend;
         	}


 		/* 4. wait for results */
 	        int bytes_recv = 0;
 		int bytes_input = MAX_SOCKET_BUF;

 	        buf_ptr = (char*)pkt_ptr;
 	        while(bytes_recv < bytes_input) {
 #ifdef ENABLE_PARSEC_UPTCPIP
 			if ((r = uptcp_recv(fd, buf_ptr+bytes_recv, (bytes_input-bytes_recv), 0)) == -1) {
 #else
 			if ((r = recv(fd, buf_ptr+bytes_recv, (bytes_input-bytes_recv), 0)) == -1) {
 #endif
 				fprintf(stderr, "recv error\n\n");
 			}

 			if(bytes_input == MAX_SOCKET_BUF){
 				bytes_input = pkt_ptr->len;
 			}

 			bytes_recv += r;

 			if(r == 0)
 				break;
 			bytes_recv += r;
 		}

 		/* 5. write results */
 		pthread_mutex_lock(&fout_lock);

 		fwrite(load->name, 1, strlen(load->name), fout);
 		fwrite(pkt_ptr->buf, 1, pkt_ptr->len, fout);
 		fwrite("\n", 1, 1, fout);

 		cnt_dequeue++;

 		if(cnt_enqueue == cnt_dequeue){ /* last item */
 			send_end_tag();
 		}

 		pthread_mutex_unlock(&fout_lock);

 		free(load->name);
 		free(load);

 #ifdef ENABLE_PARSEC_UPTCPIP
 		uptcp_close(fd);
 #else
 		close(fd);
 #endif
 	}

 	free(pkt_ptr);
 	return NULL;

 }

 int main (int argc, char *argv[])
 {
 	tdesc_t *t_load_desc;
 	tdesc_t *t_process_desc;

 	tpool_t *p_load;
 	tpool_t *p_process;

 	int ret, i;

 	if (argc < 4)
 	{
 		printf("%s <query dir> <n> <out>\n", argv[0]);
 		return 0;
 	}

 	query_dir = argv[1];
 	NTHREAD_PROCESS = atoi(argv[2]);

 	output_path = argv[3];

 	fout = fopen(output_path, "w");
 	assert(fout != NULL);

 	queue_init(&q_load_process, DEPTH, NTHREAD_LOAD);

 	t_load_desc = (tdesc_t *)calloc(NTHREAD_LOAD, sizeof(tdesc_t));
 	t_process_desc = (tdesc_t *)calloc(NTHREAD_PROCESS, sizeof(tdesc_t));

 	t_load_desc[0].attr = NULL;
 	t_load_desc[0].start_routine = t_load;
 	t_load_desc[0].arg = query_dir;

 	for (i = 1; i < NTHREAD_LOAD; i++) t_load_desc[i] = t_load_desc[0];

 	t_process_desc[0].attr = NULL;
 	t_process_desc[0].start_routine = t_process;
 	t_process_desc[0].arg = NULL;

 	for (i = 1; i < NTHREAD_PROCESS; i++) t_process_desc[i] = t_process_desc[0];

 	cnt_enqueue = cnt_dequeue = 0;

 	p_load = tpool_create(t_load_desc, NTHREAD_LOAD);
 	p_process = tpool_create(t_process_desc, NTHREAD_PROCESS);

 	tpool_join(p_process, NULL);
 	tpool_join(p_load, NULL);

 	tpool_destroy(p_load);
 	tpool_destroy(p_process);

 	free(t_load_desc);
 	free(t_process_desc);

 	queue_destroy(&q_load_process);

 	fclose(fout);

 	return 0;
 }
	/* AUTORIGHTS
	Copyright (C) 2007 Princeton University

	This file is part of Ferret Toolkit.

	Ferret Toolkit is free software; you can redistribute it and/or modify
	it under the terms of the GNU General Public License as published by
	the Free Software Foundation; either version 2, or (at your option)
	any later version.

	This program is distributed in the hope that it will be useful,
	but WITHOUT ANY WARRANTY; without even the implied warranty of
	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	GNU General Public License for more details.

	You should have received a copy of the GNU General Public License
	along with this program; if not, write to the Free Software Foundation,
	Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
	*/
	#include <stdio.h>
	#include <stdlib.h>
	#include <assert.h>
	#include <dirent.h>
	#include <string.h>
	#include <sys/types.h>
	#include <sys/stat.h>
	#include <unistd.h>
	#include <pthread.h>

	#include "tpool.h"
	#include "queue.h"

	#include <sys/socket.h>
	#include <netinet/in.h>
	#include <netdb.h>
	#include <errno.h>
	#include <fcntl.h>
	#ifdef ENABLE_PARSEC_UPTCPIP
	#include <uptcp_socket.h>
	#endif

	#define PORT 42285
	/* REPLACE with your server machine name*/
	#define SERVER_IP 0x0a0a0a0c //"10.10.10.12"
	//#define SERVER_IP 0x7f000001 //"127.0.0.1"

	#define MAX_THREADS 16
	#define MAX_SOCKET_BUF 0x4000

	#define DEFAULT_DEPTH 25

	const char *query_dir = NULL;
	const char *output_path = NULL;

	FILE *fout;
	pthread_mutex_t fout_lock = PTHREAD_MUTEX_INITIALIZER;


	int NTHREAD_LOAD = 1;
	int NTHREAD_PROCESS = 1;
	int DEPTH = DEFAULT_DEPTH;

	struct load_data
	{
	char *name;
	};

	struct queue q_load_process;


	struct packet{
	int len;
	char buf[1];
	};

	/* ------- The Helper Functions ------- */
	int cnt_enqueue;
	int cnt_dequeue;
	char path[BUFSIZ];


	int scan_dir (const char , char head);

	int dir_helper (char dir, char head)
	{
	DIR *pd = NULL;
	struct dirent *ent = NULL;
	int result = 0;
	pd = opendir(dir);
	if (pd == NULL) goto except;
	for (;;)
	{
	ent = readdir(pd);
	if (ent == NULL) break;
	if (scan_dir(ent->d_name, head) != 0) return -1;
	}
	goto final;

	except:
	result = -1;
	perror("Error:");
	final:
	if (pd != NULL) closedir(pd);
	return result;
	}

	/* the whole path to the file */
	int file_helper (const char *file)
	{
	int r;
	struct load_data *data;

	data = (struct load_data *)malloc(sizeof(struct load_data));
	assert(data != NULL);

	data->name = strdup(file);

	cnt_enqueue++;
	enqueue(&q_load_process, data);

	return 0;
	}

	int scan_dir (const char dir, char head)
	{
	struct stat st;
	int ret;
	/* test for . and .. */
	if (dir[0] == '.')
	{
	if (dir[1] == 0) return 0;
	else if (dir[1] == '.')
	{
	if (dir[2] == 0) return 0;
	}
	}

	/* append the name to the path */
	strcat(head, dir);
	ret = stat(path, &st);
	if (ret != 0)
	{
	perror("Error:");
	return -1;
	}
	if (S_ISREG(st.st_mode)) file_helper(path);
	else if (S_ISDIR(st.st_mode))
	{
	strcat(head, "/");
	dir_helper(path, head + strlen(head));
	}
	/* removed the appended part */
	head[0] = 0;
	return 0;
	}



	/* ------ The Stages ------ */
	void t_load (void dummy)
	{
	const char dir = (const char )dummy;

	path[0] = 0;

	if (strcmp(dir, ".") == 0)
	{
	dir_helper(".", path);
	}
	else
	{
	scan_dir(dir, path);
	}

	queue_signal_terminate(&q_load_process);
	return NULL;
	}

	void send_end_tag()
	{
	int fd;
	struct sockaddr_in sin;
	struct sockaddr_in pin;
	struct hostent *hp;
	int end_tag;;

	memset(&pin, 0, sizeof(pin));
	pin.sin_family = AF_INET;
	pin.sin_addr.s_addr = htonl(SERVER_IP);
	pin.sin_port = htons(PORT);

	/* grab an Internet domain socket */
	#ifdef ENABLE_PARSEC_UPTCPIP
	if ((fd = uptcp_socket(AF_INET, SOCK_STREAM, 0)) == -1) {
	#else
	if ((fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
	#endif
	fprintf(stderr, "Socket error: socket()\n");
	}

	/* connect to PORT on HOST */
	#ifdef ENABLE_PARSEC_UPTCPIP
	if (uptcp_connect(fd,(struct sockaddr *) &pin, sizeof(pin)) < 0) {
	#else
	if (connect(fd,(struct sockaddr *) &pin, sizeof(pin)) < 0) {
	#endif
	fprintf(stderr, "Socket error: connect()\n");
	}

	end_tag = -1;
	#ifdef ENABLE_PARSEC_UPTCPIP
	if (uptcp_send(fd, &end_tag, sizeof(int), 0) == -1) {
	#else
	if (send(fd, &end_tag, sizeof(int), 0) == -1) {
	#endif
	fprintf(stderr, "Socket error: send input file data error\n");
	}

	}


	void t_process (void dummy)
	{
	struct load_data *load;
	struct packet *pkt_ptr;
	char *buf_ptr;
	int fd;
	struct sockaddr_in sin;
	struct sockaddr_in pin;
	struct hostent *hp;
	struct stat filestat;


	pkt_ptr = (struct packet*)malloc(MAX_SOCKET_BUF);
	if(pkt_ptr == NULL)
	fprintf(stderr, "allocate packet buffer error\n");

	while(1){
	/* 1. dequeue */
	if(dequeue(&q_load_process, &load) < 0)
	break;

	assert(load != NULL);

	/* 2. read data */

	/* src file stat */
	if (stat(load->name, &filestat) < 0)
	fprintf(stderr, "stat() %s failed: %s\n",load->name, strerror(errno));

	if (!S_ISREG(filestat.st_mode))
	fprintf(stderr, "not a normal file: %s\n", load->name);

	pkt_ptr->len = filestat.st_size;

	/* src file open */
	if((fd = open(load->name, O_RDONLY)) < 0)
	fprintf(stderr, "%s file open error %s\n", load->name, strerror(errno));

	size_t bytes_read = 0;
	int r;
	buf_ptr = pkt_ptr->buf;
	/* Read data until buffer full */
	while(bytes_read < pkt_ptr->len) {
	r = read(fd, buf_ptr, pkt_ptr->len - bytes_read);
	if(r<0)
	fprintf(stderr, "read data from file error\n");

	if(r==0)
	break;

	bytes_read += r;
	buf_ptr += r;
	}
	close(fd);

	/* 3. send request to server */
	memset(&pin, 0, sizeof(pin));
	pin.sin_family = AF_INET;
	pin.sin_addr.s_addr = htonl(SERVER_IP);
	pin.sin_port = htons(PORT);

	/* grab an Internet domain socket */
	#ifdef ENABLE_PARSEC_UPTCPIP
	if ((fd = uptcp_socket(AF_INET, SOCK_STREAM, 0)) == -1) {
	#else
	if ((fd = socket(AF_INET, SOCK_STREAM, 0)) == -1) {
	#endif
	fprintf(stderr, "Socket error: socket()\n");
	}

	/* connect to PORT on HOST */
	#ifdef ENABLE_PARSEC_UPTCPIP
	if (uptcp_connect(fd,(struct sockaddr *) &pin, sizeof(pin)) < 0) {
	#else
	if (connect(fd,(struct sockaddr *) &pin, sizeof(pin)) < 0) {
	#endif
	fprintf(stderr, "Socket error: connect()\n");
	}

	/* Send data to server*/
	int bytes_left = pkt_ptr->len + sizeof(int);
	int bytes_tosend = 0;
	buf_ptr = (char*)pkt_ptr;
	while(bytes_left >0){
	bytes_tosend = bytes_left;
	#ifdef ENABLE_PARSEC_UPTCPIP
	if (uptcp_send(fd, buf_ptr, bytes_tosend, 0) == -1) {
	#else
	if (send(fd, buf_ptr, bytes_tosend, 0) == -1) {
	#endif
	fprintf(stderr, "Socket error: send input file data error\n");
	}
	bytes_left -= bytes_tosend;
	buf_ptr += bytes_tosend;
	}


	/* 4. wait for results */
	int bytes_recv = 0;
	int bytes_input = MAX_SOCKET_BUF;

	buf_ptr = (char*)pkt_ptr;
	while(bytes_recv < bytes_input) {
	#ifdef ENABLE_PARSEC_UPTCPIP
	if ((r = uptcp_recv(fd, buf_ptr+bytes_recv, (bytes_input-bytes_recv), 0)) == -1) {
	#else
	if ((r = recv(fd, buf_ptr+bytes_recv, (bytes_input-bytes_recv), 0)) == -1) {
	#endif
	fprintf(stderr, "recv error\n\n");
	}

	if(bytes_input == MAX_SOCKET_BUF){
	bytes_input = pkt_ptr->len;
	}

	bytes_recv += r;

	if(r == 0)
	break;
	bytes_recv += r;
	}

	/* 5. write results */
	pthread_mutex_lock(&fout_lock);

	fwrite(load->name, 1, strlen(load->name), fout);
	fwrite(pkt_ptr->buf, 1, pkt_ptr->len, fout);
	fwrite("\n", 1, 1, fout);

	cnt_dequeue++;

	if(cnt_enqueue == cnt_dequeue){ /* last item */
	send_end_tag();
	}

	pthread_mutex_unlock(&fout_lock);

	free(load->name);
	free(load);

	#ifdef ENABLE_PARSEC_UPTCPIP
	uptcp_close(fd);
	#else
	close(fd);
	#endif
	}

	free(pkt_ptr);
	return NULL;

	}

	int main (int argc, char *argv[])
	{
	tdesc_t *t_load_desc;
	tdesc_t *t_process_desc;

	tpool_t *p_load;
	tpool_t *p_process;

	int ret, i;

	if (argc < 4)
	{
	printf("%s <query dir> <n> <out>\n", argv[0]);
	return 0;
	}

	query_dir = argv[1];
	NTHREAD_PROCESS = atoi(argv[2]);

	output_path = argv[3];

	fout = fopen(output_path, "w");
	assert(fout != NULL);

	queue_init(&q_load_process, DEPTH, NTHREAD_LOAD);

	t_load_desc = (tdesc_t *)calloc(NTHREAD_LOAD, sizeof(tdesc_t));
	t_process_desc = (tdesc_t *)calloc(NTHREAD_PROCESS, sizeof(tdesc_t));

	t_load_desc[0].attr = NULL;
	t_load_desc[0].start_routine = t_load;
	t_load_desc[0].arg = query_dir;

	for (i = 1; i < NTHREAD_LOAD; i++) t_load_desc[i] = t_load_desc[0];

	t_process_desc[0].attr = NULL;
	t_process_desc[0].start_routine = t_process;
	t_process_desc[0].arg = NULL;

	for (i = 1; i < NTHREAD_PROCESS; i++) t_process_desc[i] = t_process_desc[0];

	cnt_enqueue = cnt_dequeue = 0;

	p_load = tpool_create(t_load_desc, NTHREAD_LOAD);
	p_process = tpool_create(t_process_desc, NTHREAD_PROCESS);

	tpool_join(p_process, NULL);
	tpool_join(p_load, NULL);

	tpool_destroy(p_load);
	tpool_destroy(p_process);

	free(t_load_desc);
	free(t_process_desc);

	queue_destroy(&q_load_process);

	fclose(fout);

	return 0;
	}