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.47 by root, Sun Oct 22 10:10:23 2006 UTC vs.
Revision 1.53 by root, Mon Oct 23 00:17:07 2006 UTC

… …
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 *grp, *grp_prev, *grp_next;
70
71 struct aio_cb *volatile next; 69 struct aio_cb *volatile next;
70
71 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
72 72
73 SV *self; /* the perl counterpart of this request, if any */ 73 SV *self; /* the perl counterpart of this request, if any */
74 74
75 SV *data, *callback; 75 SV *data, *callback;
76 SV *fh, *fh2; 76 SV *fh, *fh2;
… …
101static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 101static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
102 102
103static volatile aio_req reqs, reqe; /* queue start, queue end */ 103static volatile aio_req reqs, reqe; /* queue start, queue end */
104static volatile aio_req ress, rese; /* queue start, queue end */ 104static volatile aio_req ress, rese; /* queue start, queue end */
105 105
106static void req_invoke (aio_req req);
106static void req_free (aio_req req); 107static void req_free (aio_req req);
107 108
108/* must be called at most once */ 109/* must be called at most once */
109static SV *req_sv (aio_req req, const char *klass) 110static SV *req_sv (aio_req req, const char *klass)
110{ 111{
112 if (!req->self)
113 {
111 req->self = (SV *)newHV (); 114 req->self = (SV *)newHV ();
112 sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0); 115 sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);
116 }
113 117
114 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1))); 118 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1)));
115} 119}
116 120
117static aio_req SvAIO_REQ (SV *sv) 121static aio_req SvAIO_REQ (SV *sv)
118{ 122{
123 MAGIC *mg;
124
119 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv)) 125 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv))
120 croak ("object of class " AIO_REQ_KLASS " expected"); 126 croak ("object of class " AIO_REQ_KLASS " expected");
121 127
122 MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext); 128 mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
123 129
124 return mg ? (aio_req)mg->mg_ptr : 0; 130 return mg ? (aio_req)mg->mg_ptr : 0;
131}
132
133static void aio_grp_feed (aio_req grp)
134{
135 while (grp->length < grp->fd2)
136 {
137 int old_len = grp->length;
138
139 if (grp->fh2 && SvOK (grp->fh2))
140 {
141 dSP;
142
143 ENTER;
144 SAVETMPS;
145 PUSHMARK (SP);
146 XPUSHs (req_sv (grp, AIO_GRP_KLASS));
147 PUTBACK;
148 call_sv (grp->fh2, G_VOID | G_EVAL);
149 SPAGAIN;
150 FREETMPS;
151 LEAVE;
152 }
153
154 /* stop if no progress has been made */
155 if (old_len == grp->length)
156 {
157 SvREFCNT_dec (grp->fh2);
158 grp->fh2 = 0;
159 break;
160 }
161 }
162}
163
164static void aio_grp_dec (aio_req grp)
165{
166 --grp->length;
167
168 /* call feeder, if applicable */
169 aio_grp_feed (grp);
170
171 /* finish, if done */
172 if (!grp->length && grp->fd)
173 {
174 req_invoke (grp);
175 req_free (grp);
176 }
125} 177}
126 178
127static void poll_wait () 179static void poll_wait ()
128{ 180{
129 if (nreqs && !ress) 181 if (nreqs && !ress)
… …
145 return; 197 return;
146 198
147 errno = req->errorno; 199 errno = req->errorno;
148 200
149 ENTER; 201 ENTER;
202 SAVETMPS;
150 PUSHMARK (SP); 203 PUSHMARK (SP);
204 EXTEND (SP, 1);
151 205
152 switch (req->type) 206 switch (req->type)
153 { 207 {
154 case REQ_READDIR: 208 case REQ_READDIR:
155 { 209 {
… …
170 } 224 }
171 225
172 rv = sv_2mortal (newRV_noinc ((SV *)av)); 226 rv = sv_2mortal (newRV_noinc ((SV *)av));
173 } 227 }
174 228
175 XPUSHs (rv); 229 PUSHs (rv);
176 } 230 }
177 break; 231 break;
178 232
179 case REQ_OPEN: 233 case REQ_OPEN:
180 { 234 {
181 /* convert fd to fh */ 235 /* convert fd to fh */
182 SV *fh; 236 SV *fh;
183 237
184 XPUSHs (sv_2mortal (newSViv (req->result))); 238 PUSHs (sv_2mortal (newSViv (req->result)));
185 PUTBACK; 239 PUTBACK;
186 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL); 240 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
187 SPAGAIN; 241 SPAGAIN;
188 242
189 fh = SvREFCNT_inc (POPs); 243 fh = SvREFCNT_inc (POPs);
… …
191 PUSHMARK (SP); 245 PUSHMARK (SP);
192 XPUSHs (sv_2mortal (fh)); 246 XPUSHs (sv_2mortal (fh));
193 } 247 }
194 break; 248 break;
195 249
250 case REQ_GROUP:
251 req->fd = 2; /* mark group as finished */
252
253 if (req->data)
254 {
255 int i;
256 AV *av = (AV *)req->data;
257
258 EXTEND (SP, AvFILL (av) + 1);
259 for (i = 0; i <= AvFILL (av); ++i)
260 PUSHs (*av_fetch (av, i, 0));
261 }
262 break;
263
196 case REQ_SLEEP: 264 case REQ_SLEEP:
197 case REQ_GROUP:
198 break; 265 break;
199 266
200 default: 267 default:
201 XPUSHs (sv_2mortal (newSViv (req->result))); 268 PUSHs (sv_2mortal (newSViv (req->result)));
202 break; 269 break;
203 } 270 }
204 271
205 272
206 PUTBACK; 273 PUTBACK;
207 call_sv (req->callback, G_VOID | G_EVAL); 274 call_sv (req->callback, G_VOID | G_EVAL);
208 SPAGAIN; 275 SPAGAIN;
209 276
277 FREETMPS;
278 LEAVE;
279
280 errno = errorno;
281
210 if (SvTRUE (ERRSV)) 282 if (SvTRUE (ERRSV))
211 { 283 {
212 req_free (req); 284 req_free (req);
213 croak (0); 285 croak (0);
214 } 286 }
215
216 LEAVE;
217
218 errno = errorno;
219} 287}
220 288
221static void req_free (aio_req req) 289static void req_free (aio_req req)
222{ 290{
223 if (req->grp) 291 if (req->grp)
224 { 292 {
225 aio_req grp = req->grp; 293 aio_req grp = req->grp;
226 294
227 /* unlink request */ 295 /* unlink request */
228 req->grp_next->grp_prev = req->grp_prev; 296 if (req->grp_next) req->grp_next->grp_prev = req->grp_prev;
229 req->grp_prev->grp_next = req->grp_next; 297 if (req->grp_prev) req->grp_prev->grp_next = req->grp_next;
230 298
231 if (grp->grp_next == grp && grp->fd) 299 if (grp->grp_first == req)
232 { 300 grp->grp_first = req->grp_next;
233 req_invoke (grp); 301
234 req_free (grp); 302 aio_grp_dec (grp);
235 }
236 } 303 }
237 304
238 if (req->self) 305 if (req->self)
239 { 306 {
240 sv_unmagic (req->self, PERL_MAGIC_ext); 307 sv_unmagic (req->self, PERL_MAGIC_ext);
241 SvREFCNT_dec (req->self); 308 SvREFCNT_dec (req->self);
242 } 309 }
243 310
244 if (req->data)
245 SvREFCNT_dec (req->data); 311 SvREFCNT_dec (req->data);
246
247 if (req->fh)
248 SvREFCNT_dec (req->fh); 312 SvREFCNT_dec (req->fh);
249
250 if (req->fh2)
251 SvREFCNT_dec (req->fh2); 313 SvREFCNT_dec (req->fh2);
252
253 if (req->statdata)
254 Safefree (req->statdata);
255
256 if (req->callback)
257 SvREFCNT_dec (req->callback); 314 SvREFCNT_dec (req->callback);
315 Safefree (req->statdata);
258 316
259 if (req->type == REQ_READDIR && req->result >= 0) 317 if (req->type == REQ_READDIR && req->result >= 0)
260 free (req->data2ptr); 318 free (req->data2ptr);
261 319
262 Safefree (req); 320 Safefree (req);
… …
268 326
269 if (req->type == REQ_GROUP) 327 if (req->type == REQ_GROUP)
270 { 328 {
271 aio_req sub; 329 aio_req sub;
272 330
273 for (sub = req->grp_next; sub != req; sub = sub->grp_next) 331 for (sub = req->grp_first; sub; sub = sub->grp_next)
274 req_cancel (sub); 332 req_cancel (sub);
275 } 333 }
276} 334}
277 335
278static int poll_cb () 336static int poll_cb ()
… …
305 pthread_mutex_unlock (&reslock); 363 pthread_mutex_unlock (&reslock);
306 364
307 if (!req) 365 if (!req)
308 break; 366 break;
309 367
310 nreqs--; 368 --nreqs;
311 369
312 if (req->type == REQ_QUIT) 370 if (req->type == REQ_QUIT)
313 started--; 371 started--;
314 else if (req->type == REQ_GROUP && req->grp_next != req) 372 else if (req->type == REQ_GROUP && req->length)
315 { 373 {
316 req->fd = 1; /* mark request as delayed */ 374 req->fd = 1; /* mark request as delayed */
317 continue; 375 continue;
318 } 376 }
319 else 377 else
… …
366static void req_send (aio_req req) 424static void req_send (aio_req req)
367{ 425{
368 while (started < wanted && nreqs >= started) 426 while (started < wanted && nreqs >= started)
369 start_thread (); 427 start_thread ();
370 428
371 nreqs++; 429 ++nreqs;
372 430
373 pthread_mutex_lock (&reqlock); 431 pthread_mutex_lock (&reqlock);
374 432
375 req->next = 0; 433 req->next = 0;
376 434
… …
728 pthread_mutex_unlock (&reqlock); 786 pthread_mutex_unlock (&reqlock);
729 787
730 errno = 0; /* strictly unnecessary */ 788 errno = 0; /* strictly unnecessary */
731 789
732 if (!req->cancelled) 790 if (!req->cancelled)
733 switch (req->type) 791 switch (type = req->type) /* remember type for QUIT check */
734 { 792 {
735 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 793 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
736 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 794 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
737 795
738 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 796 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
… …
1149 PROTOTYPE: ;$ 1207 PROTOTYPE: ;$
1150 PPCODE: 1208 PPCODE:
1151{ 1209{
1152 dREQ; 1210 dREQ;
1153 req->type = REQ_GROUP; 1211 req->type = REQ_GROUP;
1154 req->grp_next = req;
1155 req->grp_prev = req;
1156
1157 req_send (req); 1212 req_send (req);
1158 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1213 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1159} 1214}
1160 1215
1161void 1216void
… …
1207 CODE: 1262 CODE:
1208 RETVAL = nreqs; 1263 RETVAL = nreqs;
1209 OUTPUT: 1264 OUTPUT:
1210 RETVAL 1265 RETVAL
1211 1266
1267PROTOTYPES: DISABLE
1268
1212MODULE = IO::AIO PACKAGE = IO::AIO::REQ 1269MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1213 1270
1214void 1271void
1215cancel (aio_req_ornot req) 1272cancel (aio_req_ornot req)
1216 PROTOTYPE: 1273 PROTOTYPE:
… …
1219 1276
1220MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1277MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1221 1278
1222void 1279void
1223add (aio_req grp, ...) 1280add (aio_req grp, ...)
1224 PROTOTYPE: $;@
1225 PPCODE: 1281 PPCODE:
1226{ 1282{
1227 int i; 1283 int i;
1284 aio_req req;
1285
1286 if (grp->fd == 2)
1287 croak ("cannot add requests to IO::AIO::GRP after the group finished");
1228 1288
1229 for (i = 1; i < items; ++i ) 1289 for (i = 1; i < items; ++i )
1230 { 1290 {
1231 if (GIMME_V != G_VOID) 1291 if (GIMME_V != G_VOID)
1232 XPUSHs (sv_2mortal (newSVsv (ST (i)))); 1292 XPUSHs (sv_2mortal (newSVsv (ST (i))));
1233 1293
1234 aio_req req = SvAIO_REQ (ST (i)); 1294 req = SvAIO_REQ (ST (i));
1235 1295
1236 if (req) 1296 if (req)
1237 { 1297 {
1238 req->grp_prev = grp; 1298 ++grp->length;
1239 req->grp_next = grp->grp_next;
1240 grp->grp_next->grp_prev = req;
1241 grp->grp_next = req;
1242
1243 req->grp = grp; 1299 req->grp = grp;
1300
1301 req->grp_prev = 0;
1302 req->grp_next = grp->grp_first;
1303
1304 if (grp->grp_first)
1305 grp->grp_first->grp_prev = req;
1306
1307 grp->grp_first = req;
1244 } 1308 }
1245 } 1309 }
1246} 1310}
1247 1311
1312void
1313result (aio_req grp, ...)
1314 CODE:
1315{
1316 int i;
1317 AV *av = newAV ();
1318
1319 for (i = 1; i < items; ++i )
1320 av_push (av, newSVsv (ST (i)));
1321
1322 SvREFCNT_dec (grp->data);
1323 grp->data = (SV *)av;
1324}
1325
1326void
1327lock (aio_req grp)
1328 CODE:
1329 ++grp->length;
1330
1331void
1332unlock (aio_req grp)
1333 CODE:
1334 aio_grp_dec (grp);
1335
1336void
1337feeder_limit (aio_req grp, int limit)
1338 CODE:
1339 grp->fd2 = limit;
1340 aio_grp_feed (grp);
1341
1342void
1343set_feeder (aio_req grp, SV *callback=&PL_sv_undef)
1344 CODE:
1345{
1346 SvREFCNT_dec (grp->fh2);
1347 grp->fh2 = newSVsv (callback);
1348
1349 if (grp->fd2 <= 0)
1350 grp->fd2 = 2;
1351
1352 aio_grp_feed (grp);
1353}
1354

Diff Legend

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