sctp_test.c

SCTP 协议实现源代码

	return sk;

} /* socket_r() */

int bind_r(int sk, struct sockaddr_storage *saddr)
{
        int error = 0, i = 0;
	char *host_s, *serv_s;

	if ((host_s = malloc(NI_MAXHOST)) == NULL) {
		fprintf(stderr, "\n\t\t*** host_s malloc failed!!! ***\n");
		exit(1);
	}
	if ((serv_s = malloc(NI_MAXSERV)) == NULL) {
		fprintf(stderr, "\n\t\t*** serv_s malloc failed!!! ***\n");
		exit(1);
	}

	do {
		if (i > 0) sleep(1); /* sleep a while before new try... */

		error = getnameinfo((struct sockaddr *)saddr, l_len, host_s,
				    NI_MAXHOST, serv_s, NI_MAXSERV,
				    NI_NUMERICHOST);

		if (error)
			printf("%s\n", gai_strerror(error));

       		DEBUG_PRINT(DEBUG_MIN,
			    "\tbind(sk=%d, [a:%s,p:%s])  --  attempt %d/%d\n",
       			    sk, host_s, serv_s, i+1, MAX_BIND_RETRYS);

       		error = bind(sk, (struct sockaddr *)saddr, l_len);

	        if (error != 0) {
	        	if( errno != EADDRINUSE ) {
	        		if (do_exit) {
	        		        fprintf(stderr, "\n\n\t\t***bind: can "
						"not bind to %s:%s: %s ****\n",
						host_s, serv_s,
						strerror(errno));
					exit(1);
				} else {
					return -1;
				}
			}
		}
		i++;
		if (i >= MAX_BIND_RETRYS) {
			fprintf(stderr, "Maximum bind() attempts. "
				"Die now...\n\n");
			exit(1);
		}
        } while (error < 0 && i < MAX_BIND_RETRYS);

	return 0;

} /* bind_r() */

int listen_r(int sk, int listen_count)
{
	int error = 0;
	
        DEBUG_PRINT(DEBUG_MIN, "\tlisten(sk=%d,backlog=%d)\n",
		    sk, listen_count);

 	/* Mark sk as being able to accept new associations */
        error = listen(sk, 1);
        if (error != 0) {
        	if (do_exit) {
                	fprintf(stderr, "\n\n\t\t*** listen:  %s ***\n\n\n",
                       		strerror(errno));
			exit(1);
		}
		else return -1;
        }
        return 0;

} /* listen_r() */

int receive_r(int sk, int once)
{
	int i = 0, error = 0;
        char incmsg[CMSG_SPACE(sizeof(_sctp_cmsg_data_t))];
        struct iovec iov;
        struct msghdr inmessage;

        /* Initialize inmessage with enough space for DATA... */
        memset(&inmessage, 0, sizeof(inmessage));
        if ((iov.iov_base = malloc(REALLY_BIG)) == NULL) {
		fprintf(stderr, "\n\t\t*** malloc not enough memory!!! ***\n");
		exit(1);
	}
	iov.iov_len = REALLY_BIG;
	inmessage.msg_iov = &iov;
	inmessage.msg_iovlen = 1;
	/* or a control message.  */
	inmessage.msg_control = incmsg;
	inmessage.msg_controllen = sizeof(incmsg);

	/* Get the messages sent */
	while (1) {

		DEBUG_PRINT(DEBUG_MIN, "\trecvmsg(sk=%d) ", sk);

		error = recvmsg(sk, &inmessage, MSG_WAITALL);
		if (error < 0 && error != EAGAIN) {
			fprintf(stderr, "\n\t\t*** recvmsg: %s ***\n\n",
					strerror(errno));
			fflush(stdout);
			if (do_exit) exit(1);
			else goto error_out;
		}
		else if (error == 0) {
			printf("\n\t\trecvmsg() returned 0 !!!!\n");
			fflush(stdout);
		}

		if (print_message(sk, &inmessage, error) > 0)
			continue; /* got a notification... */

		inmessage.msg_control = incmsg;
		inmessage.msg_controllen = sizeof(incmsg);
		iov.iov_len = REALLY_BIG;
		i++;
		if (once)
			break;
	}

	free(iov.iov_base);
	return 0;
error_out:
	close(sk);
	free(iov.iov_base);
	return -1;

} /* receive_r () */

int next_order(int state, int pattern)
{
	switch (pattern){
	case ORDER_PATTERN_UNORDERED:
		state = 0;
		break;
	case ORDER_PATTERN_ORDERED:
		state = 1;
		break;
	case ORDER_PATTERN_ALTERNATE:
		state = state ? 0 : 1;
		break;
	case ORDER_PATTERN_RANDOM:
		state = rand() % 2;
		break;
	}

	return state;
}

int next_stream(int state, int pattern)
{
	switch (pattern){
	case STREAM_PATTERN_RANDOM:
		state = rand() % (max_stream + 1);
		break;
	case STREAM_PATTERN_SEQUENTIAL:
		state = state + 1;
		if (state > max_stream)
			state = 0;
		break;
	}

	return state;
}

