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.49 by root, Sun Oct 22 13:33:28 2006 UTC vs.
Revision 1.61 by root, Mon Oct 23 22:44:21 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];
… …
101static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 116static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
102 117
103static volatile aio_req reqs, reqe; /* queue start, queue end */ 118static volatile aio_req reqs, reqe; /* queue start, queue end */
104static volatile aio_req ress, rese; /* queue start, queue end */ 119static volatile aio_req ress, rese; /* queue start, queue end */
105 120
121static void req_invoke (aio_req req);
106static void req_free (aio_req req); 122static void req_free (aio_req req);
107 123
108/* must be called at most once */ 124/* must be called at most once */
109static SV *req_sv (aio_req req, const char *klass) 125static SV *req_sv (aio_req req, const char *klass)
110{ 126{
… …
117 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)));
118} 134}
119 135
120static aio_req SvAIO_REQ (SV *sv) 136static aio_req SvAIO_REQ (SV *sv)
121{ 137{
138 MAGIC *mg;
139
122 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv)) 140 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv))
123 croak ("object of class " AIO_REQ_KLASS " expected"); 141 croak ("object of class " AIO_REQ_KLASS " expected");
124 142
125 MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext); 143 mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
126 144
127 return mg ? (aio_req)mg->mg_ptr : 0; 145 return mg ? (aio_req)mg->mg_ptr : 0;
128} 146}
129 147
130static void aio_grp_feed (aio_req grp) 148static void aio_grp_feed (aio_req grp)
131{ 149{
132 while (grp->length < grp->fd2) 150 while (grp->length < grp->fd2 && !(grp->flags & FLAG_CANCELLED))
133 { 151 {
134 int old_len = grp->length; 152 int old_len = grp->length;
135 153
136 if (grp->fh2 && SvOK (grp->fh2)) 154 if (grp->fh2 && SvOK (grp->fh2))
137 { 155 {
… …
156 break; 174 break;
157 } 175 }
158 } 176 }
159} 177}
160 178
179static void aio_grp_dec (aio_req grp)
180{
181 --grp->length;
182
183 /* call feeder, if applicable */
184 aio_grp_feed (grp);
185
186 /* finish, if done */
187 if (!grp->length && grp->fd)
188 {
189 req_invoke (grp);
190 req_free (grp);
191 }
192}
193
161static void poll_wait () 194static void poll_wait ()
162{ 195{
163 if (nreqs && !ress) 196 while (nreqs)
164 { 197 {
198 aio_req req;
199 pthread_mutex_lock (&reslock);
200 req = ress;
201 pthread_mutex_unlock (&reslock);
202
203 if (req)
204 return;
205
165 fd_set rfd; 206 fd_set rfd;
166 FD_ZERO(&rfd); 207 FD_ZERO(&rfd);
167 FD_SET(respipe [0], &rfd); 208 FD_SET(respipe [0], &rfd);
168 209
169 select (respipe [0] + 1, &rfd, 0, 0, 0); 210 select (respipe [0] + 1, &rfd, 0, 0, 0);
… …
173static void req_invoke (aio_req req) 214static void req_invoke (aio_req req)
174{ 215{
175 dSP; 216 dSP;
176 int errorno = errno; 217 int errorno = errno;
177 218
178 if (req->cancelled || !SvOK (req->callback)) 219 if (req->flags & FLAG_CANCELLED || !SvOK (req->callback))
179 return; 220 return;
180 221
181 errno = req->errorno; 222 errno = req->errorno;
182 223
183 ENTER; 224 ENTER;
… …
241 for (i = 0; i <= AvFILL (av); ++i) 282 for (i = 0; i <= AvFILL (av); ++i)
242 PUSHs (*av_fetch (av, i, 0)); 283 PUSHs (*av_fetch (av, i, 0));
243 } 284 }
244 break; 285 break;
245 286
287 case REQ_NOP:
246 case REQ_SLEEP: 288 case REQ_SLEEP:
247 break; 289 break;
248 290
249 default: 291 default:
250 PUSHs (sv_2mortal (newSViv (req->result))); 292 PUSHs (sv_2mortal (newSViv (req->result)));
… …
254 296
255 PUTBACK; 297 PUTBACK;
256 call_sv (req->callback, G_VOID | G_EVAL); 298 call_sv (req->callback, G_VOID | G_EVAL);
257 SPAGAIN; 299 SPAGAIN;
258 300
301 FREETMPS;
302 LEAVE;
303
304 errno = errorno;
305
259 if (SvTRUE (ERRSV)) 306 if (SvTRUE (ERRSV))
260 { 307 {
261 req_free (req); 308 req_free (req);
262 croak (0); 309 croak (0);
263 } 310 }
264
265 FREETMPS;
266 LEAVE;
267
268 errno = errorno;
269} 311}
270 312
271static void req_free (aio_req req) 313static void req_free (aio_req req)
272{ 314{
273 if (req->grp) 315 if (req->grp)
… …
279 if (req->grp_prev) req->grp_prev->grp_next = req->grp_next; 321 if (req->grp_prev) req->grp_prev->grp_next = req->grp_next;
280 322
281 if (grp->grp_first == req) 323 if (grp->grp_first == req)
282 grp->grp_first = req->grp_next; 324 grp->grp_first = req->grp_next;
283 325
284 --grp->length;
285
286 /* call feeder, if applicable */
287 aio_grp_feed (grp); 326 aio_grp_dec (grp);
288
289 /* finish, if done */
290 if (!grp->length && grp->fd)
291 {
292 req_invoke (grp);
293 req_free (grp);
294 }
295 } 327 }
296 328
297 if (req->self) 329 if (req->self)
298 { 330 {
299 sv_unmagic (req->self, PERL_MAGIC_ext); 331 sv_unmagic (req->self, PERL_MAGIC_ext);
… …
312 Safefree (req); 344 Safefree (req);
313} 345}
314 346
315static void req_cancel (aio_req req) 347static void req_cancel (aio_req req)
316{ 348{
317 req->cancelled = 1; 349 req->flags |= FLAG_CANCELLED;
318 350
319 if (req->type == REQ_GROUP) 351 if (req->type == REQ_GROUP)
320 { 352 {
321 aio_req sub; 353 aio_req sub;
322 354
… …
355 pthread_mutex_unlock (&reslock); 387 pthread_mutex_unlock (&reslock);
356 388
357 if (!req) 389 if (!req)
358 break; 390 break;
359 391
360 nreqs--; 392 --nreqs;
361 393
362 if (req->type == REQ_QUIT) 394 if (req->type == REQ_QUIT)
363 started--; 395 started--;
364 else if (req->type == REQ_GROUP && req->length) 396 else if (req->type == REQ_GROUP && req->length)
365 { 397 {
… …
416static void req_send (aio_req req) 448static void req_send (aio_req req)
417{ 449{
418 while (started < wanted && nreqs >= started) 450 while (started < wanted && nreqs >= started)
419 start_thread (); 451 start_thread ();
420 452
421 nreqs++; 453 ++nreqs;
422 454
423 pthread_mutex_lock (&reqlock); 455 pthread_mutex_lock (&reqlock);
424 456
425 req->next = 0; 457 req->next = 0;
426 458
… …
776 } 808 }
777 809
778 pthread_mutex_unlock (&reqlock); 810 pthread_mutex_unlock (&reqlock);
779 811
780 errno = 0; /* strictly unnecessary */ 812 errno = 0; /* strictly unnecessary */
813 type = req->type; /* remember type for QUIT check */
781 814
782 if (!req->cancelled) 815 if (!(req->flags & FLAG_CANCELLED))
783 switch (req->type) 816 switch (type)
784 { 817 {
785 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 818 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
786 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 819 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
787 820
788 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 821 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
… …
812 tv.tv_usec = req->fd2; 845 tv.tv_usec = req->fd2;
813 846
814 req->result = select (0, 0, 0, 0, &tv); 847 req->result = select (0, 0, 0, 0, &tv);
815 } 848 }
816 849
850 case REQ_GROUP:
851 case REQ_NOP:
817 case REQ_QUIT: 852 case REQ_QUIT:
818 break; 853 break;
819 854
820 default: 855 default:
821 req->result = ENOSYS; 856 req->result = ENOSYS;
… …
905 atfork_parent (); 940 atfork_parent ();
906} 941}
907 942
908#define dREQ \ 943#define dREQ \
909 aio_req req; \ 944 aio_req req; \
945 int req_pri = next_pri; \
946 next_pri = DEFAULT_PRI + PRI_BIAS; \
910 \ 947 \
911 if (SvOK (callback) && !SvROK (callback)) \ 948 if (SvOK (callback) && !SvROK (callback)) \
912 croak ("callback must be undef or of reference type"); \ 949 croak ("callback must be undef or of reference type"); \
913 \ 950 \
914 Newz (0, req, 1, aio_cb); \ 951 Newz (0, req, 1, aio_cb); \
915 if (!req) \ 952 if (!req) \
916 croak ("out of memory during aio_req allocation"); \ 953 croak ("out of memory during aio_req allocation"); \
917 \ 954 \
918 req->callback = newSVsv (callback) 955 req->callback = newSVsv (callback); \
956 req->pri = req_pri
919 957
920#define REQ_SEND \ 958#define REQ_SEND \
921 req_send (req); \ 959 req_send (req); \
922 \ 960 \
923 if (GIMME_V != G_VOID) \ 961 if (GIMME_V != G_VOID) \
… …
1198 SV * callback 1236 SV * callback
1199 PROTOTYPE: ;$ 1237 PROTOTYPE: ;$
1200 PPCODE: 1238 PPCODE:
1201{ 1239{
1202 dREQ; 1240 dREQ;
1241
1203 req->type = REQ_GROUP; 1242 req->type = REQ_GROUP;
1204 req_send (req); 1243 req_send (req);
1244
1205 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1245 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1206} 1246}
1247
1248void
1249aio_nop (callback=&PL_sv_undef)
1250 SV * callback
1251 PPCODE:
1252{
1253 dREQ;
1254
1255 req->type = REQ_NOP;
1256
1257 REQ_SEND;
1258}
1259
1260#if 0
1261
1262void
1263aio_pri (int pri = DEFAULT_PRI)
1264 CODE:
1265 if (pri < PRI_MIN) pri = PRI_MIN;
1266 if (pri > PRI_MAX) pri = PRI_MAX;
1267 next_pri = pri + PRI_BIAS;
1268
1269#endif
1207 1270
1208void 1271void
1209flush () 1272flush ()
1210 PROTOTYPE: 1273 PROTOTYPE:
1211 CODE: 1274 CODE:
… …
1264cancel (aio_req_ornot req) 1327cancel (aio_req_ornot req)
1265 PROTOTYPE: 1328 PROTOTYPE:
1266 CODE: 1329 CODE:
1267 req_cancel (req); 1330 req_cancel (req);
1268 1331
1332void
1333cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1334 CODE:
1335 SvREFCNT_dec (req->callback);
1336 req->callback = newSVsv (callback);
1337
1269MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1338MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1270 1339
1271void 1340void
1272add (aio_req grp, ...) 1341add (aio_req grp, ...)
1273 PPCODE: 1342 PPCODE:
1274{ 1343{
1275 int i; 1344 int i;
1345 aio_req req;
1276 1346
1277 if (grp->fd == 2) 1347 if (grp->fd == 2)
1278 croak ("cannot add requests to IO::AIO::GRP after the group finished"); 1348 croak ("cannot add requests to IO::AIO::GRP after the group finished");
1279 1349
1280 for (i = 1; i < items; ++i ) 1350 for (i = 1; i < items; ++i )
1281 { 1351 {
1282 if (GIMME_V != G_VOID) 1352 if (GIMME_V != G_VOID)
1283 XPUSHs (sv_2mortal (newSVsv (ST (i)))); 1353 XPUSHs (sv_2mortal (newSVsv (ST (i))));
1284 1354
1285 aio_req req = SvAIO_REQ (ST (i)); 1355 req = SvAIO_REQ (ST (i));
1286 1356
1287 if (req) 1357 if (req)
1288 { 1358 {
1289 ++grp->length; 1359 ++grp->length;
1290 req->grp = grp; 1360 req->grp = grp;
… …
1300 } 1370 }
1301} 1371}
1302 1372
1303void 1373void
1304result (aio_req grp, ...) 1374result (aio_req grp, ...)
1305 CODE: 1375 CODE:
1306{ 1376{
1307 int i; 1377 int i;
1308 AV *av = newAV (); 1378 AV *av = newAV ();
1309 1379
1310 for (i = 1; i < items; ++i ) 1380 for (i = 1; i < items; ++i )
1311 av_push (av, newSVsv (ST (i))); 1381 av_push (av, newSVsv (ST (i)));
1312 1382
1313 SvREFCNT_dec (grp->data); 1383 SvREFCNT_dec (grp->data);
1314 grp->data = (SV *)av; 1384 grp->data = (SV *)av;
1315} 1385}
1316 1386
1317void 1387void
1318feeder_limit (aio_req grp, int limit) 1388feed_limit (aio_req grp, int limit)
1319 CODE: 1389 CODE:
1320 grp->fd2 = limit; 1390 grp->fd2 = limit;
1321 aio_grp_feed (grp); 1391 aio_grp_feed (grp);
1322 1392
1323void 1393void
1324set_feeder (aio_req grp, SV *callback=&PL_sv_undef) 1394feed (aio_req grp, SV *callback=&PL_sv_undef)
1325 CODE: 1395 CODE:
1326{ 1396{
1327 SvREFCNT_dec (grp->fh2); 1397 SvREFCNT_dec (grp->fh2);
1328 grp->fh2 = newSVsv (callback); 1398 grp->fh2 = newSVsv (callback);
1329 1399

Diff Legend

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