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.58 by root, Mon Oct 23 18:48:08 2006 UTC vs.
Revision 1.62 by root, Mon Oct 23 22:45:18 2006 UTC

… …
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 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 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
85 unsigned char flags;
88 unsigned char pri; 86 unsigned char pri;
89 unsigned char flags; 87
88 SV *self; /* the perl counterpart of this request, if any */
89 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
90} aio_cb; 90} aio_cb;
91 91
92enum { 92enum {
93 FLAG_CANCELLED = 0x01, 93 FLAG_CANCELLED = 0x01,
94}; 94};
95 95
96typedef aio_cb *aio_req; 96typedef aio_cb *aio_req;
97typedef 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;
98 108
99static int started, wanted; 109static int started, wanted;
100static volatile int nreqs; 110static volatile int nreqs;
101static int max_outstanding = 1<<30; 111static int max_outstanding = 1<<30;
102static int respipe [2]; 112static int respipe [2];
… …
181 } 191 }
182} 192}
183 193
184static void poll_wait () 194static void poll_wait ()
185{ 195{
186 if (nreqs && !ress)
187 {
188 fd_set rfd; 196 fd_set rfd;
197
198 while (nreqs)
199 {
200 aio_req req;
201 pthread_mutex_lock (&reslock);
202 req = ress;
203 pthread_mutex_unlock (&reslock);
204
205 if (req)
206 return;
207
189 FD_ZERO(&rfd); 208 FD_ZERO(&rfd);
190 FD_SET(respipe [0], &rfd); 209 FD_SET(respipe [0], &rfd);
191 210
192 select (respipe [0] + 1, &rfd, 0, 0, 0); 211 select (respipe [0] + 1, &rfd, 0, 0, 0);
193 } 212 }
… …
790 } 809 }
791 810
792 pthread_mutex_unlock (&reqlock); 811 pthread_mutex_unlock (&reqlock);
793 812
794 errno = 0; /* strictly unnecessary */ 813 errno = 0; /* strictly unnecessary */
814 type = req->type; /* remember type for QUIT check */
795 815
796 if (!(req->flags & FLAG_CANCELLED)) 816 if (!(req->flags & FLAG_CANCELLED))
797 switch (type = req->type) /* remember type for QUIT check */ 817 switch (type)
798 { 818 {
799 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 819 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
800 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 820 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
801 821
802 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 822 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
… …
921 atfork_parent (); 941 atfork_parent ();
922} 942}
923 943
924#define dREQ \ 944#define dREQ \
925 aio_req req; \ 945 aio_req req; \
946 int req_pri = next_pri; \
947 next_pri = DEFAULT_PRI + PRI_BIAS; \
926 \ 948 \
927 if (SvOK (callback) && !SvROK (callback)) \ 949 if (SvOK (callback) && !SvROK (callback)) \
928 croak ("callback must be undef or of reference type"); \ 950 croak ("callback must be undef or of reference type"); \
929 \ 951 \
930 Newz (0, req, 1, aio_cb); \ 952 Newz (0, req, 1, aio_cb); \
931 if (!req) \ 953 if (!req) \
932 croak ("out of memory during aio_req allocation"); \ 954 croak ("out of memory during aio_req allocation"); \
933 \ 955 \
934 req->callback = newSVsv (callback) 956 req->callback = newSVsv (callback); \
957 req->pri = req_pri
935 958
936#define REQ_SEND \ 959#define REQ_SEND \
937 req_send (req); \ 960 req_send (req); \
938 \ 961 \
939 if (GIMME_V != G_VOID) \ 962 if (GIMME_V != G_VOID) \
… …
1214 SV * callback 1237 SV * callback
1215 PROTOTYPE: ;$ 1238 PROTOTYPE: ;$
1216 PPCODE: 1239 PPCODE:
1217{ 1240{
1218 dREQ; 1241 dREQ;
1242
1219 req->type = REQ_GROUP; 1243 req->type = REQ_GROUP;
1220 req_send (req); 1244 req_send (req);
1245
1221 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1246 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1222} 1247}
1223 1248
1224void 1249void
1225aio_nop (callback=&PL_sv_undef) 1250aio_nop (callback=&PL_sv_undef)
… …
1230 1255
1231 req->type = REQ_NOP; 1256 req->type = REQ_NOP;
1232 1257
1233 REQ_SEND; 1258 REQ_SEND;
1234} 1259}
1260
1261#if 0
1262
1263void
1264aio_pri (int pri = DEFAULT_PRI)
1265 CODE:
1266 if (pri < PRI_MIN) pri = PRI_MIN;
1267 if (pri > PRI_MAX) pri = PRI_MAX;
1268 next_pri = pri + PRI_BIAS;
1269
1270#endif
1235 1271
1236void 1272void
1237flush () 1273flush ()
1238 PROTOTYPE: 1274 PROTOTYPE:
1239 CODE: 1275 CODE:
… …
1293 PROTOTYPE: 1329 PROTOTYPE:
1294 CODE: 1330 CODE:
1295 req_cancel (req); 1331 req_cancel (req);
1296 1332
1297void 1333void
1298cb (aio_req req, SV *callback=&PL_sv_undef) 1334cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1299 CODE: 1335 CODE:
1300 SvREFCNT_dec (req->callback); 1336 SvREFCNT_dec (req->callback);
1301 req->callback = newSVsv (callback); 1337 req->callback = newSVsv (callback);
1302 1338
1303MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1339MODULE = IO::AIO PACKAGE = IO::AIO::GRP

Diff Legend

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