gnbdd.c

openGFS , a kind of file system.

	return setsockopt(sk, SOL_SOCKET, SO_KEEPALIVE, &iopt, sizeof(int));
}

/******************************************************************************
 * loop()
 * the actual main loop of this thing
 *
 * Try using select this time.
 */
void *
loop(void *args)
{
	int *t = (int *) args;
	int listenfd = *t;
	int i, maxi, clifd, maxfd, nread;
	int err = 0;
	theader_t ioHdr;
	struct sockaddr_in theiraddr;
	fd_set rset, allset;
	pid_t mypid;

	mypid = getpid();	/* mostly for debugging */

	FD_ZERO(&allset);

	FD_SET(listenfd, &allset);
	maxfd = listenfd;
	maxi = -1;

	while (Server_Runs) {
		rset = allset;
		if ((nread = select(maxfd + 1, &rset, NULL, NULL, NULL)) < 0) {
			perror("loop: select error");
			exit(1);
		}

		/* If someone else is doing an accept, just skip over it. */
		if (pthread_mutex_trylock(&AcceptMux) == 0) {
			if (FD_ISSET(listenfd, &rset)) {
				int len;
				len = sizeof(struct sockaddr_in);
				if ((clifd = accept(listenfd,
						    (struct sockaddr *)
						    &theiraddr, &len)) < 0) {
					perror("loop: error in accept");
					exit(1);
				}
				if (setkeepalive(clifd) < 0) {
					perror
					    ("loop: setkeepalive failed (ignoring)");
				}
				i = client_add(clifd);
				FD_SET(clifd, &allset);
				if (clifd > maxfd)
					maxfd = clifd;
				if (i > maxi)
					maxi = i;

				printf("Added new client[%d] fd=%d\n", i,
				       clifd);
				{	/*send login info */
					tio_info_t serv_stat;
					serv_stat.devsize = htonll(datasize);
					serv_stat.readonly = readonly;
					nread =
					    cpt_send(clifd, &serv_stat,
						     sizeof(tio_info_t), 0);
				}
			}
			pthread_mutex_unlock(&AcceptMux);
		}

		for (i = 1; i <= maxi; i++) {
			if ((clifd = Clients[i].fd) < 0)
				continue;
			if (FD_ISSET(clifd, &rset)) {
				if (pthread_mutex_trylock(&Clients[i].mux) == 0) {
#if 0
					printf
					    ("thread %d: got Clients[%d].mux\n",
					     mypid, i);
#endif
					if ((nread =
					     cpt_recv(clifd, &ioHdr,
						      sizeof(theader_t),
						      0)) < 0)
						perror("loop: Read error");
					if (nread <= 0) {
						goto hungup;
					} else {
						if (handleRequest
						    (&Clients[i], &ioHdr) < 0) {
						      hungup:
							printf
							    ("loop: closed fd %d\n",
							     clifd);
							client_del(clifd);
							FD_CLR(clifd, &allset);
							close(clifd);
						}
					}
					pthread_mutex_unlock(&Clients[i].mux);
#if 0
					printf
					    ("thread %d: unlocked Clients[%d].mux\n",
					     mypid, i);
				} else {
					printf
					    ("thread %d: missed Clients[%d].mux\n",
					     mypid, i);
#endif
				}
			}
		}
	}
	pthread_exit(&err);
}

/******************************************************************************
 * signal handler.
 * Send a SIGUSR1 to stop threads.
 */
void
mysigact(int sig)
{
	if (sig == SIGUSR1)
		Server_Runs = 0;
}

int
send_to_server(unsigned int val, uint16_t port)
{
	char addr[4] = { 127, 0, 0, 1 };
	uint32_t *addr_32 = (uint32_t *) & addr[0];
	struct in_addr host;
	int sock_fd;
	struct sockaddr_in server;

	/*create the socket and fill the sockaddr with the inputs */
	host.s_addr = *addr_32;  /* FIXME: IPv6 */
	sock_fd = socket(AF_INET, SOCK_DGRAM, 0); 
	if (sock_fd < 0) {
		fprintf(stderr, "error creating socket: %s\n", strerror(errno));
		return -1;
	}
	server.sin_family = AF_INET;
	server.sin_addr = host;
	server.sin_port = htons(port);

	if (sendto(sock_fd, &val, sizeof(unsigned int), 0,
		   (struct sockaddr *) &server, sizeof(server)) < 0) {
		perror("main: couldn't send the port number");
		close(sock_fd);
		return -1;
	}
	fprintf(stderr, "time to close\n");
	close(sock_fd);
	return 0;
}

/******************************************************************************
 * main()
 * Parse some options, checkout the world, listen to sockets, spawn
 * threads. You know, same old, same old.
 */
