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.45 by root, Sun Oct 22 00:19:05 2006 UTC vs.
Revision 1.64 by root, Mon Oct 23 23:54:41 2006 UTC

1#if __linux
2# define _GNU_SOURCE
3#endif
4
1#define _REENTRANT 1 5#define _REENTRANT 1
6
2#include <errno.h> 7#include <errno.h>
3 8
4#include "EXTERN.h" 9#include "EXTERN.h"
5#include "perl.h" 10#include "perl.h"
6#include "XSUB.h" 11#include "XSUB.h"
55 REQ_STAT, REQ_LSTAT, REQ_FSTAT, 60 REQ_STAT, REQ_LSTAT, REQ_FSTAT,
56 REQ_FSYNC, REQ_FDATASYNC, 61 REQ_FSYNC, REQ_FDATASYNC,
57 REQ_UNLINK, REQ_RMDIR, REQ_RENAME, 62 REQ_UNLINK, REQ_RMDIR, REQ_RENAME,
58 REQ_READDIR, 63 REQ_READDIR,
59 REQ_LINK, REQ_SYMLINK, 64 REQ_LINK, REQ_SYMLINK,
65 REQ_GROUP, REQ_NOP,
60 REQ_SLEEP, 66 REQ_SLEEP,
61 REQ_GROUP,
62}; 67};
63 68
64#define AIO_REQ_KLASS "IO::AIO::REQ" 69#define AIO_REQ_KLASS "IO::AIO::REQ"
65#define AIO_GRP_KLASS "IO::AIO::GRP" 70#define AIO_GRP_KLASS "IO::AIO::GRP"
66 71
67typedef struct aio_cb 72typedef struct aio_cb
68{ 73{
69 struct aio_cb *grp, *grp_prev, *grp_next;
70
71 struct aio_cb *volatile next; 74 struct aio_cb *volatile next;
72
73 SV *self; /* the perl counterpart of this request, if any */
74 75
75 SV *data, *callback; 76 SV *data, *callback;
76 SV *fh, *fh2; 77 SV *fh, *fh2;
77 void *dataptr, *data2ptr; 78 void *dataptr, *data2ptr;
78 Stat_t *statdata; 79 Stat_t *statdata;
79 off_t offset; 80 off_t offset;
80 size_t length; 81 size_t length;
81 ssize_t result; 82 ssize_t result;
82 83
84 STRLEN dataoffset;
83 int type; 85 int type;
84 int fd, fd2; 86 int fd, fd2;
85 int errorno; 87 int errorno;
86 STRLEN dataoffset;
87 mode_t mode; /* open */ 88 mode_t mode; /* open */
89
88 unsigned char cancelled; 90 unsigned char flags;
91 unsigned char pri;
92
93 SV *self; /* the perl counterpart of this request, if any */
94 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
89} aio_cb; 95} aio_cb;
96
97enum {
98 FLAG_CANCELLED = 0x01,
99};
90 100
91typedef aio_cb *aio_req; 101typedef aio_cb *aio_req;
92typedef aio_cb *aio_req_ornot; 102typedef aio_cb *aio_req_ornot;
93typedef aio_cb *aio_group; 103
104enum {
105 PRI_MIN = -4,
106 PRI_MAX = 4,
107
108 DEFAULT_PRI = 0,
109 PRI_BIAS = -PRI_MIN,
110};
111
112static int next_pri = DEFAULT_PRI + PRI_BIAS;
94 113
95static int started, wanted; 114static int started, wanted;
96static volatile int nreqs; 115static volatile int nreqs;
97static int max_outstanding = 1<<30; 116static int max_outstanding = 1<<30;
98static int respipe [2]; 117static int respipe [2];
99 118
119#if __linux && defined (PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP)
120# define AIO_MUTEX_INIT PTHREAD_ADAPTIVE_MUTEX_INITIALIZER_NP
121#else
122# define AIO_MUTEX_INIT PTHREAD_MUTEX_INITIALIZER
123#endif
124
100static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER; 125static pthread_mutex_t reslock = AIO_MUTEX_INIT;
101static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER; 126static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
102static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 127static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
103 128
104static volatile aio_req reqs, reqe; /* queue start, queue end */ 129static volatile aio_req reqs, reqe; /* queue start, queue end */
105static volatile aio_req ress, rese; /* queue start, queue end */ 130static volatile aio_req ress, rese; /* queue start, queue end */
106 131
132static void req_invoke (aio_req req);
107static void req_free (aio_req req); 133static void req_free (aio_req req);
108 134
109/* must be called at most once */ 135/* must be called at most once */
110static SV *req_sv (aio_req req, const char *klass) 136static SV *req_sv (aio_req req, const char *klass)
111{ 137{
138 if (!req->self)
139 {
112 req->self = (SV *)newHV (); 140 req->self = (SV *)newHV ();
113 sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0); 141 sv_magic (req->self, 0, PERL_MAGIC_ext, (char *)req, 0);
142 }
114 143
115 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1))); 144 return sv_2mortal (sv_bless (newRV_inc (req->self), gv_stashpv (klass, 1)));
116} 145}
117 146
118static aio_req SvAIO_REQ (SV *sv) 147static aio_req SvAIO_REQ (SV *sv)
119{ 148{
149 MAGIC *mg;
150
120 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv)) 151 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv))
121 croak ("object of class " AIO_REQ_KLASS " expected"); 152 croak ("object of class " AIO_REQ_KLASS " expected");
122 153
123 MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext); 154 mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
124 155
125 return mg ? (aio_req)mg->mg_ptr : 0; 156 return mg ? (aio_req)mg->mg_ptr : 0;
126} 157}
127 158
159static void aio_grp_feed (aio_req grp)
160{
161 while (grp->length < grp->fd2 && !(grp->flags & FLAG_CANCELLED))
162 {
163 int old_len = grp->length;
164
165 if (grp->fh2 && SvOK (grp->fh2))
166 {
167 dSP;
168
169 ENTER;
170 SAVETMPS;
171 PUSHMARK (SP);
172 XPUSHs (req_sv (grp, AIO_GRP_KLASS));
173 PUTBACK;
174 call_sv (grp->fh2, G_VOID | G_EVAL);
175 SPAGAIN;
176 FREETMPS;
177 LEAVE;
178 }
179
180 /* stop if no progress has been made */
181 if (old_len == grp->length)
182 {
183 SvREFCNT_dec (grp->fh2);
184 grp->fh2 = 0;
185 break;
186 }
187 }
188}
189
190static void aio_grp_dec (aio_req grp)
191{
192 --grp->length;
193
194 /* call feeder, if applicable */
195 aio_grp_feed (grp);
196
197 /* finish, if done */
198 if (!grp->length && grp->fd)
199 {
200 req_invoke (grp);
201 req_free (grp);
202 }
203}
204
128static void poll_wait () 205static void poll_wait ()
129{ 206{
130 if (nreqs && !ress)
131 {
132 fd_set rfd; 207 fd_set rfd;
208
209 while (nreqs)
210 {
211 aio_req req;
212#if !(__x86 || __x86_64) /* safe without sempahore on this archs */
213 pthread_mutex_lock (&reslock);
214#endif
215 req = ress;
216#if !(__x86 || __x86_64) /* safe without sempahore on this archs */
217 pthread_mutex_unlock (&reslock);
218#endif
219
220 if (req)
221 return;
222
133 FD_ZERO(&rfd); 223 FD_ZERO(&rfd);
134 FD_SET(respipe [0], &rfd); 224 FD_SET(respipe [0], &rfd);
135 225
136 select (respipe [0] + 1, &rfd, 0, 0, 0); 226 select (respipe [0] + 1, &rfd, 0, 0, 0);
137 } 227 }
140static void req_invoke (aio_req req) 230static void req_invoke (aio_req req)
141{ 231{
142 dSP; 232 dSP;
143 int errorno = errno; 233 int errorno = errno;
144 234
145 if (req->cancelled || !SvOK (req->callback)) 235 if (req->flags & FLAG_CANCELLED || !SvOK (req->callback))
146 return; 236 return;
147 237
148 errno = req->errorno; 238 errno = req->errorno;
149 239
150 ENTER; 240 ENTER;
241 SAVETMPS;
151 PUSHMARK (SP); 242 PUSHMARK (SP);
243 EXTEND (SP, 1);
152 244
153 switch (req->type) 245 switch (req->type)
154 { 246 {
155 case REQ_READDIR: 247 case REQ_READDIR:
156 { 248 {
171 } 263 }
172 264
173 rv = sv_2mortal (newRV_noinc ((SV *)av)); 265 rv = sv_2mortal (newRV_noinc ((SV *)av));
174 } 266 }
175 267
176 XPUSHs (rv); 268 PUSHs (rv);
177 } 269 }
178 break; 270 break;
179 271
180 case REQ_OPEN: 272 case REQ_OPEN:
181 { 273 {
182 /* convert fd to fh */ 274 /* convert fd to fh */
183 SV *fh; 275 SV *fh;
184 276
185 XPUSHs (sv_2mortal (newSViv (req->result))); 277 PUSHs (sv_2mortal (newSViv (req->result)));
186 PUTBACK; 278 PUTBACK;
187 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL); 279 call_pv ("IO::AIO::_fd2fh", G_SCALAR | G_EVAL);
188 SPAGAIN; 280 SPAGAIN;
189 281
190 fh = SvREFCNT_inc (POPs); 282 fh = SvREFCNT_inc (POPs);
192 PUSHMARK (SP); 284 PUSHMARK (SP);
193 XPUSHs (sv_2mortal (fh)); 285 XPUSHs (sv_2mortal (fh));
194 } 286 }
195 break; 287 break;
196 288
289 case REQ_GROUP:
290 req->fd = 2; /* mark group as finished */
291
292 if (req->data)
293 {
294 int i;
295 AV *av = (AV *)req->data;
296
297 EXTEND (SP, AvFILL (av) + 1);
298 for (i = 0; i <= AvFILL (av); ++i)
299 PUSHs (*av_fetch (av, i, 0));
300 }
301 break;
302
303 case REQ_NOP:
197 case REQ_SLEEP: 304 case REQ_SLEEP:
198 case REQ_GROUP:
199 break; 305 break;
200 306
201 default: 307 default:
202 XPUSHs (sv_2mortal (newSViv (req->result))); 308 PUSHs (sv_2mortal (newSViv (req->result)));
203 break; 309 break;
204 } 310 }
205 311
206 312
207 PUTBACK; 313 PUTBACK;
208 call_sv (req->callback, G_VOID | G_EVAL); 314 call_sv (req->callback, G_VOID | G_EVAL);
209 SPAGAIN; 315 SPAGAIN;
210 316
317 FREETMPS;
318 LEAVE;
319
320 errno = errorno;
321
211 if (SvTRUE (ERRSV)) 322 if (SvTRUE (ERRSV))
212 { 323 {
213 req_free (req); 324 req_free (req);
214 croak (0); 325 croak (0);
215 } 326 }
216
217 LEAVE;
218
219 errno = errorno;
220} 327}
221 328
222static void req_free (aio_req req) 329static void req_free (aio_req req)
223{ 330{
224 if (req->grp) 331 if (req->grp)
225 { 332 {
226 aio_req grp = req->grp; 333 aio_req grp = req->grp;
227 334
228 /* unlink request */ 335 /* unlink request */
229 req->grp_next->grp_prev = req->grp_prev; 336 if (req->grp_next) req->grp_next->grp_prev = req->grp_prev;
230 req->grp_prev->grp_next = req->grp_next; 337 if (req->grp_prev) req->grp_prev->grp_next = req->grp_next;
231 338
232 if (grp->grp_next == grp) 339 if (grp->grp_first == req)
233 { 340 grp->grp_first = req->grp_next;
234 req_invoke (grp); 341
235 req_free (grp); 342 aio_grp_dec (grp);
236 }
237 } 343 }
238 344
239 if (req->self) 345 if (req->self)
240 { 346 {
241 sv_unmagic (req->self, PERL_MAGIC_ext); 347 sv_unmagic (req->self, PERL_MAGIC_ext);
242 SvREFCNT_dec (req->self); 348 SvREFCNT_dec (req->self);
243 } 349 }
244 350
245 if (req->data)
246 SvREFCNT_dec (req->data); 351 SvREFCNT_dec (req->data);
247
248 if (req->fh)
249 SvREFCNT_dec (req->fh); 352 SvREFCNT_dec (req->fh);
250
251 if (req->fh2)
252 SvREFCNT_dec (req->fh2); 353 SvREFCNT_dec (req->fh2);
253
254 if (req->statdata)
255 Safefree (req->statdata);
256
257 if (req->callback)
258 SvREFCNT_dec (req->callback); 354 SvREFCNT_dec (req->callback);
355 Safefree (req->statdata);
259 356
260 if (req->type == REQ_READDIR && req->result >= 0) 357 if (req->type == REQ_READDIR && req->result >= 0)
261 free (req->data2ptr); 358 free (req->data2ptr);
262 359
263 Safefree (req); 360 Safefree (req);
264} 361}
265 362
266static void req_cancel (aio_req req) 363static void req_cancel (aio_req req)
267{ 364{
268 req->cancelled = 1; 365 req->flags |= FLAG_CANCELLED;
269 366
270 if (req->type == REQ_GROUP) 367 if (req->type == REQ_GROUP)
271 { 368 {
272 aio_req sub; 369 aio_req sub;
273 370
274 for (sub = req->grp_next; sub != req; sub = sub->grp_next) 371 for (sub = req->grp_first; sub; sub = sub->grp_next)
275 req_cancel (sub); 372 req_cancel (sub);
276 } 373 }
277} 374}
278 375
279static int poll_cb () 376static int poll_cb ()
306 pthread_mutex_unlock (&reslock); 403 pthread_mutex_unlock (&reslock);
307 404
308 if (!req) 405 if (!req)
309 break; 406 break;
310 407
311 nreqs--; 408 --nreqs;
312 409
313 if (req->type == REQ_QUIT) 410 if (req->type == REQ_QUIT)
314 started--; 411 started--;
315 else if (req->type == REQ_GROUP && req->grp_next != req) 412 else if (req->type == REQ_GROUP && req->length)
413 {
414 req->fd = 1; /* mark request as delayed */
316 continue; 415 continue;
416 }
317 else 417 else
318 { 418 {
319 if (req->type == REQ_READ) 419 if (req->type == REQ_READ)
320 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0)); 420 SvCUR_set (req->data, req->dataoffset + (req->result > 0 ? req->result : 0));
321 421
364static void req_send (aio_req req) 464static void req_send (aio_req req)
365{ 465{
366 while (started < wanted && nreqs >= started) 466 while (started < wanted && nreqs >= started)
367 start_thread (); 467 start_thread ();
368 468
369 nreqs++; 469 ++nreqs;
370 470
371 pthread_mutex_lock (&reqlock); 471 pthread_mutex_lock (&reqlock);
372 472
373 req->next = 0; 473 req->next = 0;
374 474
724 } 824 }
725 825
726 pthread_mutex_unlock (&reqlock); 826 pthread_mutex_unlock (&reqlock);
727 827
728 errno = 0; /* strictly unnecessary */ 828 errno = 0; /* strictly unnecessary */
829 type = req->type; /* remember type for QUIT check */
729 830
730 if (!req->cancelled) 831 if (!(req->flags & FLAG_CANCELLED))
731 switch (req->type) 832 switch (type)
732 { 833 {
733 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 834 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
734 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 835 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
735 836
736 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 837 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
760 tv.tv_usec = req->fd2; 861 tv.tv_usec = req->fd2;
761 862
762 req->result = select (0, 0, 0, 0, &tv); 863 req->result = select (0, 0, 0, 0, &tv);
763 } 864 }
764 865
866 case REQ_GROUP:
867 case REQ_NOP:
765 case REQ_QUIT: 868 case REQ_QUIT:
766 break; 869 break;
767 870
768 default: 871 default:
769 req->result = ENOSYS; 872 req->result = ENOSYS;
853 atfork_parent (); 956 atfork_parent ();
854} 957}
855 958
856#define dREQ \ 959#define dREQ \
857 aio_req req; \ 960 aio_req req; \
961 int req_pri = next_pri; \
962 next_pri = DEFAULT_PRI + PRI_BIAS; \
858 \ 963 \
859 if (SvOK (callback) && !SvROK (callback)) \ 964 if (SvOK (callback) && !SvROK (callback)) \
860 croak ("callback must be undef or of reference type"); \ 965 croak ("callback must be undef or of reference type"); \
861 \ 966 \
862 Newz (0, req, 1, aio_cb); \ 967 Newz (0, req, 1, aio_cb); \
863 if (!req) \ 968 if (!req) \
864 croak ("out of memory during aio_req allocation"); \ 969 croak ("out of memory during aio_req allocation"); \
865 \ 970 \
866 req->callback = newSVsv (callback) 971 req->callback = newSVsv (callback); \
972 req->pri = req_pri
867 973
868#define REQ_SEND \ 974#define REQ_SEND \
869 req_send (req); \ 975 req_send (req); \
870 \ 976 \
871 if (GIMME_V != G_VOID) \ 977 if (GIMME_V != G_VOID) \
1142} 1248}
1143 1249
1144void 1250void
1145aio_group (callback=&PL_sv_undef) 1251aio_group (callback=&PL_sv_undef)
1146 SV * callback 1252 SV * callback
1147 PROTOTYPE: ;& 1253 PROTOTYPE: ;$
1148 PPCODE: 1254 PPCODE:
1149{ 1255{
1150 dREQ; 1256 dREQ;
1257
1151 req->type = REQ_GROUP; 1258 req->type = REQ_GROUP;
1152 req->grp_next = req;
1153 req->grp_prev = req;
1154
1155 req_send (req); 1259 req_send (req);
1260
1156 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1261 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1157} 1262}
1263
1264void
1265aio_nop (callback=&PL_sv_undef)
1266 SV * callback
1267 PPCODE:
1268{
1269 dREQ;
1270
1271 req->type = REQ_NOP;
1272
1273 REQ_SEND;
1274}
1275
1276#if 0
1277
1278void
1279aio_pri (int pri = DEFAULT_PRI)
1280 CODE:
1281 if (pri < PRI_MIN) pri = PRI_MIN;
1282 if (pri > PRI_MAX) pri = PRI_MAX;
1283 next_pri = pri + PRI_BIAS;
1284
1285#endif
1158 1286
1159void 1287void
1160flush () 1288flush ()
1161 PROTOTYPE: 1289 PROTOTYPE:
1162 CODE: 1290 CODE:
1205 CODE: 1333 CODE:
1206 RETVAL = nreqs; 1334 RETVAL = nreqs;
1207 OUTPUT: 1335 OUTPUT:
1208 RETVAL 1336 RETVAL
1209 1337
1338PROTOTYPES: DISABLE
1339
1210MODULE = IO::AIO PACKAGE = IO::AIO::REQ 1340MODULE = IO::AIO PACKAGE = IO::AIO::REQ
1211 1341
1212void 1342void
1213cancel (aio_req_ornot req) 1343cancel (aio_req_ornot req)
1214 PROTOTYPE: 1344 PROTOTYPE:
1215 CODE: 1345 CODE:
1216 req_cancel (req); 1346 req_cancel (req);
1217 1347
1348void
1349cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1350 CODE:
1351 SvREFCNT_dec (req->callback);
1352 req->callback = newSVsv (callback);
1353
1218MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1354MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1219 1355
1220void 1356void
1221add (aio_req grp, ...) 1357add (aio_req grp, ...)
1222 PROTOTYPE: $;@
1223 PPCODE: 1358 PPCODE:
1224{ 1359{
1225 int i; 1360 int i;
1361 aio_req req;
1362
1363 if (grp->fd == 2)
1364 croak ("cannot add requests to IO::AIO::GRP after the group finished");
1226 1365
1227 for (i = 1; i < items; ++i ) 1366 for (i = 1; i < items; ++i )
1228 { 1367 {
1229 aio_req req = SvAIO_REQ (ST (i));
1230
1231 req->grp_prev = grp;
1232 req->grp_next = grp->grp_next;
1233 grp->grp_next->grp_prev = req;
1234 grp->grp_next = req;
1235
1236 req->grp = grp;
1237
1238 if (GIMME_V != G_VOID) 1368 if (GIMME_V != G_VOID)
1239 XPUSHs (sv_2mortal (newSVsv (ST (i)))); 1369 XPUSHs (sv_2mortal (newSVsv (ST (i))));
1370
1371 req = SvAIO_REQ (ST (i));
1372
1373 if (req)
1374 {
1375 ++grp->length;
1376 req->grp = grp;
1377
1378 req->grp_prev = 0;
1379 req->grp_next = grp->grp_first;
1380
1381 if (grp->grp_first)
1382 grp->grp_first->grp_prev = req;
1383
1384 grp->grp_first = req;
1385 }
1240 } 1386 }
1241} 1387}
1242 1388
1389void
1390result (aio_req grp, ...)
1391 CODE:
1392{
1393 int i;
1394 AV *av = newAV ();
1395
1396 for (i = 1; i < items; ++i )
1397 av_push (av, newSVsv (ST (i)));
1398
1399 SvREFCNT_dec (grp->data);
1400 grp->data = (SV *)av;
1401}
1402
1403void
1404feed_limit (aio_req grp, int limit)
1405 CODE:
1406 grp->fd2 = limit;
1407 aio_grp_feed (grp);
1408
1409void
1410feed (aio_req grp, SV *callback=&PL_sv_undef)
1411 CODE:
1412{
1413 SvREFCNT_dec (grp->fh2);
1414 grp->fh2 = newSVsv (callback);
1415
1416 if (grp->fd2 <= 0)
1417 grp->fd2 = 2;
1418
1419 aio_grp_feed (grp);
1420}
1421

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines