ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/IO-AIO/AIO.xs
(Generate patch)

Comparing IO-AIO/AIO.xs (file contents):
Revision 1.3 by root, Sun Jul 10 20:07:11 2005 UTC vs.
Revision 1.9 by root, Tue Jul 12 11:02:54 2005 UTC

1#define PERL_NO_GET_CONTEXT
2
3#include "EXTERN.h" 1#include "EXTERN.h"
4#include "perl.h" 2#include "perl.h"
5#include "XSUB.h" 3#include "XSUB.h"
6 4
7#include <sys/types.h> 5#include <sys/types.h>
8#include <sys/stat.h> 6#include <sys/stat.h>
9#include <unistd.h> 7#include <unistd.h>
10#include <fcntl.h> 8#include <fcntl.h>
11#include <signal.h> 9#include <signal.h>
12#include <sched.h> 10#include <sched.h>
13#include <endian.h>
14 11
15#include <pthread.h> 12#include <pthread.h>
16#include <sys/syscall.h> 13#include <sys/syscall.h>
17 14
18typedef void *InputStream; /* hack, but 5.6.1 is simply toooo old ;) */ 15typedef void *InputStream; /* hack, but 5.6.1 is simply toooo old ;) */
19typedef void *OutputStream; /* hack, but 5.6.1 is simply toooo old ;) */ 16typedef void *OutputStream; /* hack, but 5.6.1 is simply toooo old ;) */
20typedef void *InOutStream; /* hack, but 5.6.1 is simply toooo old ;) */ 17typedef void *InOutStream; /* hack, but 5.6.1 is simply toooo old ;) */
21 18
22#if __i386 || __amd64
23# define STACKSIZE ( 256 * sizeof (long))
24#elif __ia64 19#if __ia64
25# define STACKSIZE (8192 * sizeof (long)) 20# define STACKSIZE 65536
26#else 21#else
27# define STACKSIZE ( 512 * sizeof (long)) 22# define STACKSIZE 4096
28#endif 23#endif
29 24
30enum { 25enum {
31 REQ_QUIT, 26 REQ_QUIT,
32 REQ_OPEN, REQ_CLOSE, 27 REQ_OPEN, REQ_CLOSE,
… …
55 50
56typedef aio_cb *aio_req; 51typedef aio_cb *aio_req;
57 52
58static int started; 53static int started;
59static int nreqs; 54static int nreqs;
55static int max_outstanding = 1<<30;
60static int respipe [2]; 56static int respipe [2];
61 57
62static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER; 58static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;
63static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER; 59static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER;
64static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 60static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
65 61
66static volatile aio_req reqs, reqe; /* queue start, queue end */ 62static volatile aio_req reqs, reqe; /* queue start, queue end */
67static volatile aio_req ress, rese; /* queue start, queue end */ 63static volatile aio_req ress, rese; /* queue start, queue end */
64
65static void
66poll_wait ()
67{
68 if (!nreqs)
69 return;
70
71 fd_set rfd;
72 FD_ZERO(&rfd);
73 FD_SET(respipe [0], &rfd);
74
75 select (respipe [0] + 1, &rfd, 0, 0, 0);
76}
77
78static int
79poll_cb ()
80{
81 dSP;
82 int count = 0;
83 aio_req req;
84
85 {
86 /* read and signals sent by the worker threads */
87 char buf [32];
88 while (read (respipe [0], buf, 32) > 0)
89 ;
90 }
91
92 for (;;)
93 {
94 pthread_mutex_lock (&reslock);
95
96 req = ress;
97
98 if (ress)
99 {
100 ress = ress->next;
101 if (!ress) rese = 0;
102 }
103
104 pthread_mutex_unlock (&reslock);
105
106 if (!req)
107 break;
108
109 nreqs--;
110
111 if (req->type == REQ_QUIT)
112 started--;
113 else
114 {
115 int errorno = errno;
116 errno = req->errorno;
117
118 if (req->type == REQ_READ)
119 SvCUR_set (req->data, req->dataoffset
120 + req->result > 0 ? req->result : 0);
121
122 if (req->data)
123 SvREFCNT_dec (req->data);
124
125 if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT)
126 {
127 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
128 PL_laststatval = req->result;
129 PL_statcache = *(req->statdata);
130
131 Safefree (req->statdata);
132 }
133
134 PUSHMARK (SP);
135 XPUSHs (sv_2mortal (newSViv (req->result)));
136
137 if (req->type == REQ_OPEN)
138 {
139 /* convert fd to fh */
140 SV *fh;
141
142 PUTBACK;
143 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
144 SPAGAIN;
145
146 fh = POPs;
147
148 PUSHMARK (SP);
149 XPUSHs (fh);
150 }
151
152 if (SvOK (req->callback))
153 {
154 PUTBACK;
155 call_sv (req->callback, G_VOID | G_EVAL);
156 SPAGAIN;
157 }
158
159 if (req->callback)
160 SvREFCNT_dec (req->callback);
161
162 errno = errorno;
163 count++;
164 }
165
166 Safefree (req);
167 }
168
169 return count;
170}
68 171
69static void *aio_proc(void *arg); 172static void *aio_proc(void *arg);
70 173
71static void 174static void
72start_thread (void) 175start_thread (void)
… …
105 else 208 else
106 reqe = reqs = req; 209 reqe = reqs = req;
107 210
108 pthread_cond_signal (&reqwait); 211 pthread_cond_signal (&reqwait);
109 pthread_mutex_unlock (&reqlock); 212 pthread_mutex_unlock (&reqlock);
213
214 while (nreqs > max_outstanding)
215 {
216 poll_wait ();
217 poll_cb ();
218 }
110} 219}
111 220
112static void 221static void
113end_thread (void) 222end_thread (void)
114{ 223{
… …
118 227
119 send_req (req); 228 send_req (req);
120} 229}
121 230
122static void 231static void
123read_write (pTHX_
124 int dowrite, int fd, off_t offset, size_t length, 232read_write (int dowrite, int fd, off_t offset, size_t length,
125 SV *data, STRLEN dataoffset, SV *callback) 233 SV *data, STRLEN dataoffset, SV *callback)
126{ 234{
127 aio_req req; 235 aio_req req;
128 STRLEN svlen; 236 STRLEN svlen;
129 char *svptr = SvPV (data, svlen); 237 char *svptr = SvPV (data, svlen);
… …
166 req->callback = SvREFCNT_inc (callback); 274 req->callback = SvREFCNT_inc (callback);
167 275
168 send_req (req); 276 send_req (req);
169} 277}
170 278
171static void
172poll_wait ()
173{
174 if (!nreqs)
175 return;
176
177 fd_set rfd;
178 FD_ZERO(&rfd);
179 FD_SET(respipe [0], &rfd);
180
181 select (respipe [0] + 1, &rfd, 0, 0, 0);
182}
183
184static int
185poll_cb (pTHX)
186{
187 dSP;
188 int count = 0;
189 aio_req req;
190
191 {
192 /* read and signals sent by the worker threads */
193 char buf [32];
194 while (read (respipe [0], buf, 32) > 0)
195 ;
196 }
197
198 for (;;)
199 {
200 pthread_mutex_lock (&reslock);
201
202 req = ress;
203
204 if (ress)
205 {
206 ress = ress->next;
207 if (!ress) rese = 0;
208 }
209
210 pthread_mutex_unlock (&reslock);
211
212 if (!req)
213 break;
214
215 nreqs--;
216
217 if (req->type == REQ_QUIT)
218 started--;
219 else
220 {
221 int errorno = errno;
222 errno = req->errorno;
223
224 if (req->type == REQ_READ)
225 SvCUR_set (req->data, req->dataoffset
226 + req->result > 0 ? req->result : 0);
227
228 if (req->data)
229 SvREFCNT_dec (req->data);
230
231 if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT)
232 {
233 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
234 PL_laststatval = req->result;
235 PL_statcache = *(req->statdata);
236
237 Safefree (req->statdata);
238 }
239
240 PUSHMARK (SP);
241 XPUSHs (sv_2mortal (newSViv (req->result)));
242
243 if (req->type == REQ_OPEN)
244 {
245 /* convert fd to fh */
246 SV *fh;
247
248 PUTBACK;
249 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
250 SPAGAIN;
251
252 fh = POPs;
253
254 PUSHMARK (SP);
255 XPUSHs (fh);
256 }
257
258 PUTBACK;
259 call_sv (req->callback, G_VOID | G_EVAL);
260 SPAGAIN;
261
262 if (req->callback)
263 SvREFCNT_dec (req->callback);
264
265 errno = errorno;
266 count++;
267 }
268
269 Safefree (req);
270 }
271
272 return count;
273}
274
275static void * 279static void *
276aio_proc (void *thr_arg) 280aio_proc (void *thr_arg)
277{ 281{
278 aio_req req; 282 aio_req req;
279 int type; 283 int type;
… …
359 return 0; 363 return 0;
360} 364}
361 365
362MODULE = IO::AIO PACKAGE = IO::AIO 366MODULE = IO::AIO PACKAGE = IO::AIO
363 367
368PROTOTYPES: ENABLE
369
364BOOT: 370BOOT:
365{ 371{
366 if (pipe (respipe)) 372 if (pipe (respipe))
367 croak ("unable to initialize result pipe"); 373 croak ("unable to initialize result pipe");
368 374
… …
395 } 401 }
396 402
397 while (started > nthreads) 403 while (started > nthreads)
398 { 404 {
399 poll_wait (); 405 poll_wait ();
400 poll_cb (aTHX); 406 poll_cb ();
401 } 407 }
402} 408}
403 409
410int
411max_outstanding(nreqs)
412 int nreqs
413 PROTOTYPE: $
414 CODE:
415 RETVAL = max_outstanding;
416 max_outstanding = nreqs;
417
404void 418void
405aio_open(pathname,flags,mode,callback) 419aio_open(pathname,flags,mode,callback=&PL_sv_undef)
406 SV * pathname 420 SV * pathname
407 int flags 421 int flags
408 int mode 422 int mode
409 SV * callback 423 SV * callback
410 PROTOTYPE: $$$$ 424 PROTOTYPE: $$$;$
411 CODE: 425 CODE:
412{ 426{
413 aio_req req; 427 aio_req req;
414 428
415 Newz (0, req, 1, aio_cb); 429 Newz (0, req, 1, aio_cb);
… …
426 440
427 send_req (req); 441 send_req (req);
428} 442}
429 443
430void 444void
431aio_close(fh,callback) 445aio_close(fh,callback=&PL_sv_undef)
432 InputStream fh 446 InputStream fh
433 SV * callback 447 SV * callback
434 PROTOTYPE: $$ 448 PROTOTYPE: $;$
435 ALIAS: 449 ALIAS:
436 aio_close = REQ_CLOSE 450 aio_close = REQ_CLOSE
437 aio_fsync = REQ_FSYNC 451 aio_fsync = REQ_FSYNC
438 aio_fdatasync = REQ_FDATASYNC 452 aio_fdatasync = REQ_FDATASYNC
439 CODE: 453 CODE:
… …
451 465
452 send_req (req); 466 send_req (req);
453} 467}
454 468
455void 469void
456aio_read(fh,offset,length,data,dataoffset,callback) 470aio_read(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
457 InputStream fh 471 InputStream fh
458 UV offset 472 UV offset
459 IV length 473 IV length
460 SV * data 474 SV * data
461 IV dataoffset 475 IV dataoffset
462 SV * callback 476 SV * callback
463 PROTOTYPE: $$$$$$ 477 PROTOTYPE: $$$$$;$
464 CODE: 478 CODE:
465 read_write (aTHX_ 0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback); 479 read_write (0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
466 480
467void 481void
468aio_write(fh,offset,length,data,dataoffset,callback) 482aio_write(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
469 OutputStream fh 483 OutputStream fh
470 UV offset 484 UV offset
471 IV length 485 IV length
472 SV * data 486 SV * data
473 IV dataoffset 487 IV dataoffset
474 SV * callback 488 SV * callback
475 PROTOTYPE: $$$$$$ 489 PROTOTYPE: $$$$$;$
476 CODE: 490 CODE:
477 read_write (aTHX_ 1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback); 491 read_write (1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
478 492
479void 493void
480aio_readahead(fh,offset,length,callback) 494aio_readahead(fh,offset,length,callback=&PL_sv_undef)
481 InputStream fh 495 InputStream fh
482 UV offset 496 UV offset
483 IV length 497 IV length
484 SV * callback 498 SV * callback
485 PROTOTYPE: $$$$ 499 PROTOTYPE: $$$;$
486 CODE: 500 CODE:
487{ 501{
488 aio_req req; 502 aio_req req;
489 503
490 if (length < 0) 504 if (length < 0)
… …
503 517
504 send_req (req); 518 send_req (req);
505} 519}
506 520
507void 521void
508aio_stat(fh_or_path,callback) 522aio_stat(fh_or_path,callback=&PL_sv_undef)
509 SV * fh_or_path 523 SV * fh_or_path
510 SV * callback 524 SV * callback
511 PROTOTYPE: $$
512 ALIAS: 525 ALIAS:
526 aio_stat = REQ_STAT
513 aio_lstat = 1 527 aio_lstat = REQ_LSTAT
514 CODE: 528 CODE:
515{ 529{
516 aio_req req; 530 aio_req req;
517 531
518 Newz (0, req, 1, aio_cb); 532 Newz (0, req, 1, aio_cb);
… …
525 if (!req->statdata) 539 if (!req->statdata)
526 croak ("out of memory during aio_req->statdata allocation"); 540 croak ("out of memory during aio_req->statdata allocation");
527 541
528 if (SvPOK (fh_or_path)) 542 if (SvPOK (fh_or_path))
529 { 543 {
530 req->type = ix ? REQ_LSTAT : REQ_STAT; 544 req->type = ix;
531 req->data = newSVsv (fh_or_path); 545 req->data = newSVsv (fh_or_path);
532 req->dataptr = SvPV_nolen (req->data); 546 req->dataptr = SvPV_nolen (req->data);
533 } 547 }
534 else 548 else
535 { 549 {
… …
541 555
542 send_req (req); 556 send_req (req);
543} 557}
544 558
545void 559void
546aio_unlink(pathname,callback) 560aio_unlink(pathname,callback=&PL_sv_undef)
547 SV * pathname 561 SV * pathname
548 SV * callback 562 SV * callback
549 PROTOTYPE: $$
550 CODE: 563 CODE:
551{ 564{
552 aio_req req; 565 aio_req req;
553 566
554 Newz (0, req, 1, aio_cb); 567 Newz (0, req, 1, aio_cb);
… …
562 req->callback = SvREFCNT_inc (callback); 575 req->callback = SvREFCNT_inc (callback);
563 576
564 send_req (req); 577 send_req (req);
565} 578}
566 579
580void
581flush()
582 PROTOTYPE:
583 CODE:
584 while (nreqs)
585 {
586 poll_wait ();
587 poll_cb ();
588 }
589
590void
591poll()
592 PROTOTYPE:
593 CODE:
594 if (nreqs)
595 {
596 poll_wait ();
597 poll_cb ();
598 }
599
567int 600int
568poll_fileno() 601poll_fileno()
569 PROTOTYPE: 602 PROTOTYPE:
570 CODE: 603 CODE:
571 RETVAL = respipe [0]; 604 RETVAL = respipe [0];
… …
574 607
575int 608int
576poll_cb(...) 609poll_cb(...)
577 PROTOTYPE: 610 PROTOTYPE:
578 CODE: 611 CODE:
579 RETVAL = poll_cb (aTHX); 612 RETVAL = poll_cb ();
580 OUTPUT: 613 OUTPUT:
581 RETVAL 614 RETVAL
582 615
583void 616void
584poll_wait() 617poll_wait()

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines