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 	errno = 0;
237 	nn = read(fd, a, n);
238 	if(nn <= 0){
239 		if(errno == EINTR)
240 			goto again;
241 		if(errno == EAGAIN || errno == EWOULDBLOCK){
242 			_threadfdwait(fd, 'r', getcallerpc(&fd));
243 			goto again;
244 		}
245 	}
246 	return nn;
247 }
248 
249 int
250 threadrecvfd(int fd)
251 {
252 	int nn;
253 
254 	threadfdnoblock(fd);
255 again:
256 	nn = recvfd(fd);
257 	if(nn < 0){
258 		if(errno == EINTR)
259 			goto again;
260 		if(errno == EAGAIN || errno == EWOULDBLOCK){
261 			_threadfdwait(fd, 'r', getcallerpc(&fd));
262 			goto again;
263 		}
264 	}
265 	return nn;
266 }
267 
268 int
269 threadsendfd(int fd, int sfd)
270 {
271 	int nn;
272 
273 	threadfdnoblock(fd);
274 again:
275 	nn = sendfd(fd, sfd);
276 	if(nn < 0){
277 		if(errno == EINTR)
278 			goto again;
279 		if(errno == EAGAIN || errno == EWOULDBLOCK){
280 			_threadfdwait(fd, 'w', getcallerpc(&fd));
281 			goto again;
282 		}
283 	}
284 	return nn;
285 }
286 
287 long
288 threadreadn(int fd, void *a, long n)
289 {
290 	int tot, nn;
291 
292 	for(tot = 0; tot<n; tot+=nn){
293 		nn = threadread(fd, (char*)a+tot, n-tot);
294 		if(nn <= 0){
295 			if(tot == 0)
296 				return nn;
297 			return tot;
298 		}
299 	}
300 	return tot;
301 }
302 
303 long
304 _threadwrite(int fd, const void *a, long n)
305 {
306 	int nn;
307 
308 	threadfdnoblock(fd);
309 again:
310 	nn = write(fd, a, n);
311 	if(nn < 0){
312 		if(errno == EINTR)
313 			goto again;
314 		if(errno == EAGAIN || errno == EWOULDBLOCK){
315 			_threadfdwait(fd, 'w', getcallerpc(&fd));
316 			goto again;
317 		}
318 	}
319 	return nn;
320 }
321 
322 long
323 threadwrite(int fd, const void *a, long n)
324 {
325 	int tot, nn;
326 
327 	for(tot = 0; tot<n; tot+=nn){
328 		nn = _threadwrite(fd, (char*)a+tot, n-tot);
329 		if(nn <= 0){
330 			if(tot == 0)
331 				return nn;
332 			return tot;
333 		}
334 	}
335 	return tot;
336 }
337