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.53 by root, Mon Oct 23 00:17:07 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 *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
… …
130 return mg ? (aio_req)mg->mg_ptr : 0; 156 return mg ? (aio_req)mg->mg_ptr : 0;
131} 157}
132 158
133static void aio_grp_feed (aio_req grp) 159static void aio_grp_feed (aio_req grp)
134{ 160{
135 while (grp->length < grp->fd2) 161 while (grp->length < grp->fd2 && !(grp->flags & FLAG_CANCELLED))
136 { 162 {
137 int old_len = grp->length; 163 int old_len = grp->length;
138 164
139 if (grp->fh2 && SvOK (grp->fh2)) 165 if (grp->fh2 && SvOK (grp->fh2))
140 { 166 {
… …
176 } 202 }
177} 203}
178 204
179static void poll_wait () 205static void poll_wait ()
180{ 206{
181 if (nreqs && !ress)
182 {
183 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
184 FD_ZERO(&rfd); 223 FD_ZERO(&rfd);
185 FD_SET(respipe [0], &rfd); 224 FD_SET(respipe [0], &rfd);
186 225
187 select (respipe [0] + 1, &rfd, 0, 0, 0); 226 select (respipe [0] + 1, &rfd, 0, 0, 0);
188 } 227 }
… …
191static void req_invoke (aio_req req) 230static void req_invoke (aio_req req)
192{ 231{
193 dSP; 232 dSP;
194 int errorno = errno; 233 int errorno = errno;
195 234
196 if (req->cancelled || !SvOK (req->callback)) 235 if (req->flags & FLAG_CANCELLED || !SvOK (req->callback))
197 return; 236 return;
198 237
199 errno = req->errorno; 238 errno = req->errorno;
200 239
201 ENTER; 240 ENTER;
… …
259 for (i = 0; i <= AvFILL (av); ++i) 298 for (i = 0; i <= AvFILL (av); ++i)
260 PUSHs (*av_fetch (av, i, 0)); 299 PUSHs (*av_fetch (av, i, 0));
261 } 300 }
262 break; 301 break;
263 302
303 case REQ_NOP:
264 case REQ_SLEEP: 304 case REQ_SLEEP:
265 break; 305 break;
266 306
267 default: 307 default:
268 PUSHs (sv_2mortal (newSViv (req->result))); 308 PUSHs (sv_2mortal (newSViv (req->result)));
… …
320 Safefree (req); 360 Safefree (req);
321} 361}
322 362
323static void req_cancel (aio_req req) 363static void req_cancel (aio_req req)
324{ 364{
325 req->cancelled = 1; 365 req->flags |= FLAG_CANCELLED;
326 366
327 if (req->type == REQ_GROUP) 367 if (req->type == REQ_GROUP)
328 { 368 {
329 aio_req sub; 369 aio_req sub;
330 370
… …
784 } 824 }
785 825
786 pthread_mutex_unlock (&reqlock); 826 pthread_mutex_unlock (&reqlock);
787 827
788 errno = 0; /* strictly unnecessary */ 828 errno = 0; /* strictly unnecessary */
789
790 if (!req->cancelled)
791 switch (type = req->type) /* remember type for QUIT check */ 829 type = req->type; /* remember type for QUIT check */
830
831 if (!(req->flags & FLAG_CANCELLED))
832 switch (type)
792 { 833 {
793 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;
794 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;
795 836
796 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;
… …
820 tv.tv_usec = req->fd2; 861 tv.tv_usec = req->fd2;
821 862
822 req->result = select (0, 0, 0, 0, &tv); 863 req->result = select (0, 0, 0, 0, &tv);
823 } 864 }
824 865
866 case REQ_GROUP:
867 case REQ_NOP:
825 case REQ_QUIT: 868 case REQ_QUIT:
826 break; 869 break;
827 870
828 default: 871 default:
829 req->result = ENOSYS; 872 req->result = ENOSYS;
… …
913 atfork_parent (); 956 atfork_parent ();
914} 957}
915 958
916#define dREQ \ 959#define dREQ \
917 aio_req req; \ 960 aio_req req; \
961 int req_pri = next_pri; \
962 next_pri = DEFAULT_PRI + PRI_BIAS; \
918 \ 963 \
919 if (SvOK (callback) && !SvROK (callback)) \ 964 if (SvOK (callback) && !SvROK (callback)) \
920 croak ("callback must be undef or of reference type"); \ 965 croak ("callback must be undef or of reference type"); \
921 \ 966 \
922 Newz (0, req, 1, aio_cb); \ 967 Newz (0, req, 1, aio_cb); \
923 if (!req) \ 968 if (!req) \
924 croak ("out of memory during aio_req allocation"); \ 969 croak ("out of memory during aio_req allocation"); \
925 \ 970 \
926 req->callback = newSVsv (callback) 971 req->callback = newSVsv (callback); \
972 req->pri = req_pri
927 973
928#define REQ_SEND \ 974#define REQ_SEND \
929 req_send (req); \ 975 req_send (req); \
930 \ 976 \
931 if (GIMME_V != G_VOID) \ 977 if (GIMME_V != G_VOID) \
… …
1206 SV * callback 1252 SV * callback
1207 PROTOTYPE: ;$ 1253 PROTOTYPE: ;$
1208 PPCODE: 1254 PPCODE:
1209{ 1255{
1210 dREQ; 1256 dREQ;
1257
1211 req->type = REQ_GROUP; 1258 req->type = REQ_GROUP;
1212 req_send (req); 1259 req_send (req);
1260
1213 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1261 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1214} 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
1215 1286
1216void 1287void
1217flush () 1288flush ()
1218 PROTOTYPE: 1289 PROTOTYPE:
1219 CODE: 1290 CODE:
… …
1272cancel (aio_req_ornot req) 1343cancel (aio_req_ornot req)
1273 PROTOTYPE: 1344 PROTOTYPE:
1274 CODE: 1345 CODE:
1275 req_cancel (req); 1346 req_cancel (req);
1276 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
1277MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1354MODULE = IO::AIO PACKAGE = IO::AIO::GRP
1278 1355
1279void 1356void
1280add (aio_req grp, ...) 1357add (aio_req grp, ...)
1281 PPCODE: 1358 PPCODE:
… …
1322 SvREFCNT_dec (grp->data); 1399 SvREFCNT_dec (grp->data);
1323 grp->data = (SV *)av; 1400 grp->data = (SV *)av;
1324} 1401}
1325 1402
1326void 1403void
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) 1404feed_limit (aio_req grp, int limit)
1338 CODE: 1405 CODE:
1339 grp->fd2 = limit; 1406 grp->fd2 = limit;
1340 aio_grp_feed (grp); 1407 aio_grp_feed (grp);
1341 1408
1342void 1409void
1343set_feeder (aio_req grp, SV *callback=&PL_sv_undef) 1410feed (aio_req grp, SV *callback=&PL_sv_undef)
1344 CODE: 1411 CODE:
1345{ 1412{
1346 SvREFCNT_dec (grp->fh2); 1413 SvREFCNT_dec (grp->fh2);
1347 grp->fh2 = newSVsv (callback); 1414 grp->fh2 = newSVsv (callback);
1348 1415

Diff Legend

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