usr.bin/csup/mux.c

144 struct chan {
170 	struct chan	*channels[MUX_MAXCHAN];
194 static struct chan	*chan_new(struct mux *);
195 static struct chan	*chan_get(struct mux *, int);
196 static struct chan	*chan_connect(struct mux *, int);
197 static void		 chan_lock(struct chan *);
198 static void		 chan_unlock(struct chan *);
199 static int		 chan_insert(struct mux *, struct chan *);
200 static void		 chan_free(struct chan *);
307 mux_open(int sock, struct chan **chan)
310 	struct chan *chan0;
334 	*chan = chan0;
345 	struct chan *chan;
350 		chan = m->channels[i];
351 		if (chan != NULL)
352 			chan_free(chan);
365 chan_close(struct chan *chan)
368 	chan_lock(chan);
369 	if (chan->state == CS_ESTABLISHED) {
370 		chan->state = CS_WRCLOSED;
371 		chan->flags |= CF_CLOSE;
372 	} else if (chan->state == CS_RDCLOSED) {
373 		chan->state = CS_CLOSED;
374 		chan->flags |= CF_CLOSE;
375 	} else if (chan->state == CS_WRCLOSED || chan->state == CS_CLOSED) {
376 		chan_unlock(chan);
379 		chan_unlock(chan);
382 	chan_unlock(chan);
383 	sender_wakeup(chan->mux);
388 chan_wait(struct chan *chan)
391 	chan_lock(chan);
392 	while (chan->state != CS_CLOSED)
393 		pthread_cond_wait(&chan->rdready, &chan->lock);
394 	chan_unlock(chan);
401 	struct chan *chan;
406 		chan = m->channels[i];
407 		chan_lock(chan);
408 		if (chan->state == CS_UNUSED) {
410 			chan->state = CS_LISTENING;
411 			chan_unlock(chan);
414 		chan_unlock(chan);
417 	chan = chan_new(m);
418 	chan->state = CS_LISTENING;
419 	i = chan_insert(m, chan);
421 		chan_free(chan);
425 struct chan *
428 	struct chan *chan;
430 	chan = chan_get(m, id);
431 	while (chan->state == CS_LISTENING)
432 		pthread_cond_wait(&chan->rdready, &chan->lock);
433 	if (chan->state != CS_ESTABLISHED) {
435 		chan_unlock(chan);
438 	chan_unlock(chan);
439 	return (chan);
444 chan_read(struct chan *chan, void *buf, size_t size)
450 	chan_lock(chan);
452 		if (chan->state == CS_RDCLOSED || chan->state == CS_CLOSED) {
453 			chan_unlock(chan);
456 		if (chan->state != CS_ESTABLISHED &&
457 		    chan->state != CS_WRCLOSED) {
458 			chan_unlock(chan);
462 		count = buf_count(chan->recvbuf);
465 		pthread_cond_wait(&chan->rdready, &chan->lock);
468 	buf_get(chan->recvbuf, cp, n);
469 	chan->recvseq += n;
470 	chan->flags |= CF_WINDOW;
471 	chan_unlock(chan);
473 	sender_wakeup(chan->mux);
479 chan_write(struct chan *chan, const void *buf, size_t size)
486 	chan_lock(chan);
489 			if (chan->state != CS_ESTABLISHED &&
490 			    chan->state != CS_RDCLOSED) {
491 				chan_unlock(chan);
495 			avail = buf_avail(chan->sendbuf);
498 			pthread_cond_wait(&chan->wrready, &chan->lock);
501 		buf_put(chan->sendbuf, cp + pos, n);
504 	chan_unlock(chan);
505 	sender_wakeup(chan->mux);
513 static struct chan *
516 	struct chan *chan;
518 	chan = chan_get(m, id);
519 	if (chan->state != CS_UNUSED) {
520 		chan_unlock(chan);
523 	chan->state = CS_CONNECTING;
524 	chan->flags |= CF_CONNECT;
525 	chan_unlock(chan);
527 	chan_lock(chan);
528 	while (chan->state == CS_CONNECTING)
529 		pthread_cond_wait(&chan->wrready, &chan->lock);
530 	if (chan->state != CS_ESTABLISHED) {
531 		chan_unlock(chan);
534 	chan_unlock(chan);
535 	return (chan);
542 static struct chan *
545 	struct chan *chan;
549 	chan = m->channels[id];
550 	if (chan == NULL) {
551 		chan = chan_new(m);
552 		m->channels[id] = chan;
555 	chan_lock(chan);
557 	return (chan);
562 chan_lock(struct chan *chan)
566 	error = pthread_mutex_lock(&chan->lock);
572 chan_unlock(struct chan *chan)
576 	error = pthread_mutex_unlock(&chan->lock);
583 static struct chan *
586 	struct chan *chan;
588 	chan = xmalloc(sizeof(struct chan));
589 	chan->state = CS_UNUSED;
590 	chan->flags = 0;
591 	chan->mux = m;
592 	chan->sendbuf = buf_new(CHAN_SBSIZE);
593 	chan->sendseq = 0;
594 	chan->sendwin = 0;
595 	chan->sendmss = 0;
596 	chan->recvbuf = buf_new(CHAN_RBSIZE);
597 	chan->recvseq = 0;
598 	chan->recvmss = CHAN_MAXSEGSIZE;
599 	pthread_mutex_init(&chan->lock, NULL);
600 	pthread_cond_init(&chan->rdready, NULL);
601 	pthread_cond_init(&chan->wrready, NULL);
602 	return (chan);
607 chan_free(struct chan *chan)
610 	pthread_cond_destroy(&chan->rdready);
611 	pthread_cond_destroy(&chan->wrready);
612 	pthread_mutex_destroy(&chan->lock);
613 	buf_free(chan->recvbuf);
614 	buf_free(chan->sendbuf);
615 	free(chan);
620 chan_insert(struct mux *m, struct chan *chan)
627 			m->channels[i] = chan;
700 	struct chan *chan;
730 			chan = m->channels[i];
731 			chan_lock(chan);
732 			if (chan->state != CS_UNUSED) {
733 				chan->state = CS_CLOSED;
734 				chan->flags = 0;
735 				pthread_cond_broadcast(&chan->rdready);
736 				pthread_cond_broadcast(&chan->wrready);
738 			chan_unlock(chan);
781 	struct chan *chan;
791 	chan = chan_get(m, id);
797 		mh.mh_connect.mss = htons(chan->recvmss);
798 		mh.mh_connect.window = htonl(chan->recvseq +
799 		    chan->recvbuf->size);
805 		mh.mh_accept.mss = htons(chan->recvmss);
806 		mh.mh_accept.window = htonl(chan->recvseq +
807 		    chan->recvbuf->size);
818 		mh.mh_window.window = htonl(chan->recvseq +
819 		    chan->recvbuf->size);
825 		size = min(buf_count(chan->sendbuf), chan->sendmss);
826 		winsize = chan->sendwin - chan->sendseq;
849 		buf = chan->sendbuf;
866 		chan_unlock(chan);
870 		chan_lock(chan);
871 		chan->sendseq += size;
875 		pthread_cond_signal(&chan->wrready);
876 		chan_unlock(chan);
878 		chan_unlock(chan);
928 	struct chan *chan;
938 		chan = m->channels[id];
939 		chan_lock(chan);
940 		if (chan->state != CS_UNUSED) {
941 			if (chan->sendseq != chan->sendwin &&
942 			    buf_count(chan->sendbuf) > 0)
943 				chan->flags |= CF_DATA;
944 			if (chan->flags) {
946 				if (chan->flags & CF_CONNECT)
948 				else if (chan->flags & CF_ACCEPT)
950 				else if (chan->flags & CF_RESET)
952 				else if (chan->flags & CF_WINDOW)
954 				else if (chan->flags & CF_DATA)
956 				else if (chan->flags & CF_CLOSE)
958 				chan->flags &= ~*what;
959 				chan_unlock(chan);
964 		chan_unlock(chan);
978 	struct chan *chan;
991 			chan = chan_get(m, mh.mh_connect.id);
992 			if (chan->state == CS_LISTENING) {
993 				chan->state = CS_ESTABLISHED;
994 				chan->sendmss = ntohs(mh.mh_connect.mss);
995 				chan->sendwin = ntohl(mh.mh_connect.window);
996 				chan->flags |= CF_ACCEPT;
997 				pthread_cond_signal(&chan->rdready);
999 				chan->flags |= CF_RESET;
1000 			chan_unlock(chan);
1007 			chan = chan_get(m, mh.mh_accept.id);
1008 			if (chan->state == CS_CONNECTING) {
1009 				chan->sendmss = ntohs(mh.mh_accept.mss);
1010 				chan->sendwin = ntohl(mh.mh_accept.window);
1011 				chan->state = CS_ESTABLISHED;
1012 				pthread_cond_signal(&chan->wrready);
1013 				chan_unlock(chan);
1015 				chan->flags |= CF_RESET;
1016 				chan_unlock(chan);
1029 			chan = chan_get(m, mh.mh_window.id);
1030 			if (chan->state == CS_ESTABLISHED ||
1031 			    chan->state == CS_RDCLOSED) {
1032 				chan->sendwin = ntohl(mh.mh_window.window);
1033 				chan_unlock(chan);
1036 				chan_unlock(chan);
1043 			chan = chan_get(m, mh.mh_data.id);
1045 			buf = chan->recvbuf;
1046 			if ((chan->state != CS_ESTABLISHED &&
1047 			     chan->state != CS_WRCLOSED) ||
1049 			     len > chan->recvmss)) {
1050 				chan_unlock(chan);
1058 			chan_unlock(chan);
1071 			chan_lock(chan);
1075 			pthread_cond_signal(&chan->rdready);
1076 			chan_unlock(chan);
1082 			chan = chan_get(m, mh.mh_close.id);
1083 			if (chan->state == CS_ESTABLISHED)
1084 				chan->state = CS_RDCLOSED;
1085 			else if (chan->state == CS_WRCLOSED)
1086 				chan->state = CS_CLOSED;
1089 			pthread_cond_signal(&chan->rdready);
1090 			chan_unlock(chan);