1 #define NOPLAN9DEFINES
2 #include <u.h>
3 #include <libc.h>
4 #include <thread.h>
5 
6 #include <errno.h>
7 #include <unistd.h>
8 #include <fcntl.h>
9 
10 #define debugpoll 0
11 
12 #ifdef __APPLE__
13 #include <sys/time.h>
14 enum { POLLIN=1, POLLOUT=2, POLLERR=4 };
15 struct pollfd
16 {
17 	int fd;
18 	int events;
19 	int revents;
20 };
21 
22 int
23 poll(struct pollfd *p, int np, int ms)
24 {
25 	int i, maxfd, n;
26 	struct timeval tv, *tvp;
27 	fd_set rfd, wfd, efd;
28 	
29 	maxfd = -1;
30 	FD_ZERO(&rfd);
31 	FD_ZERO(&wfd);
32 	FD_ZERO(&efd);
33 	for(i=0; i<np; i++){
34 		p[i].revents = 0;
35 		if(p[i].fd == -1)
36 			continue;
37 		if(p[i].fd > maxfd)
38 			maxfd = p[i].fd;
39 		if(p[i].events & POLLIN)
40 			FD_SET(p[i].fd,	&rfd);
41 		if(p[i].events & POLLOUT)
42 			FD_SET(p[i].fd, &wfd);
43 		FD_SET(p[i].fd, &efd);
44 	}
45 
46 	if(ms != -1){
47 		tv.tv_usec = (ms%1000)*1000;
48 		tv.tv_sec = ms/1000;
49 		tvp = &tv;
50 	}else
51 		tvp = nil;
52 
53 	if(debugpoll){
54 		fprint(2, "select %d:", maxfd+1);
55 		for(i=0; i<=maxfd; i++){
56 			if(FD_ISSET(i, &rfd))
57 				fprint(2, " r%d", i);
58 			if(FD_ISSET(i, &wfd))
59 				fprint(2, " w%d", i);
60 			if(FD_ISSET(i, &efd))
61 				fprint(2, " e%d", i);
62 		}
63 		fprint(2, "; tp=%p, t=%d.%d\n", tvp, tv.tv_sec, tv.tv_usec);
64 	}
65 
66 	n = select(maxfd+1, &rfd, &wfd, &efd, tvp);
67 
68 	if(n <= 0)
69 		return n;
70 
71 	for(i=0; i<np; i++){
72 		if(p[i].fd == -1)
73 			continue;
74 		if(FD_ISSET(p[i].fd, &rfd))
75 			p[i].revents |= POLLIN;
76 		if(FD_ISSET(p[i].fd, &wfd))
77 			p[i].revents |= POLLOUT;
78 		if(FD_ISSET(p[i].fd, &efd))
79 			p[i].revents |= POLLERR;
80 	} 
81 	return n;
82 }
83 
84 #else
85 #include <poll.h>
86 #endif
87 
88 /*
89  * Poll file descriptors in an idle loop.
90  */
91 
92 typedef struct Poll Poll;
93 
94 struct Poll
95 {
96 	Channel *c;	/* for sending back */
97 };
98 
99 static Channel *sleepchan[64];
100 static int sleeptime[64];
101 static int nsleep;
102 
103 static struct pollfd pfd[64];
104 static struct Poll polls[64];
105 static int npoll;
106 
107 static void
108 pollidle(void *v)
109 {
110 	int i, n, t;
111 	uint now;
112 
113 	for(;; yield()){
114 		if(debugpoll) fprint(2, "poll %d:", npoll);
115 		for(i=0; i<npoll; i++){
116 			if(debugpoll) fprint(2, " %d%c", pfd[i].fd, pfd[i].events==POLLIN ? 'r' : 'w');
117 			pfd[i].revents = 0;
118 		}
119 		t = -1;
120 		now = p9nsec()/1000000;
121 		for(i=0; i<nsleep; i++){
122 			n = sleeptime[i] - now;
123 			if(debugpoll) fprint(2, " s%d", n);
124 			if(n < 0)
125 				n = 0;
126 			if(t == -1 || n < t)
127 				t = n;
128 		}
129 		if(debugpoll) fprint(2, "; t=%d\n", t);
130 	
131 		n = poll(pfd, npoll, t);
132 		//fprint(2, "poll ret %d:", n);
133 		now = p9nsec()/1000000;
134 		for(i=0; i<nsleep; i++){
135 			if((int)(sleeptime[i] - now) < 0){
136 				nbsendul(sleepchan[i], 0);
137 				nsleep--;
138 				sleepchan[i] = sleepchan[nsleep];
139 				sleeptime[i] = sleeptime[nsleep];
140 				i--;
141 			}
142 		}
143 				
144 		if(n <= 0)
145 			continue;
146 		for(i=0; i<npoll; i++)
147 			if(pfd[i].fd != -1 && pfd[i].revents){
148 				//fprint(2, " %d", pfd[i].fd);
149 				pfd[i].fd = -1;
150 				pfd[i].events = 0;
151 				pfd[i].revents = 0;
152 				nbsendul(polls[i].c, 1);
153 				//fprint(2, " x%d", pfd[i].fd);
154 			}
155 		//fprint(2, "\n");
156 	}
157 }
158 
159 void
160 threadfdwaitsetup(void)
161 {
162 	static int setup = 0;
163 
164 	if(!setup){
165 		setup = 1;
166 		threadcreateidle(pollidle, nil, 16384);
167 	}
168 }
169 
170 void
171 _threadfdwait(int fd, int rw, ulong pc)
172 {
173 	int i;
174 
175 	struct {
176 		Channel c;
177 		ulong x;
178 	} s;
179 
180 	threadfdwaitsetup();
181 	chaninit(&s.c, sizeof(ulong), 1);
182 	for(i=0; i<npoll; i++)
183 		if(pfd[i].fd == -1)
184 			break;
185 	if(i==npoll){
186 		if(npoll >= nelem(polls)){
187 			fprint(2, "Too many polled fds.\n");
188 			abort();
189 		}
190 		npoll++;
191 	}
192 
193 	pfd[i].fd = fd;
194 	pfd[i].events = rw=='r' ? POLLIN : POLLOUT;
195 	polls[i].c = &s.c;
196 	if(0) fprint(2, "%s [%3d] fdwait %d %c list *0x%lux\n",
197 		argv0, threadid(), fd, rw, pc);
198 	recvul(&s.c);
199 }
200 
201 void
202 threadfdwait(int fd, int rw)
203 {
204 	_threadfdwait(fd, rw, getcallerpc(&fd));
205 }
206 
207 void
208 threadsleep(int ms)
209 {
210 	struct {
211 		Channel c;
212 		ulong x;
213 	} s;
214 
215 	threadfdwaitsetup();
216 	chaninit(&s.c, sizeof(ulong), 1);
217 
218 	sleepchan[nsleep] = &s.c;
219 	sleeptime[nsleep++] = p9nsec()/1000000+ms;
220 	recvul(&s.c);
221 }
222 
223 void
224 threadfdnoblock(int fd)
225 {
226 	fcntl(fd, F_SETFL, fcntl(fd, F_GETFL, 0)|O_NONBLOCK);
227 }
228 
229 long
230 threadread(int fd, void *a, long n)
231 {
232 	int nn;
233 
234 	threadfdnoblock(fd);
235 again:
236 	nn = read(fd, a, n);
237 	if(nn < 0){
238 		if(errno == EINTR)
239 			goto again;
240 		if(errno == EAGAIN || errno == EWOULDBLOCK){
241 			_threadfdwait(fd, 'r', getcallerpc(&fd));
242 			goto again;
243 		}
244 	}
245 	return nn;
246 }
247 
248 int
249 threadrecvfd(int fd)
250 {
251 	int nn;
252 
253 	threadfdnoblock(fd);
254 again:
255 	nn = recvfd(fd);
256 	if(nn < 0){
257 		if(errno == EINTR)
258 			goto again;
259 		if(errno == EAGAIN || errno == EWOULDBLOCK){
260 			_threadfdwait(fd, 'r', getcallerpc(&fd));
261 			goto again;
262 		}
263 	}
264 	return nn;
265 }
266 
267 int
268 threadsendfd(int fd, int sfd)
269 {
270 	int nn;
271 
272 	threadfdnoblock(fd);
273 again:
274 	nn = sendfd(fd, sfd);
275 	if(nn < 0){
276 		if(errno == EINTR)
277 			goto again;
278 		if(errno == EAGAIN || errno == EWOULDBLOCK){
279 			_threadfdwait(fd, 'w', getcallerpc(&fd));
280 			goto again;
281 		}
282 	}
283 	return nn;
284 }
285 
286 long
287 threadreadn(int fd, void *a, long n)
288 {
289 	int tot, nn;
290 
291 	for(tot = 0; tot<n; tot+=nn){
292 		nn = threadread(fd, (char*)a+tot, n-tot);
293 		if(nn <= 0){
294 			if(tot == 0)
295 				return nn;
296 			return tot;
297 		}
298 	}
299 	return tot;
300 }
301 
302 long
303 _threadwrite(int fd, const void *a, long n)
304 {
305 	int nn;
306 
307 	threadfdnoblock(fd);
308 again:
309 	nn = write(fd, a, n);
310 	if(nn < 0){
311 		if(errno == EINTR)
312 			goto again;
313 		if(errno == EAGAIN || errno == EWOULDBLOCK){
314 			_threadfdwait(fd, 'w', getcallerpc(&fd));
315 			goto again;
316 		}
317 	}
318 	return nn;
319 }
320 
321 long
322 threadwrite(int fd, const void *a, long n)
323 {
324 	int tot, nn;
325 
326 	for(tot = 0; tot<n; tot+=nn){
327 		nn = _threadwrite(fd, (char*)a+tot, n-tot);
328 		if(nn <= 0){
329 			if(tot == 0)
330 				return nn;
331 			return tot;
332 		}
333 	}
334 	return tot;
335 }
336