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.51 by root, Sun Oct 22 22:14:33 2006 UTC vs.
Revision 1.63 by root, Mon Oct 23 23:48:31 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 *volatile next; 74 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 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;
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;
93 113
94static int started, wanted; 114static int started, wanted;
95static volatile int nreqs; 115static volatile int nreqs;
96static int max_outstanding = 1<<30; 116static int max_outstanding = 1<<30;
97static int respipe [2]; 117static int respipe [2];
98 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
99static pthread_mutex_t reslock = PTHREAD_MUTEX_INITIALIZER; 125static pthread_mutex_t reslock = AIO_MUTEX_INIT;
100static pthread_mutex_t reqlock = PTHREAD_MUTEX_INITIALIZER; 126static pthread_mutex_t reqlock = AIO_MUTEX_INIT;
101static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER; 127static pthread_cond_t reqwait = PTHREAD_COND_INITIALIZER;
102 128
103static volatile aio_req reqs, reqe; /* queue start, queue end */ 129static volatile aio_req reqs, reqe; /* queue start, queue end */
104static volatile aio_req ress, rese; /* queue start, queue end */ 130static volatile aio_req ress, rese; /* queue start, queue end */
105 131
… …
118 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)));
119} 145}
120 146
121static aio_req SvAIO_REQ (SV *sv) 147static aio_req SvAIO_REQ (SV *sv)
122{ 148{
149 MAGIC *mg;
150
123 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv)) 151 if (!sv_derived_from (sv, AIO_REQ_KLASS) || !SvROK (sv))
124 croak ("object of class " AIO_REQ_KLASS " expected"); 152 croak ("object of class " AIO_REQ_KLASS " expected");
125 153
126 MAGIC *mg = mg_find (SvRV (sv), PERL_MAGIC_ext); 154 mg = mg_find (SvRV (sv), PERL_MAGIC_ext);
127 155
128 return mg ? (aio_req)mg->mg_ptr : 0; 156 return mg ? (aio_req)mg->mg_ptr : 0;
129} 157}
130 158
131static void aio_grp_feed (aio_req grp) 159static void aio_grp_feed (aio_req grp)
132{ 160{
133 while (grp->length < grp->fd2) 161 while (grp->length < grp->fd2 && !(grp->flags & FLAG_CANCELLED))
134 { 162 {
135 int old_len = grp->length; 163 int old_len = grp->length;
136 164
137 if (grp->fh2 && SvOK (grp->fh2)) 165 if (grp->fh2 && SvOK (grp->fh2))
138 { 166 {
… …
174 } 202 }
175} 203}
176 204
177static void poll_wait () 205static void poll_wait ()
178{ 206{
179 if (nreqs && !ress)
180 {
181 fd_set rfd; 207 fd_set rfd;
208
209 while (nreqs)
210 {
211 aio_req req;
212 pthread_mutex_lock (&reslock);
213 req = ress;
214 pthread_mutex_unlock (&reslock);
215
216 if (req)
217 return;
218
182 FD_ZERO(&rfd); 219 FD_ZERO(&rfd);
183 FD_SET(respipe [0], &rfd); 220 FD_SET(respipe [0], &rfd);
184 221
185 select (respipe [0] + 1, &rfd, 0, 0, 0); 222 select (respipe [0] + 1, &rfd, 0, 0, 0);
186 } 223 }
… …
189static void req_invoke (aio_req req) 226static void req_invoke (aio_req req)
190{ 227{
191 dSP; 228 dSP;
192 int errorno = errno; 229 int errorno = errno;
193 230
194 if (req->cancelled || !SvOK (req->callback)) 231 if (req->flags & FLAG_CANCELLED || !SvOK (req->callback))
195 return; 232 return;
196 233
197 errno = req->errorno; 234 errno = req->errorno;
198 235
199 ENTER; 236 ENTER;
… …
257 for (i = 0; i <= AvFILL (av); ++i) 294 for (i = 0; i <= AvFILL (av); ++i)
258 PUSHs (*av_fetch (av, i, 0)); 295 PUSHs (*av_fetch (av, i, 0));
259 } 296 }
260 break; 297 break;
261 298
299 case REQ_NOP:
262 case REQ_SLEEP: 300 case REQ_SLEEP:
263 break; 301 break;
264 302
265 default: 303 default:
266 PUSHs (sv_2mortal (newSViv (req->result))); 304 PUSHs (sv_2mortal (newSViv (req->result)));
… …
318 Safefree (req); 356 Safefree (req);
319} 357}
320 358
321static void req_cancel (aio_req req) 359static void req_cancel (aio_req req)
322{ 360{
323 req->cancelled = 1; 361 req->flags |= FLAG_CANCELLED;
324 362
325 if (req->type == REQ_GROUP) 363 if (req->type == REQ_GROUP)
326 { 364 {
327 aio_req sub; 365 aio_req sub;
328 366
… …
782 } 820 }
783 821
784 pthread_mutex_unlock (&reqlock); 822 pthread_mutex_unlock (&reqlock);
785 823
786 errno = 0; /* strictly unnecessary */ 824 errno = 0; /* strictly unnecessary */
787
788 if (!req->cancelled)
789 switch (type = req->type) /* remember type for QUIT check */ 825 type = req->type; /* remember type for QUIT check */
826
827 if (!(req->flags & FLAG_CANCELLED))
828 switch (type)
790 { 829 {
791 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 830 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; 831 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
793 832
794 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 833 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
… …
818 tv.tv_usec = req->fd2; 857 tv.tv_usec = req->fd2;
819 858
820 req->result = select (0, 0, 0, 0, &tv); 859 req->result = select (0, 0, 0, 0, &tv);
821 } 860 }
822 861
862 case REQ_GROUP:
863 case REQ_NOP:
823 case REQ_QUIT: 864 case REQ_QUIT:
824 break; 865 break;
825 866
826 default: 867 default:
827 req->result = ENOSYS; 868 req->result = ENOSYS;
… …
832 873
833 pthread_mutex_lock (&reslock); 874 pthread_mutex_lock (&reslock);
834 875
835 req->next = 0; 876 req->next = 0;
836 877
837 printf ("queue rese %p\n", rese);//D
838 if (rese) 878 if (rese)
839 { 879 {
840 rese->next = req; 880 rese->next = req;
841 rese = req; 881 rese = req;
842 } 882 }
… …
912 atfork_parent (); 952 atfork_parent ();
913} 953}
914 954
915#define dREQ \ 955#define dREQ \
916 aio_req req; \ 956 aio_req req; \
957 int req_pri = next_pri; \
958 next_pri = DEFAULT_PRI + PRI_BIAS; \
917 \ 959 \
918 if (SvOK (callback) && !SvROK (callback)) \ 960 if (SvOK (callback) && !SvROK (callback)) \
919 croak ("callback must be undef or of reference type"); \ 961 croak ("callback must be undef or of reference type"); \
920 \ 962 \
921 Newz (0, req, 1, aio_cb); \ 963 Newz (0, req, 1, aio_cb); \
922 if (!req) \ 964 if (!req) \
923 croak ("out of memory during aio_req allocation"); \ 965 croak ("out of memory during aio_req allocation"); \
924 \ 966 \
925 req->callback = newSVsv (callback) 967 req->callback = newSVsv (callback); \
968 req->pri = req_pri
926 969
927#define REQ_SEND \ 970#define REQ_SEND \
928 req_send (req); \ 971 req_send (req); \
929 \ 972 \
930 if (GIMME_V != G_VOID) \ 973 if (GIMME_V != G_VOID) \
… …
1205 SV * callback 1248 SV * callback
1206 PROTOTYPE: ;$ 1249 PROTOTYPE: ;$
1207 PPCODE: 1250 PPCODE:
1208{ 1251{
1209 dREQ; 1252 dREQ;
1253
1210 req->type = REQ_GROUP; 1254 req->type = REQ_GROUP;
1211 req_send (req); 1255 req_send (req);
1256
1212 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1257 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1213} 1258}
1259
1260void
1261aio_nop (callback=&PL_sv_undef)
1262 SV * callback
1263 PPCODE:
1264{
1265 dREQ;
1266
1267 req->type = REQ_NOP;
1268
1269 REQ_SEND;
1270}
1271
1272#if 0
1273
1274void
1275aio_pri (int pri = DEFAULT_PRI)
1276 CODE:
1277 if (pri < PRI_MIN) pri = PRI_MIN;
1278 if (pri > PRI_MAX) pri = PRI_MAX;
1279 next_pri = pri + PRI_BIAS;
1280
1281#endif
1214 1282
1215void 1283void
1216flush () 1284flush ()
1217 PROTOTYPE: 1285 PROTOTYPE:
1218 CODE: 1286 CODE:
… …
1271cancel (aio_req_ornot req) 1339cancel (aio_req_ornot req)
1272 PROTOTYPE: 1340 PROTOTYPE:
1273 CODE: 1341 CODE:
1274 req_cancel (req); 1342 req_cancel (req);
1275 1343
1344void
1345cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1346 CODE:
1347 SvREFCNT_dec (req->callback);
1348 req->callback = newSVsv (callback);
1349
1276MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1350MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1277 1351
1278void 1352void
1279add (aio_req grp, ...) 1353add (aio_req grp, ...)
1280 PPCODE: 1354 PPCODE:
1281{ 1355{
1282 int i; 1356 int i;
1357 aio_req req;
1283 1358
1284 if (grp->fd == 2) 1359 if (grp->fd == 2)
1285 croak ("cannot add requests to IO::AIO::GRP after the group finished"); 1360 croak ("cannot add requests to IO::AIO::GRP after the group finished");
1286 1361
1287 for (i = 1; i < items; ++i ) 1362 for (i = 1; i < items; ++i )
1288 { 1363 {
1289 if (GIMME_V != G_VOID) 1364 if (GIMME_V != G_VOID)
1290 XPUSHs (sv_2mortal (newSVsv (ST (i)))); 1365 XPUSHs (sv_2mortal (newSVsv (ST (i))));
1291 1366
1292 aio_req req = SvAIO_REQ (ST (i)); 1367 req = SvAIO_REQ (ST (i));
1293 1368
1294 if (req) 1369 if (req)
1295 { 1370 {
1296 ++grp->length; 1371 ++grp->length;
1297 req->grp = grp; 1372 req->grp = grp;
… …
1320 SvREFCNT_dec (grp->data); 1395 SvREFCNT_dec (grp->data);
1321 grp->data = (SV *)av; 1396 grp->data = (SV *)av;
1322} 1397}
1323 1398
1324void 1399void
1325lock (aio_req grp)
1326 CODE:
1327 ++grp->length;
1328
1329void
1330unlock (aio_req grp)
1331 CODE:
1332 aio_grp_dec (grp);
1333
1334void
1335feeder_limit (aio_req grp, int limit) 1400feed_limit (aio_req grp, int limit)
1336 CODE: 1401 CODE:
1337 grp->fd2 = limit; 1402 grp->fd2 = limit;
1338 aio_grp_feed (grp); 1403 aio_grp_feed (grp);
1339 1404
1340void 1405void
1341set_feeder (aio_req grp, SV *callback=&PL_sv_undef) 1406feed (aio_req grp, SV *callback=&PL_sv_undef)
1342 CODE: 1407 CODE:
1343{ 1408{
1344 SvREFCNT_dec (grp->fh2); 1409 SvREFCNT_dec (grp->fh2);
1345 grp->fh2 = newSVsv (callback); 1410 grp->fh2 = newSVsv (callback);
1346 1411

Diff Legend

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