int
main(int argc, char **argv)
{
	int listenfd;
	struct sockaddr_in myaddr;
	pthread_t tmpTid;
	unsigned int port;
	int c;
	int length = sizeof(struct sockaddr_in);
	extern char *optarg;
	extern int optind;
	int threads = 1;

	/* get options. */
	if (argc < 2) {
		printf("Usage: %s "	/* [-p port] [-s size ] */
		       "<device or file>\n",	/*[-t max thread cnt ] */
		       argv[0]);
		return (1);
	}
	while ((c = getopt(argc, argv, "s:t:rh?")) != -1) {
		switch (c) {
		case 's':
			datasize = atoi(optarg);
			/*
			   case 'p':
			   port=atoi(optarg);
			   break;
			 */
		case 'r':
			readonly = 1;
			break;
		case 't':
			threads = atoi(optarg);
			break;
		case '?':
		case 'h':
			printf("Usage: %s [-p port] "	/*[-s size] */
			       "<device or file>\n",	/*[-t max thread cnt ] */
			       argv[0]);
			return (1);
			break;
		}
	}

#if 0
	printf(" sizeof theader_t %u\n", sizeof(theader_t));
	printf(" sizeof tio_info_t %u\n", sizeof(tio_info_t));
#endif

	/* setup signal handlers */
	signal(SIGUSR1, mysigact);
	signal(SIGPIPE, SIG_IGN);

	/*open the device. */
	data_fd = open(argv[optind], (readonly ? O_RDONLY : O_RDWR) | O_SYNC);
	if (data_fd < 0) {
		perror("main: Failed to open data file");
		return 1;
	}
	if (datasize == 0) {
		struct stat st;
		if (fstat(data_fd, &st)) {
			perror("Could not fstat exported file");
			return 1;
		}
		if (S_ISBLK(st.st_mode)) {
			if (ioctl(data_fd, BLKGETSIZE, (long *) &datasize) < 0) {
				perror
				    ("Couldn't get sector count on block device");
				return 1;
			}
			datasize *= 512;	/*sector size is hardcoded. */
		} else {
			datasize = (uint64_t) st.st_size;
		}
		if (datasize == 0) {
			fprintf(stderr,
				"Could not determin size of exported data\n");
			return 1;
		}
	}
	printf("Device size = %llx\n", datasize);

	/*attempt to mmap it. */
	/*datamap = mmap( NULL, datasize,
	   readonly?PROT_READ:(PROT_READ|PROT_WRITE),
	   MAP_SHARED,
	   data_fd, 0);
	   if(datamap==(caddr_t)-1 || datamap==NULL) {
	   printf("mmap() failed.  Will use lseek/read/write instead.\n");
	   datamap=NULL;
	   } */

	datamap = NULL;

	pthread_mutex_init(&AcceptMux, NULL);

	/* setup socket */
	if ((listenfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
		perror("loop: Socket biffed");
		exit(1);
	}

	/* Set myself up to listen to the world. */
	myaddr.sin_family = AF_INET;
	myaddr.sin_addr.s_addr = INADDR_ANY;
	myaddr.sin_port = 0;
	if (bind(listenfd, (struct sockaddr *) &myaddr,
		 sizeof(struct sockaddr_in)) < 0) {
		perror("loop: Bind Biffed");
		exit(1);
	}

	if (listen(listenfd, 5) < 0) {
		perror("loop: Listen Biffed");
		exit(1);
	}
	if (getsockname(listenfd, (struct sockaddr *) &myaddr, &length) < 0) {
		perror("error geting the port number");
		exit(1);
	}

	client_add(listenfd);

	/* Start Accept loops */
	for (threads--; threads > 0; threads--) {
		if (pthread_create(&tmpTid, NULL, loop, (void *) &listenfd) < 0)
			perror("main: thread_create biffed");
		if (pthread_detach(tmpTid) < 0)
			perror("main: thread_detach biffed");
	}

	/* Link the main thread to the last thread we create. This is kinda bad.
	 * really should build a counting semaphore here, and have each worker
	 * thread dec that, and when it hits the end, then the main thread can
	 * exit. 
	 * As it is now, when this attached thread dies, the app dies. Even if
	 * there is activity on the other threads.
	 *
	 * But this will work for proof of concept just fine,
	 */
	if (pthread_create(&tmpTid, NULL, loop, (void *) &listenfd) < 0)
		perror("main: thread_create biffed");

	fprintf(stderr, "all threads created\n");

	port = (unsigned int) ntohs(myaddr.sin_port);
	if (send_to_server(port, 24314) < 0)
		exit(1);

	if (pthread_join(tmpTid, (void *) &c) < 0) {
		perror("main: thread_join biffed");
		exit(1);
	}

	if (datamap)
		munmap(datamap, datasize);
	close(data_fd);
	return 0;
}