int next_msg_size(int msg_cnt)
{
	int msg_size;

	if (size_arg) {
		msg_size = size_arg;
	} else if (test_case < NCASES) {
		msg_size = msg_sizes[test_case][msg_cnt];
	} else {
		msg_size = (rand() % max_msgsize) + 1;
	}

	return msg_size;

} /* next_msg_size() */

int next_assoc(int i, int state, int pattern)
{
	int j;
	int found = 0;
	_assoc_state *as;


	switch (pattern){
	case ASSOC_PATTERN_RANDOM:
		state = rand() % tosend;
		break;
	case ASSOC_PATTERN_SEQUENTIAL:
		state = state + 1;
		if (state >= tosend)
			state = 0;
		break;
	}

	as = poll_sks[i].assoc_state;
	j = state;
	do {
		if (as[j].msg_sent < repeat_count) {
			found = 1;
			break;
		}
		if (++j >= tosend) {
			j = 0;
		}
	} while (j != state);

	if (found) {
		return j;
	} else {
		return -1;
	}
	
} /* next_assoc() */

int send_r(int sk, int stream, int order, int send_size, int assoc_i)
{
	int error = 0;
	struct msghdr outmsg;
	struct iovec iov;
	char *message = NULL;
	int msglen = 0;
	char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
	struct cmsghdr *cmsg;
	struct sctp_sndrcvinfo *sinfo;

	if (send_size > 0) {
		message = build_msg(send_size);
		msglen = strlen(message) + 1;
		iov.iov_base = message;
		iov.iov_len = msglen;
	}
	else {
		if (do_exit) {
			exit(1);
		} else {
			goto error_out;
		}
	}

	outmsg.msg_name = &s_rem;
       	outmsg.msg_namelen = sizeof(struct sockaddr_storage);
	outmsg.msg_iov = &iov;
	outmsg.msg_iovlen = 1;
	outmsg.msg_control = outcmsg;
	outmsg.msg_controllen = sizeof(outcmsg);
	outmsg.msg_flags = 0;

	cmsg = CMSG_FIRSTHDR(&outmsg);
	cmsg->cmsg_level = IPPROTO_SCTP;
	cmsg->cmsg_type = SCTP_SNDRCV;
	cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));

	outmsg.msg_controllen = cmsg->cmsg_len;
	sinfo = (struct sctp_sndrcvinfo *)CMSG_DATA(cmsg);
	memset(sinfo, 0, sizeof(struct sctp_sndrcvinfo));
	sinfo->sinfo_ppid = rand();
	sinfo->sinfo_stream = stream;
	sinfo->sinfo_flags = 0;
	if (!order)
		sinfo->sinfo_flags = SCTP_UNORDERED;

	DEBUG_PRINT(DEBUG_MIN, "\tsendmsg(sk=%d, assoc=%d) %4d bytes.\n",
		    sk, assoc_i, send_size);
	DEBUG_PRINT(DEBUG_MAX, "\t  SNDRCV");
	if (DEBUG_MAX == debug_level) {
        	printf("(stream=%u ", 	sinfo->sinfo_stream);
		printf("flags=0x%x ",	sinfo->sinfo_flags);
		printf("ppid=%u\n",	sinfo->sinfo_ppid);
	}

	/* Send to our neighbor.  */
	error = sendmsg(sk, &outmsg, MSG_WAITALL);
	if (error != msglen) {
		fprintf(stderr, "\n\t\t*** sendmsg: %s ***\n\n",
			strerror(errno));
		fflush(stdout);
	
		if (do_exit) {
			exit(1);
		} else {
			if (!drain)
				goto error_out;
		}
	}

	if (send_size > 0) free(message);
	return 0;
error_out:
	if (send_size > 0) free(message);
	return -1;

} /* send_r() */

int close_r(int sk)
{
	int error = 0;
	
	DEBUG_PRINT(DEBUG_MIN, "\tclose(sk=%d)\n",sk);

	error = close(sk);
	if (error != 0) {
		if (do_exit) {
			fprintf(stderr, "\n\n\t\t*** close: %s ***\n\n",
				strerror(errno));
			exit(1);
		} else {
			return -1;
		}
	}
	fflush(stdout);
	return 0;

} /* close_r() */

void
server(int sk)
{
	if (max_msgsize > DEFAULT_MAX_WINDOW) {
		if (setsockopt(sk, SOL_SOCKET, SO_RCVBUF, &max_msgsize,
			       sizeof(max_msgsize)) < 0) {
			perror("setsockopt(SO_RCVBUF)");
			exit(1);
		} 	
	}

	receive_r(sk, 0);

} /* server() */

