1 /* Copyright (C) 2003 Russ Cox, Massachusetts Institute of Technology */
2 /* See COPYRIGHT */
3 
4 #include <u.h>
5 #include <libc.h>
6 #include <mux.h>
7 
8 /*
9  * If you fork off two procs running muxrecvproc and muxsendproc,
10  * then muxrecv/muxsend (and thus muxrpc) will never block except on 
11  * rendevouses, which is nice when it's running in one thread of many.
12  */
13 void
14 _muxrecvproc(void *v)
15 {
16 	void *p;
17 	Mux *mux;
18 	Muxqueue *q;
19 
20 	mux = v;
21 	q = _muxqalloc();
22 
23 	qlock(&mux->lk);
24 	mux->readq = q;
25 	qlock(&mux->inlk);
26 	rwakeup(&mux->rpcfork);
27 	qunlock(&mux->lk);
28 
29 	while((p = mux->recv(mux)) != nil)
30 		if(_muxqsend(q, p) < 0){
31 			free(p);
32 			break;
33 		}
34 	qunlock(&mux->inlk);
35 	qlock(&mux->lk);
36 	_muxqhangup(q);
37 	while((p = _muxnbqrecv(q)) != nil)
38 		free(p);
39 	free(q);
40 	mux->readq = nil;
41 	rwakeup(&mux->rpcfork);
42 	qunlock(&mux->lk);
43 }
44 
45 void
46 _muxsendproc(void *v)
47 {
48 	Muxqueue *q;
49 	void *p;
50 	Mux *mux;
51 
52 	mux = v;
53 	q = _muxqalloc();
54 
55 	qlock(&mux->lk);
56 	mux->writeq = q;
57 	qlock(&mux->outlk);
58 	rwakeup(&mux->rpcfork);
59 	qunlock(&mux->lk);
60 
61 	while((p = _muxqrecv(q)) != nil)
62 		if(mux->send(mux, p) < 0)
63 			break;
64 	qunlock(&mux->outlk);
65 	qlock(&mux->lk);
66 	_muxqhangup(q);
67 	while((p = _muxnbqrecv(q)) != nil)
68 		free(p);
69 	free(q);
70 	mux->writeq = nil;
71 	rwakeup(&mux->rpcfork);
72 	qunlock(&mux->lk);
73 	return;
74 }
75 
76 void*
77 _muxrecv(Mux *mux)
78 {
79 	void *p;
80 
81 	qlock(&mux->lk);
82 /*
83 	if(mux->state != VtStateConnected){
84 		werrstr("not connected");
85 		qunlock(&mux->lk);
86 		return nil;
87 	}
88 */
89 	if(mux->readq){
90 		qunlock(&mux->lk);
91 		return _muxqrecv(mux->readq);
92 	}
93 
94 	qlock(&mux->inlk);
95 	qunlock(&mux->lk);
96 	p = mux->recv(mux);
97 	qunlock(&mux->inlk);
98 /*
99 	if(!p)
100 		vthangup(mux);
101 */
102 	return p;
103 }
104 
105 int
106 _muxsend(Mux *mux, void *p)
107 {
108 	qlock(&mux->lk);
109 /*
110 	if(mux->state != VtStateConnected){
111 		packetfree(p);
112 		werrstr("not connected");
113 		qunlock(&mux->lk);
114 		return -1;
115 	}
116 */
117 	if(mux->writeq){
118 		qunlock(&mux->lk);
119 		if(_muxqsend(mux->writeq, p) < 0){
120 			free(p);
121 			return -1;
122 		}
123 		return 0;
124 	}
125 
126 	qlock(&mux->outlk);
127 	qunlock(&mux->lk);
128 	if(mux->send(mux, p) < 0){
129 		qunlock(&mux->outlk);
130 		/* vthangup(mux); */
131 		return -1;	
132 	}
133 	qunlock(&mux->outlk);
134 	return 0;
135 }
136