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.8 by root, Mon Jul 11 02:53:59 2005 UTC vs.
Revision 1.27 by root, Wed Aug 17 04:47:02 2005 UTC

1#define _REENTRANT 1
2#include <errno.h>
3
1#include "EXTERN.h" 4#include "EXTERN.h"
2#include "perl.h" 5#include "perl.h"
3#include "XSUB.h" 6#include "XSUB.h"
4 7
8#include "autoconf/config.h"
9
5#include <sys/types.h> 10#include <sys/types.h>
6#include <sys/stat.h> 11#include <sys/stat.h>
12
7#include <unistd.h> 13#include <unistd.h>
8#include <fcntl.h> 14#include <fcntl.h>
9#include <signal.h> 15#include <signal.h>
10#include <sched.h> 16#include <sched.h>
11#include <endian.h>
12 17
13#include <pthread.h> 18#include <pthread.h>
14#include <sys/syscall.h>
15 19
16typedef void *InputStream; /* hack, but 5.6.1 is simply toooo old ;) */ 20typedef void *InputStream; /* hack, but 5.6.1 is simply toooo old ;) */
17typedef void *OutputStream; /* hack, but 5.6.1 is simply toooo old ;) */ 21typedef void *OutputStream; /* hack, but 5.6.1 is simply toooo old ;) */
18typedef void *InOutStream; /* hack, but 5.6.1 is simply toooo old ;) */ 22typedef void *InOutStream; /* hack, but 5.6.1 is simply toooo old ;) */
19 23
… …
25 29
26enum { 30enum {
27 REQ_QUIT, 31 REQ_QUIT,
28 REQ_OPEN, REQ_CLOSE, 32 REQ_OPEN, REQ_CLOSE,
29 REQ_READ, REQ_WRITE, REQ_READAHEAD, 33 REQ_READ, REQ_WRITE, REQ_READAHEAD,
30 REQ_STAT, REQ_LSTAT, REQ_FSTAT, REQ_UNLINK, 34 REQ_STAT, REQ_LSTAT, REQ_FSTAT,
31 REQ_FSYNC, REQ_FDATASYNC, 35 REQ_FSYNC, REQ_FDATASYNC,
36 REQ_UNLINK, REQ_RMDIR,
37 REQ_SYMLINK,
32}; 38};
33 39
34typedef struct aio_cb { 40typedef struct aio_cb {
35 struct aio_cb *volatile next; 41 struct aio_cb *volatile next;
36 42
… …
40 off_t offset; 46 off_t offset;
41 size_t length; 47 size_t length;
42 ssize_t result; 48 ssize_t result;
43 mode_t mode; /* open */ 49 mode_t mode; /* open */
44 int errorno; 50 int errorno;
45 SV *data, *callback; 51 SV *data, *callback, *fh;
46 void *dataptr; 52 void *dataptr, *data2ptr;
47 STRLEN dataoffset; 53 STRLEN dataoffset;
48 54
49 Stat_t *statdata; 55 Stat_t *statdata;
50} aio_cb; 56} aio_cb;
51 57
52typedef aio_cb *aio_req; 58typedef aio_cb *aio_req;
53 59
54static int started; 60static int started;
55static int nreqs; 61static volatile int nreqs;
56static int max_outstanding = 1<<30; 62static int max_outstanding = 1<<30;
57static int respipe [2]; 63static int respipe [2];
58 64
59static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER; 65static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER;
60static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER; 66static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER;
61static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 67static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
62 68
63static volatile aio_req reqs, reqe; /* queue start, queue end */ 69static volatile aio_req reqs, reqe; /* queue start, queue end */
64static volatile aio_req ress, rese; /* queue start, queue end */ 70static volatile aio_req ress, rese; /* queue start, queue end */
65 71
72static void free_req (aio_req req)
73{
74 if (req->data)
75 SvREFCNT_dec (req->data);
76
77 if (req->fh)
78 SvREFCNT_dec (req->fh);
79
80 if (req->statdata)
81 Safefree (req->statdata);
82
83 if (req->callback)
84 SvREFCNT_dec (req->callback);
85
86 Safefree (req);
87}
88
66static void 89static void
67poll_wait () 90poll_wait ()
68{ 91{
69 if (!nreqs) 92 if (nreqs && !ress)
70 return; 93 {
71
72 fd_set rfd; 94 fd_set rfd;
73 FD_ZERO(&rfd); 95 FD_ZERO(&rfd);
74 FD_SET(respipe [0], &rfd); 96 FD_SET(respipe [0], &rfd);
75 97
76 select (respipe [0] + 1, &rfd, 0, 0, 0); 98 select (respipe [0] + 1, &rfd, 0, 0, 0);
99 }
77} 100}
78 101
79static int 102static int
80poll_cb () 103poll_cb ()
81{ 104{
82 dSP; 105 dSP;
83 int count = 0; 106 int count = 0;
107 int do_croak = 0;
84 aio_req req; 108 aio_req req;
85
86 {
87 /* read and signals sent by the worker threads */
88 char buf [32];
89 while (read (respipe [0], buf, 32) > 0)
90 ;
91 }
92 109
93 for (;;) 110 for (;;)
94 { 111 {
95 pthread_mutex_lock (&reslock); 112 pthread_mutex_lock (&reslock);
96
97 req = ress; 113 req = ress;
98 114
99 if (ress) 115 if (req)
100 { 116 {
101 ress = ress->next; 117 ress = req->next;
118
102 if (!ress) rese = 0; 119 if (!ress)
120 {
121 rese = 0;
122
123 /* read any signals sent by the worker threads */
124 char buf [32];
125 while (read (respipe [0], buf, 32) == 32)
126 ;
127 }
103 } 128 }
104 129
105 pthread_mutex_unlock (&reslock); 130 pthread_mutex_unlock (&reslock);
106 131
107 if (!req) 132 if (!req)
… …
115 { 140 {
116 int errorno = errno; 141 int errorno = errno;
117 errno = req->errorno; 142 errno = req->errorno;
118 143
119 if (req->type == REQ_READ) 144 if (req->type == REQ_READ)
120 SvCUR_set (req->data, req->dataoffset 145 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
121 + req->result > 0 ? req->result : 0);
122 146
123 if (req->data) 147 if (req->statdata)
124 SvREFCNT_dec (req->data);
125
126 if (req->type == REQ_STAT || req->type == REQ_LSTAT || req->type == REQ_FSTAT)
127 { 148 {
128 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT; 149 PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
129 PL_laststatval = req->result; 150 PL_laststatval = req->result;
130 PL_statcache = *(req->statdata); 151 PL_statcache = *(req->statdata);
131
132 Safefree (req->statdata);
133 } 152 }
134 153
154 ENTER;
135 PUSHMARK (SP); 155 PUSHMARK (SP);
136 XPUSHs (sv_2mortal (newSViv (req->result))); 156 XPUSHs (sv_2mortal (newSViv (req->result)));
137 157
138 if (req->type == REQ_OPEN) 158 if (req->type == REQ_OPEN)
139 { 159 {
… …
142 162
143 PUTBACK; 163 PUTBACK;
144 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL); 164 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
145 SPAGAIN; 165 SPAGAIN;
146 166
147 fh = POPs; 167 fh = SvREFCNT_inc (POPs);
148 168
149 PUSHMARK (SP); 169 PUSHMARK (SP);
150 XPUSHs (fh); 170 XPUSHs (sv_2mortal (fh));
151 } 171 }
152 172
153 if (SvOK (req->callback)) 173 if (SvOK (req->callback))
154 { 174 {
155 PUTBACK; 175 PUTBACK;
156 call_sv (req->callback, G_VOID | G_EVAL); 176 call_sv (req->callback, G_VOID | G_EVAL);
157 SPAGAIN; 177 SPAGAIN;
178
179 if (SvTRUE (ERRSV))
180 {
181 free_req (req);
182 croak (0);
183 }
158 } 184 }
185
159 186 LEAVE;
160 if (req->callback)
161 SvREFCNT_dec (req->callback);
162 187
163 errno = errorno; 188 errno = errorno;
164 count++; 189 count++;
165 } 190 }
166 191
167 Safefree (req); 192 free_req (req);
168 } 193 }
169 194
170 return count; 195 return count;
171} 196}
172 197
… …
210 reqe = reqs = req; 235 reqe = reqs = req;
211 236
212 pthread_cond_signal (&reqwait); 237 pthread_cond_signal (&reqwait);
213 pthread_mutex_unlock (&reqlock); 238 pthread_mutex_unlock (&reqlock);
214 239
215 while (nreqs > max_outstanding) 240 if (nreqs > max_outstanding)
241 for (;;)
242 {
243 poll_cb ();
244
245 if (nreqs <= max_outstanding)
246 break;
247
248 poll_wait ();
249 }
250}
251
252static void
253end_thread (void)
254{
255 aio_req req;
256 Newz (0, req, 1, aio_cb);
257 req->type = REQ_QUIT;
258
259 send_req (req);
260}
261
262static void min_parallel (int nthreads)
263{
264 while (nthreads > started)
265 start_thread ();
266}
267
268static void max_parallel (int nthreads)
269{
270 int cur = started;
271
272 while (cur > nthreads)
273 {
274 end_thread ();
275 cur--;
276 }
277
278 while (started > nthreads)
216 { 279 {
217 poll_wait (); 280 poll_wait ();
218 poll_cb (); 281 poll_cb ();
219 } 282 }
220} 283}
221 284
222static void 285static void create_pipe ()
223end_thread (void)
224{ 286{
225 aio_req req; 287 if (pipe (respipe))
226 New (0, req, 1, aio_cb); 288 croak ("unable to initialize result pipe");
227 req->type = REQ_QUIT;
228 289
229 send_req (req); 290 if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
230} 291 croak ("cannot set result pipe to nonblocking mode");
231 292
232static void 293 if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
233read_write (int dowrite, int fd, off_t offset, size_t length, 294 croak ("cannot set result pipe to nonblocking mode");
234 SV *data, STRLEN dataoffset, SV *callback) 295}
235{
236 aio_req req;
237 STRLEN svlen;
238 char *svptr = SvPV (data, svlen);
239 296
240 SvUPGRADE (data, SVt_PV); 297static void atfork_prepare (void)
241 SvPOK_on (data); 298{
299 for (;;) {
242 300
243 if (dataoffset < 0) 301 for (;;)
244 dataoffset += svlen; 302 {
303 poll_cb ();
245 304
246 if (dataoffset < 0 || dataoffset > svlen) 305 if (!nreqs)
247 croak ("data offset outside of string"); 306 break;
248 307
249 if (dowrite) 308 poll_wait ();
309 }
310
311 pthread_mutex_lock (&reqlock);
312
313 if (!nreqs)
314 break;
315
316 pthread_mutex_unlock (&reqlock);
317 }
318
319 pthread_mutex_lock (&reslock);
320
321 assert (!nreqs && !reqs && !ress);
322}
323
324static void atfork_parent (void)
325{
326 pthread_mutex_unlock (&reslock);
327 pthread_mutex_unlock (&reqlock);
328}
329
330static void atfork_child (void)
331{
332 int restart = started;
333 started = 0;
334
335 atfork_parent ();
336
337 min_parallel (restart);
338}
339
340/*****************************************************************************/
341/* work around various missing functions */
342
343#if !HAVE_PREADWRITE
344# define pread aio_pread
345# define pwrite aio_pwrite
346
347/*
348 * make our pread/pwrite safe against themselves, but not against
349 * normal read/write by using a mutex. slows down execution a lot,
350 * but that's your problem, not mine.
351 */
352static pthread_mutex_t iolock = PTHREAD_MUTEX_INITIALIZER;
353
354static ssize_t
355pread (int fd, void *buf, size_t count, off_t offset)
356{
357 ssize_t res;
358 off_t ooffset;
359
360 pthread_mutex_lock (&iolock);
361 ooffset = lseek (fd, 0, SEEK_CUR);
362 lseek (fd, offset, SEEK_SET);
363 res = read (fd, buf, count);
364 lseek (fd, ooffset, SEEK_SET);
365 pthread_mutex_unlock (&iolock);
366
367 return res;
368}
369
370static ssize_t
371pwrite (int fd, void *buf, size_t count, off_t offset)
372{
373 ssize_t res;
374 off_t ooffset;
375
376 pthread_mutex_lock (&iolock);
377 ooffset = lseek (fd, 0, SEEK_CUR);
378 lseek (fd, offset, SEEK_SET);
379 res = write (fd, buf, count);
380 lseek (fd, offset, SEEK_SET);
381 pthread_mutex_unlock (&iolock);
382
383 return res;
384}
385#endif
386
387#if !HAVE_FDATASYNC
388# define fdatasync fsync
389#endif
390
391#if !HAVE_READAHEAD
392# define readahead aio_readahead
393
394static char readahead_buf[4096];
395
396static ssize_t
397readahead (int fd, off_t offset, size_t count)
398{
399 while (count > 0)
250 { 400 {
251 /* write: check length and adjust. */ 401 size_t len = count < sizeof (readahead_buf) ? count : sizeof (readahead_buf);
252 if (length < 0 || length + dataoffset > svlen) 402
253 length = svlen - dataoffset; 403 pread (fd, readahead_buf, len, offset);
404 offset += len;
405 count -= len;
254 } 406 }
255 else
256 {
257 /* read: grow scalar as necessary */
258 svptr = SvGROW (data, length + dataoffset);
259 }
260 407
261 if (length < 0) 408 errno = 0;
262 croak ("length must not be negative");
263
264 Newz (0, req, 1, aio_cb);
265
266 if (!req)
267 croak ("out of memory during aio_req allocation");
268
269 req->type = dowrite ? REQ_WRITE : REQ_READ;
270 req->fd = fd;
271 req->offset = offset;
272 req->length = length;
273 req->data = SvREFCNT_inc (data);
274 req->dataptr = (char *)svptr + dataoffset;
275 req->callback = SvREFCNT_inc (callback);
276
277 send_req (req);
278} 409}
410#endif
411
412/*****************************************************************************/
279 413
280static void * 414static void *
281aio_proc (void *thr_arg) 415aio_proc (void *thr_arg)
282{ 416{
283 aio_req req; 417 aio_req req;
… …
309 443
310 type = req->type; 444 type = req->type;
311 445
312 switch (type) 446 switch (type)
313 { 447 {
314 case REQ_READ: req->result = pread64 (req->fd, req->dataptr, req->length, req->offset); break; 448 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
315 case REQ_WRITE: req->result = pwrite64 (req->fd, req->dataptr, req->length, req->offset); break; 449 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
316#if SYS_readahead 450
317 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 451 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
318#else
319 case REQ_READAHEAD: req->result = -1; errno = ENOSYS; break;
320#endif
321 452
322 case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break; 453 case REQ_STAT: req->result = stat (req->dataptr, req->statdata); break;
323 case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break; 454 case REQ_LSTAT: req->result = lstat (req->dataptr, req->statdata); break;
324 case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break; 455 case REQ_FSTAT: req->result = fstat (req->fd , req->statdata); break;
325 456
326 case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break; 457 case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break;
327 case REQ_CLOSE: req->result = close (req->fd); break; 458 case REQ_CLOSE: req->result = close (req->fd); break;
328 case REQ_UNLINK: req->result = unlink (req->dataptr); break; 459 case REQ_UNLINK: req->result = unlink (req->dataptr); break;
460 case REQ_RMDIR: req->result = rmdir (req->dataptr); break;
461 case REQ_SYMLINK: req->result = symlink (req->data2ptr, req->dataptr); break;
329 462
463 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
330 case REQ_FSYNC: req->result = fsync (req->fd); break; 464 case REQ_FSYNC: req->result = fsync (req->fd); break;
331 case REQ_FDATASYNC: req->result = fdatasync (req->fd); break;
332 465
333 case REQ_QUIT: 466 case REQ_QUIT:
334 break; 467 break;
335 468
336 default: 469 default:
… …
362 while (type != REQ_QUIT); 495 while (type != REQ_QUIT);
363 496
364 return 0; 497 return 0;
365} 498}
366 499
500#define dREQ \
501 aio_req req; \
502 \
503 if (SvOK (callback) && !SvROK (callback)) \
504 croak ("clalback must be undef or of reference type"); \
505 \
506 Newz (0, req, 1, aio_cb); \
507 if (!req) \
508 croak ("out of memory during aio_req allocation"); \
509 \
510 req->callback = newSVsv (callback);
511
367MODULE = IO::AIO PACKAGE = IO::AIO 512MODULE = IO::AIO PACKAGE = IO::AIO
368 513
369PROTOTYPES: ENABLE 514PROTOTYPES: ENABLE
370 515
371BOOT: 516BOOT:
372{ 517{
373 if (pipe (respipe)) 518 create_pipe ();
374 croak ("unable to initialize result pipe"); 519 pthread_atfork (atfork_prepare, atfork_parent, atfork_child);
375
376 if (fcntl (respipe [0], F_SETFL, O_NONBLOCK))
377 croak ("cannot set result pipe to nonblocking mode");
378
379 if (fcntl (respipe [1], F_SETFL, O_NONBLOCK))
380 croak ("cannot set result pipe to nonblocking mode");
381} 520}
382 521
383void 522void
384min_parallel(nthreads) 523min_parallel(nthreads)
385 int nthreads 524 int nthreads
386 PROTOTYPE: $ 525 PROTOTYPE: $
387 CODE:
388 while (nthreads > started)
389 start_thread ();
390 526
391void 527void
392max_parallel(nthreads) 528max_parallel(nthreads)
393 int nthreads 529 int nthreads
394 PROTOTYPE: $ 530 PROTOTYPE: $
395 CODE:
396{
397 int cur = started;
398 while (cur > nthreads)
399 {
400 end_thread ();
401 cur--;
402 }
403
404 while (started > nthreads)
405 {
406 poll_wait ();
407 poll_cb ();
408 }
409}
410 531
411int 532int
412max_outstanding(nreqs) 533max_outstanding(nreqs)
413 int nreqs 534 int nreqs
414 PROTOTYPE: $ 535 PROTOTYPE: $
… …
423 int mode 544 int mode
424 SV * callback 545 SV * callback
425 PROTOTYPE: $$$;$ 546 PROTOTYPE: $$$;$
426 CODE: 547 CODE:
427{ 548{
428 aio_req req; 549 dREQ;
429
430 Newz (0, req, 1, aio_cb);
431
432 if (!req)
433 croak ("out of memory during aio_req allocation");
434 550
435 req->type = REQ_OPEN; 551 req->type = REQ_OPEN;
436 req->data = newSVsv (pathname); 552 req->data = newSVsv (pathname);
437 req->dataptr = SvPV_nolen (req->data); 553 req->dataptr = SvPVbyte_nolen (req->data);
438 req->fd = flags; 554 req->fd = flags;
439 req->mode = mode; 555 req->mode = mode;
440 req->callback = SvREFCNT_inc (callback);
441 556
442 send_req (req); 557 send_req (req);
443} 558}
444 559
445void 560void
446aio_close(fh,callback=&PL_sv_undef) 561aio_close(fh,callback=&PL_sv_undef)
447 InputStream fh 562 SV * fh
448 SV * callback 563 SV * callback
449 PROTOTYPE: $;$ 564 PROTOTYPE: $;$
450 ALIAS: 565 ALIAS:
451 aio_close = REQ_CLOSE 566 aio_close = REQ_CLOSE
452 aio_fsync = REQ_FSYNC 567 aio_fsync = REQ_FSYNC
453 aio_fdatasync = REQ_FDATASYNC 568 aio_fdatasync = REQ_FDATASYNC
454 CODE: 569 CODE:
455{ 570{
456 aio_req req; 571 dREQ;
457
458 Newz (0, req, 1, aio_cb);
459
460 if (!req)
461 croak ("out of memory during aio_req allocation");
462 572
463 req->type = ix; 573 req->type = ix;
574 req->fh = newSVsv (fh);
464 req->fd = PerlIO_fileno (fh); 575 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
465 req->callback = SvREFCNT_inc (callback);
466 576
467 send_req (req); 577 send_req (req);
468} 578}
469 579
470void 580void
471aio_read(fh,offset,length,data,dataoffset,callback=&PL_sv_undef) 581aio_read(fh,offset,length,data,dataoffset,callback=&PL_sv_undef)
472 InputStream fh 582 SV * fh
473 UV offset 583 UV offset
474 IV length 584 IV length
475 SV * data 585 SV * data
476 IV dataoffset 586 IV dataoffset
477 SV * callback 587 SV * callback
588 ALIAS:
589 aio_read = REQ_READ
590 aio_write = REQ_WRITE
478 PROTOTYPE: $$$$$;$ 591 PROTOTYPE: $$$$$;$
479 CODE: 592 CODE:
480 read_write (0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback); 593{
594 aio_req req;
595 STRLEN svlen;
596 char *svptr = SvPVbyte (data, svlen);
481 597
482void 598 SvUPGRADE (data, SVt_PV);
483aio_write(fh,offset,length,data,dataoffset,callback=&PL_sv_undef) 599 SvPOK_on (data);
484 OutputStream fh 600
485 UV offset 601 if (dataoffset < 0)
486 IV length 602 dataoffset += svlen;
487 SV * data 603
488 IV dataoffset 604 if (dataoffset < 0 || dataoffset > svlen)
489 SV * callback 605 croak ("data offset outside of string");
490 PROTOTYPE: $$$$$;$ 606
491 CODE: 607 if (ix == REQ_WRITE)
492 read_write (1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback); 608 {
609 /* write: check length and adjust. */
610 if (length < 0 || length + dataoffset > svlen)
611 length = svlen - dataoffset;
612 }
613 else
614 {
615 /* read: grow scalar as necessary */
616 svptr = SvGROW (data, length + dataoffset);
617 }
618
619 if (length < 0)
620 croak ("length must not be negative");
621
622 {
623 dREQ;
624
625 req->type = ix;
626 req->fh = newSVsv (fh);
627 req->fd = PerlIO_fileno (ix == REQ_READ ? IoIFP (sv_2io (fh))
628 : IoOFP (sv_2io (fh)));
629 req->offset = offset;
630 req->length = length;
631 req->data = SvREFCNT_inc (data);
632 req->dataptr = (char *)svptr + dataoffset;
633
634 send_req (req);
635 }
636}
493 637
494void 638void
495aio_readahead(fh,offset,length,callback=&PL_sv_undef) 639aio_readahead(fh,offset,length,callback=&PL_sv_undef)
496 InputStream fh 640 SV * fh
497 UV offset 641 UV offset
498 IV length 642 IV length
499 SV * callback 643 SV * callback
500 PROTOTYPE: $$$;$ 644 PROTOTYPE: $$$;$
501 CODE: 645 CODE:
502{ 646{
503 aio_req req; 647 dREQ;
504
505 if (length < 0)
506 croak ("length must not be negative");
507
508 Newz (0, req, 1, aio_cb);
509
510 if (!req)
511 croak ("out of memory during aio_req allocation");
512 648
513 req->type = REQ_READAHEAD; 649 req->type = REQ_READAHEAD;
650 req->fh = newSVsv (fh);
514 req->fd = PerlIO_fileno (fh); 651 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh)));
515 req->offset = offset; 652 req->offset = offset;
516 req->length = length; 653 req->length = length;
517 req->callback = SvREFCNT_inc (callback);
518 654
519 send_req (req); 655 send_req (req);
520} 656}
521 657
522void 658void
… …
526 ALIAS: 662 ALIAS:
527 aio_stat = REQ_STAT 663 aio_stat = REQ_STAT
528 aio_lstat = REQ_LSTAT 664 aio_lstat = REQ_LSTAT
529 CODE: 665 CODE:
530{ 666{
531 aio_req req; 667 dREQ;
532
533 Newz (0, req, 1, aio_cb);
534
535 if (!req)
536 croak ("out of memory during aio_req allocation");
537 668
538 New (0, req->statdata, 1, Stat_t); 669 New (0, req->statdata, 1, Stat_t);
539
540 if (!req->statdata) 670 if (!req->statdata)
671 {
672 free_req (req);
541 croak ("out of memory during aio_req->statdata allocation"); 673 croak ("out of memory during aio_req->statdata allocation");
674 }
542 675
543 if (SvPOK (fh_or_path)) 676 if (SvPOK (fh_or_path))
544 { 677 {
545 req->type = ix; 678 req->type = ix;
546 req->data = newSVsv (fh_or_path); 679 req->data = newSVsv (fh_or_path);
547 req->dataptr = SvPV_nolen (req->data); 680 req->dataptr = SvPVbyte_nolen (req->data);
548 } 681 }
549 else 682 else
550 { 683 {
551 req->type = REQ_FSTAT; 684 req->type = REQ_FSTAT;
685 req->fh = newSVsv (fh_or_path);
552 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path))); 686 req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
553 } 687 }
554
555 req->callback = SvREFCNT_inc (callback);
556 688
557 send_req (req); 689 send_req (req);
558} 690}
559 691
560void 692void
561aio_unlink(pathname,callback=&PL_sv_undef) 693aio_unlink(pathname,callback=&PL_sv_undef)
562 SV * pathname 694 SV * pathname
563 SV * callback 695 SV * callback
696 ALIAS:
697 aio_unlink = REQ_UNLINK
698 aio_rmdir = REQ_RMDIR
564 CODE: 699 CODE:
565{ 700{
566 aio_req req; 701 dREQ;
567 702
568 Newz (0, req, 1, aio_cb); 703 req->type = ix;
704 req->data = newSVsv (pathname);
705 req->dataptr = SvPVbyte_nolen (req->data);
569 706
570 if (!req) 707 send_req (req);
571 croak ("out of memory during aio_req allocation"); 708}
709
710void
711aio_symlink(oldpath,newpath,callback=&PL_sv_undef)
712 SV * oldpath
713 SV * newpath
714 SV * callback
715 CODE:
716{
717 dREQ;
572 718
573 req->type = REQ_UNLINK; 719 req->type = REQ_SYMLINK;
720 req->fh = newSVsv (oldpath);
721 req->data2ptr = SvPVbyte_nolen (req->fh);
574 req->data = newSVsv (pathname); 722 req->data = newSVsv (newpath);
575 req->dataptr = SvPV_nolen (req->data); 723 req->dataptr = SvPVbyte_nolen (req->data);
576 req->callback = SvREFCNT_inc (callback);
577 724
578 send_req (req); 725 send_req (req);
579} 726}
580 727
581void 728void

Diff Legend

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