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.50 by root, Sun Oct 22 21:13:47 2006 UTC vs.
Revision 1.62 by root, Mon Oct 23 22:45:18 2006 UTC

… …
55 REQ_STAT, REQ_LSTAT, REQ_FSTAT, 55 REQ_STAT, REQ_LSTAT, REQ_FSTAT,
56 REQ_FSYNC, REQ_FDATASYNC, 56 REQ_FSYNC, REQ_FDATASYNC,
57 REQ_UNLINK, REQ_RMDIR, REQ_RENAME, 57 REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
58 REQ_READDIR, 58 REQ_READDIR,
59 REQ_LINK, REQ_SYMLINK, 59 REQ_LINK, REQ_SYMLINK,
60 REQ_GROUP, REQ_NOP,
60 REQ_SLEEP, 61 REQ_SLEEP,
61 REQ_GROUP,
62}; 62};
63 63
64#define AIO_REQ_KLASS "IO::AIO::REQ" 64#define AIO_REQ_KLASS "IO::AIO::REQ"
65#define AIO_GRP_KLASS "IO::AIO::GRP" 65#define AIO_GRP_KLASS "IO::AIO::GRP"
66 66
67typedef struct aio_cb 67typedef struct aio_cb
68{ 68{
69 struct aio_cb *volatile next; 69 struct aio_cb *volatile next;
70
71 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
72
73 SV *self; /* the perl counterpart of this request, if any */
74 70
75 SV *data, *callback; 71 SV *data, *callback;
76 SV *fh, *fh2; 72 SV *fh, *fh2;
77 void *dataptr, *data2ptr; 73 void *dataptr, *data2ptr;
78 Stat_t *statdata; 74 Stat_t *statdata;
79 off_t offset; 75 off_t offset;
80 size_t length; 76 size_t length;
81 ssize_t result; 77 ssize_t result;
82 78
79 STRLEN dataoffset;
83 int type; 80 int type;
84 int fd, fd2; 81 int fd, fd2;
85 int errorno; 82 int errorno;
86 STRLEN dataoffset;
87 mode_t mode; /* open */ 83 mode_t mode; /* open */
84
88 unsigned char cancelled; 85 unsigned char flags;
86 unsigned char pri;
87
88 SV *self; /* the perl counterpart of this request, if any */
89 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
89} aio_cb; 90} aio_cb;
91
92enum {
93 FLAG_CANCELLED = 0x01,
94};
90 95
91typedef aio_cb *aio_req; 96typedef aio_cb *aio_req;
92typedef aio_cb *aio_req_ornot; 97typedef aio_cb *aio_req_ornot;
98
99enum {
100 PRI_MIN = -4,
101 PRI_MAX = 4,
102
103 DEFAULT_PRI = 0,
104 PRI_BIAS = -PRI_MIN,
105};
106
107static int next_pri = DEFAULT_PRI + PRI_BIAS;
93 108
94static int started, wanted; 109static int started, wanted;
95static volatile int nreqs; 110static volatile int nreqs;
96static int max_outstanding = 1<<30; 111static int max_outstanding = 1<<30;
97static int respipe [2]; 112static int respipe [2];
… …
118 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1))); 133 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1)));
119} 134}
120 135
121static aio_req SvAIO_REQ (SV *sv) 136static aio_req SvAIO_REQ (SV *sv)
122{ 137{
138 MAGIC *mg;
139
123 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv)) 140 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv))
124 croak ("object of class " AIO_REQ_KLASS " expected"); 141 croak ("object of class " AIO_REQ_KLASS " expected");
125 142
126 MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext); 143 mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
127 144
128 return mg ? (aio_req)mg->mg_ptr : 0; 145 return mg ? (aio_req)mg->mg_ptr : 0;
129} 146}
130 147
131static void aio_grp_feed (aio_req grp) 148static void aio_grp_feed (aio_req grp)
132{ 149{
133 while (grp->length < grp->fd2) 150 while (grp->length < grp->fd2 && !(grp->flags & FLAG_CANCELLED))
134 { 151 {
135 int old_len = grp->length; 152 int old_len = grp->length;
136 153
137 if (grp->fh2 && SvOK (grp->fh2)) 154 if (grp->fh2 && SvOK (grp->fh2))
138 { 155 {
… …
174 } 191 }
175} 192}
176 193
177static void poll_wait () 194static void poll_wait ()
178{ 195{
179 if (nreqs && !ress)
180 {
181 fd_set rfd; 196 fd_set rfd;
197
198 while (nreqs)
199 {
200 aio_req req;
201 pthread_mutex_lock (&reslock);
202 req = ress;
203 pthread_mutex_unlock (&reslock);
204
205 if (req)
206 return;
207
182 FD_ZERO(&rfd); 208 FD_ZERO(&rfd);
183 FD_SET(respipe [0], &rfd); 209 FD_SET(respipe [0], &rfd);
184 210
185 select (respipe [0] + 1, &rfd, 0, 0, 0); 211 select (respipe [0] + 1, &rfd, 0, 0, 0);
186 } 212 }
… …
189static void req_invoke (aio_req req) 215static void req_invoke (aio_req req)
190{ 216{
191 dSP; 217 dSP;
192 int errorno = errno; 218 int errorno = errno;
193 219
194 if (req->cancelled || !SvOK (req->callback)) 220 if (req->flags & FLAG_CANCELLED || !SvOK (req->callback))
195 return; 221 return;
196 222
197 errno = req->errorno; 223 errno = req->errorno;
198 224
199 ENTER; 225 ENTER;
… …
257 for (i = 0; i <= AvFILL (av); ++i) 283 for (i = 0; i <= AvFILL (av); ++i)
258 PUSHs (*av_fetch (av, i, 0)); 284 PUSHs (*av_fetch (av, i, 0));
259 } 285 }
260 break; 286 break;
261 287
288 case REQ_NOP:
262 case REQ_SLEEP: 289 case REQ_SLEEP:
263 break; 290 break;
264 291
265 default: 292 default:
266 PUSHs (sv_2mortal (newSViv (req->result))); 293 PUSHs (sv_2mortal (newSViv (req->result)));
… …
270 297
271 PUTBACK; 298 PUTBACK;
272 call_sv (req->callback, G_VOID | G_EVAL); 299 call_sv (req->callback, G_VOID | G_EVAL);
273 SPAGAIN; 300 SPAGAIN;
274 301
302 FREETMPS;
303 LEAVE;
304
305 errno = errorno;
306
275 if (SvTRUE (ERRSV)) 307 if (SvTRUE (ERRSV))
276 { 308 {
277 req_free (req); 309 req_free (req);
278 croak (0); 310 croak (0);
279 } 311 }
280
281 FREETMPS;
282 LEAVE;
283
284 errno = errorno;
285} 312}
286 313
287static void req_free (aio_req req) 314static void req_free (aio_req req)
288{ 315{
289 if (req->grp) 316 if (req->grp)
… …
318 Safefree (req); 345 Safefree (req);
319} 346}
320 347
321static void req_cancel (aio_req req) 348static void req_cancel (aio_req req)
322{ 349{
323 req->cancelled = 1; 350 req->flags |= FLAG_CANCELLED;
324 351
325 if (req->type == REQ_GROUP) 352 if (req->type == REQ_GROUP)
326 { 353 {
327 aio_req sub; 354 aio_req sub;
328 355
… …
782 } 809 }
783 810
784 pthread_mutex_unlock (&reqlock); 811 pthread_mutex_unlock (&reqlock);
785 812
786 errno = 0; /* strictly unnecessary */ 813 errno = 0; /* strictly unnecessary */
787
788 if (!req->cancelled)
789 switch (type = req->type) /* remember type for QUIT check */ 814 type = req->type; /* remember type for QUIT check */
815
816 if (!(req->flags & FLAG_CANCELLED))
817 switch (type)
790 { 818 {
791 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 819 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
792 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 820 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
793 821
794 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 822 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
… …
818 tv.tv_usec = req->fd2; 846 tv.tv_usec = req->fd2;
819 847
820 req->result = select (0, 0, 0, 0, &tv); 848 req->result = select (0, 0, 0, 0, &tv);
821 } 849 }
822 850
851 case REQ_GROUP:
852 case REQ_NOP:
823 case REQ_QUIT: 853 case REQ_QUIT:
824 break; 854 break;
825 855
826 default: 856 default:
827 req->result = ENOSYS; 857 req->result = ENOSYS;
… …
911 atfork_parent (); 941 atfork_parent ();
912} 942}
913 943
914#define dREQ \ 944#define dREQ \
915 aio_req req; \ 945 aio_req req; \
946 int req_pri = next_pri; \
947 next_pri = DEFAULT_PRI + PRI_BIAS; \
916 \ 948 \
917 if (SvOK (callback) && !SvROK (callback)) \ 949 if (SvOK (callback) && !SvROK (callback)) \
918 croak ("callback must be undef or of reference type"); \ 950 croak ("callback must be undef or of reference type"); \
919 \ 951 \
920 Newz (0, req, 1, aio_cb); \ 952 Newz (0, req, 1, aio_cb); \
921 if (!req) \ 953 if (!req) \
922 croak ("out of memory during aio_req allocation"); \ 954 croak ("out of memory during aio_req allocation"); \
923 \ 955 \
924 req->callback = newSVsv (callback) 956 req->callback = newSVsv (callback); \
957 req->pri = req_pri
925 958
926#define REQ_SEND \ 959#define REQ_SEND \
927 req_send (req); \ 960 req_send (req); \
928 \ 961 \
929 if (GIMME_V != G_VOID) \ 962 if (GIMME_V != G_VOID) \
… …
1204 SV * callback 1237 SV * callback
1205 PROTOTYPE: ;$ 1238 PROTOTYPE: ;$
1206 PPCODE: 1239 PPCODE:
1207{ 1240{
1208 dREQ; 1241 dREQ;
1242
1209 req->type = REQ_GROUP; 1243 req->type = REQ_GROUP;
1210 req_send (req); 1244 req_send (req);
1245
1211 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1246 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1212} 1247}
1248
1249void
1250aio_nop (callback=&PL_sv_undef)
1251 SV * callback
1252 PPCODE:
1253{
1254 dREQ;
1255
1256 req->type = REQ_NOP;
1257
1258 REQ_SEND;
1259}
1260
1261#if 0
1262
1263void
1264aio_pri (int pri = DEFAULT_PRI)
1265 CODE:
1266 if (pri < PRI_MIN) pri = PRI_MIN;
1267 if (pri > PRI_MAX) pri = PRI_MAX;
1268 next_pri = pri + PRI_BIAS;
1269
1270#endif
1213 1271
1214void 1272void
1215flush () 1273flush ()
1216 PROTOTYPE: 1274 PROTOTYPE:
1217 CODE: 1275 CODE:
… …
1270cancel (aio_req_ornot req) 1328cancel (aio_req_ornot req)
1271 PROTOTYPE: 1329 PROTOTYPE:
1272 CODE: 1330 CODE:
1273 req_cancel (req); 1331 req_cancel (req);
1274 1332
1333void
1334cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1335 CODE:
1336 SvREFCNT_dec (req->callback);
1337 req->callback = newSVsv (callback);
1338
1275MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1339MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1276 1340
1277void 1341void
1278add (aio_req grp, ...) 1342add (aio_req grp, ...)
1279 PPCODE: 1343 PPCODE:
1280{ 1344{
1281 int i; 1345 int i;
1346 aio_req req;
1282 1347
1283 if (grp->fd == 2) 1348 if (grp->fd == 2)
1284 croak ("cannot add requests to IO::AIO::GRP after the group finished"); 1349 croak ("cannot add requests to IO::AIO::GRP after the group finished");
1285 1350
1286 for (i = 1; i < items; ++i ) 1351 for (i = 1; i < items; ++i )
1287 { 1352 {
1288 if (GIMME_V != G_VOID) 1353 if (GIMME_V != G_VOID)
1289 XPUSHs (sv_2mortal (newSVsv (ST (i)))); 1354 XPUSHs (sv_2mortal (newSVsv (ST (i))));
1290 1355
1291 aio_req req = SvAIO_REQ (ST (i)); 1356 req = SvAIO_REQ (ST (i));
1292 1357
1293 if (req) 1358 if (req)
1294 { 1359 {
1295 ++grp->length; 1360 ++grp->length;
1296 req->grp = grp; 1361 req->grp = grp;
… …
1305 } 1370 }
1306 } 1371 }
1307} 1372}
1308 1373
1309void 1374void
1310lock (aio_req grp) 1375result (aio_req grp, ...)
1311 CODE: 1376 CODE:
1312 ++grp->length; 1377{
1378 int i;
1379 AV *av = newAV ();
1313 1380
1314void 1381 for (i = 1; i < items; ++i )
1315unlock (aio_req grp) 1382 av_push (av, newSVsv (ST (i)));
1316 CODE:
1317 aio_grp_dec (grp);
1318 1383
1384 SvREFCNT_dec (grp->data);
1385 grp->data = (SV *)av;
1386}
1387
1319void 1388void
1320feeder_limit (aio_req grp, int limit) 1389feed_limit (aio_req grp, int limit)
1321 CODE: 1390 CODE:
1322 grp->fd2 = limit; 1391 grp->fd2 = limit;
1323 aio_grp_feed (grp); 1392 aio_grp_feed (grp);
1324 1393
1325void 1394void
1326set_feeder (aio_req grp, SV *callback=&PL_sv_undef) 1395feed (aio_req grp, SV *callback=&PL_sv_undef)
1327 CODE: 1396 CODE:
1328{ 1397{
1329 SvREFCNT_dec (grp->fh2); 1398 SvREFCNT_dec (grp->fh2);
1330 grp->fh2 = newSVsv (callback); 1399 grp->fh2 = newSVsv (callback);
1331 1400

Diff Legend

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