void
client(int sk)
{
	int msg_size;
	int i;

	for (i = 0; i < msg_cnt; i++) {

		msg_size = next_msg_size(i);
		order_state = next_order(order_state, order_pattern);
		stream_state = next_stream(stream_state, stream_pattern);

		if (send_r(sk, stream_state, order_state, msg_size, 0) < 0) {
			close(sk);
			break;
		}
		/* The sender is echoing so do discard the echoed data. */
		if (drain) {
			receive_r(sk, 1);
		}
	}
} /* client() */

void
mixed_mode_test(void)
{
	int error, i, j, max_fd, sks, size;
	int assoc_i, n_msg_size, n_order, n_stream;
	int done = 0;
	fd_set *ibitsp = NULL, *obitsp = NULL, *xbitsp = NULL;
        char incmsg[CMSG_SPACE(sizeof(_sctp_cmsg_data_t))];
        struct iovec iov;
        struct msghdr inmessage;
	_assoc_state *as;

	
	/* Set up the listeners.  If listeners is 0, set up one socket for
	 * transmitting only.
	 */
	iov.iov_base = NULL;
	max_fd = -1;
	sks = (0 == listeners) ? 1 : listeners;
	memset(poll_sks, 0, sizeof(sks * sizeof(_poll_sks)));

	for (i = 0; i < sks; i++) {
		poll_sks[i].sk = socket_r();

		if (s_loc.ss_family == AF_INET6)
			( (struct sockaddr_in6 *)&s_loc)->sin6_port =
				htons(local_port + i);
		else
			( (struct sockaddr_in *)&s_loc)->sin_port =
				htons(local_port + i);

		bind_r(poll_sks[i].sk, &s_loc);
		if (listeners) {
			listen_r(poll_sks[i].sk, 100);
		}
		if (max_msgsize > DEFAULT_MAX_WINDOW) {
			if (setsockopt(poll_sks[i].sk, SOL_SOCKET, SO_RCVBUF,
				&max_msgsize, sizeof(max_msgsize)) < 0) {
				perror("setsockopt(SO_RCVBUF)");
				exit(1);
			}
		} 	

		if (tosend) {
			if ((poll_sks[i].assoc_state = (_assoc_state *)malloc(
				sizeof(_assoc_state) * tosend)) == NULL) {
				printf("Can't allocate memory.\n");
				goto clean_up;
			}
			memset(poll_sks[i].assoc_state, 0,
				sizeof(_assoc_state) * tosend);
		}

		if (poll_sks[i].sk > max_fd) {
			max_fd = poll_sks[i].sk;
		}
	}

	size = howmany(max_fd + 1, NFDBITS) * sizeof(fd_mask);
	if ((ibitsp = (fd_set *)malloc(size)) == NULL) {
		printf("Can't allocate memory.\n");
		goto clean_up;
	}
	if ((obitsp = (fd_set *)malloc(size)) == NULL) {
		printf("Can't allocate memory.\n");
		goto clean_up;
	}
	if ((xbitsp = (fd_set *)malloc(size)) == NULL) {
		printf("Can't allocate memory.\n");
		goto clean_up;
	}

	memset(ibitsp, 0, size);
	memset(obitsp, 0, size);
	memset(xbitsp, 0, size);


        /* Initialize inmessage with enough space for DATA... */
        memset(&inmessage, 0, sizeof(inmessage));
        if ((iov.iov_base = malloc(REALLY_BIG)) == NULL) {
		fprintf(stderr, "\n\t\t*** malloc not enough memory!!! ***\n");
		goto clean_up;
	}
	iov.iov_len = REALLY_BIG;
	inmessage.msg_iov = &iov;
	inmessage.msg_iovlen = 1;
	/* or a control message.  */
	inmessage.msg_control = incmsg;
	inmessage.msg_controllen = sizeof(incmsg);

	/* Set up the remote port number per association for output.  */
	for (i = 0; i < sks; i++) {
		as = poll_sks[i].assoc_state;
		for (j = 0; j < tosend; j++) {
			as[j].rem_port = remote_port + j;
		}
	}

	while (!done) {

		for (i = 0; i < sks; i++) {
			FD_SET(poll_sks[i].sk, ibitsp);
			FD_SET(poll_sks[i].sk, obitsp);
			FD_SET(poll_sks[i].sk, xbitsp);
		}
		if ((error = select(max_fd + 1, ibitsp, obitsp, xbitsp,
			(struct timeval *)0)) < 0) {
			fprintf(stderr, "\n\t\t*** select() failed ");
			fprintf(stderr, "with error: %s\n\n",
				strerror(errno));
			fflush(stdout);
			goto clean_up;
		}
		
		for (i = 0; i < sks; i++) {
			/* Is there anything to read from the socket?  */
			if (listeners && FD_ISSET(poll_sks[i].sk, ibitsp)) {

				FD_CLR(poll_sks[i].sk, ibitsp);
				error = recvmsg(poll_sks[i].sk, &inmessage,
					MSG_WAITALL);
				if (error < 0) {
					fprintf(stderr,
						"\n\t\t*** recvmsg: %s ***\n\n",
						strerror(errno));
					fflush(stdout);
					goto clean_up;
				}
				else if (error == 0) {