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.56 by root, Mon Oct 23 14:49:51 2006 UTC vs.
Revision 1.60 by root, Mon Oct 23 20:05:27 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
88 unsigned char cancelled; 85 unsigned char flags;
86 unsigned char pri;
87
88 SV *self; /* the perl counterpart of this request, if any */
89 struct aio_cb *grp, *grp_prev, *grp_next, *grp_first;
89} aio_cb; 90} aio_cb;
91
92enum {
93 FLAG_CANCELLED = 0x01,
94};
90 95
91typedef aio_cb *aio_req; 96typedef aio_cb *aio_req;
92typedef 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;
93 108
94static int started, wanted; 109static int started, wanted;
95static volatile int nreqs; 110static volatile int nreqs;
96static int max_outstanding = 1<<30; 111static int max_outstanding = 1<<30;
97static int respipe [2]; 112static int respipe [2];
… …
130 return mg ? (aio_req)mg->mg_ptr : 0; 145 return mg ? (aio_req)mg->mg_ptr : 0;
131} 146}
132 147
133static void aio_grp_feed (aio_req grp) 148static void aio_grp_feed (aio_req grp)
134{ 149{
135 while (grp->length < grp->fd2) 150 while (grp->length < grp->fd2 && !(grp->flags & FLAG_CANCELLED))
136 { 151 {
137 int old_len = grp->length; 152 int old_len = grp->length;
138 153
139 if (grp->fh2 && SvOK (grp->fh2)) 154 if (grp->fh2 && SvOK (grp->fh2))
140 { 155 {
… …
176 } 191 }
177} 192}
178 193
179static void poll_wait () 194static void poll_wait ()
180{ 195{
181 if (nreqs && !ress) 196 while (nreqs)
182 { 197 {
198 aio_req req;
199 pthread_mutex_lock (&reslock);
200 req = ress;
201 pthread_mutex_unlock (&reslock);
202
203 if (req)
204 return;
205
183 fd_set rfd; 206 fd_set rfd;
184 FD_ZERO(&rfd); 207 FD_ZERO(&rfd);
185 FD_SET(respipe [0], &rfd); 208 FD_SET(respipe [0], &rfd);
186 209
187 select (respipe [0] + 1, &rfd, 0, 0, 0); 210 select (respipe [0] + 1, &rfd, 0, 0, 0);
… …
191static void req_invoke (aio_req req) 214static void req_invoke (aio_req req)
192{ 215{
193 dSP; 216 dSP;
194 int errorno = errno; 217 int errorno = errno;
195 218
196 if (req->cancelled || !SvOK (req->callback)) 219 if (req->flags & FLAG_CANCELLED || !SvOK (req->callback))
197 return; 220 return;
198 221
199 errno = req->errorno; 222 errno = req->errorno;
200 223
201 ENTER; 224 ENTER;
… …
321 Safefree (req); 344 Safefree (req);
322} 345}
323 346
324static void req_cancel (aio_req req) 347static void req_cancel (aio_req req)
325{ 348{
326 req->cancelled = 1; 349 req->flags |= FLAG_CANCELLED;
327 350
328 if (req->type == REQ_GROUP) 351 if (req->type == REQ_GROUP)
329 { 352 {
330 aio_req sub; 353 aio_req sub;
331 354
… …
785 } 808 }
786 809
787 pthread_mutex_unlock (&reqlock); 810 pthread_mutex_unlock (&reqlock);
788 811
789 errno = 0; /* strictly unnecessary */ 812 errno = 0; /* strictly unnecessary */
790
791 if (!req->cancelled)
792 switch (type = req->type) /* remember type for QUIT check */ 813 type = req->type; /* remember type for QUIT check */
814
815 if (!(req->flags & FLAG_CANCELLED))
816 switch (type)
793 { 817 {
794 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break; 818 case REQ_READ: req->result = pread (req->fd, req->dataptr, req->length, req->offset); break;
795 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break; 819 case REQ_WRITE: req->result = pwrite (req->fd, req->dataptr, req->length, req->offset); break;
796 820
797 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break; 821 case REQ_READAHEAD: req->result = readahead (req->fd, req->offset, req->length); break;
… …
916 atfork_parent (); 940 atfork_parent ();
917} 941}
918 942
919#define dREQ \ 943#define dREQ \
920 aio_req req; \ 944 aio_req req; \
945 int req_pri = next_pri; \
946 next_pri = DEFAULT_PRI + PRI_BIAS; \
921 \ 947 \
922 if (SvOK (callback) && !SvROK (callback)) \ 948 if (SvOK (callback) && !SvROK (callback)) \
923 croak ("callback must be undef or of reference type"); \ 949 croak ("callback must be undef or of reference type"); \
924 \ 950 \
925 Newz (0, req, 1, aio_cb); \ 951 Newz (0, req, 1, aio_cb); \
926 if (!req) \ 952 if (!req) \
927 croak ("out of memory during aio_req allocation"); \ 953 croak ("out of memory during aio_req allocation"); \
928 \ 954 \
929 req->callback = newSVsv (callback) 955 req->callback = newSVsv (callback); \
956 req->pri = req_pri
930 957
931#define REQ_SEND \ 958#define REQ_SEND \
932 req_send (req); \ 959 req_send (req); \
933 \ 960 \
934 if (GIMME_V != G_VOID) \ 961 if (GIMME_V != G_VOID) \
… …
1209 SV * callback 1236 SV * callback
1210 PROTOTYPE: ;$ 1237 PROTOTYPE: ;$
1211 PPCODE: 1238 PPCODE:
1212{ 1239{
1213 dREQ; 1240 dREQ;
1241
1214 req->type = REQ_GROUP; 1242 req->type = REQ_GROUP;
1215 req_send (req); 1243 req_send (req);
1244
1216 XPUSHs (req_sv (req, AIO_GRP_KLASS)); 1245 XPUSHs (req_sv (req, AIO_GRP_KLASS));
1217} 1246}
1218 1247
1219void 1248void
1220aio_nop (callback=&PL_sv_undef) 1249aio_nop (callback=&PL_sv_undef)
… …
1225 1254
1226 req->type = REQ_NOP; 1255 req->type = REQ_NOP;
1227 1256
1228 REQ_SEND; 1257 REQ_SEND;
1229} 1258}
1259
1260#if 0
1261
1262void
1263aio_pri (int pri = DEFAULT_PRI)
1264 CODE:
1265 if (pri > PRI_MIN) pri = PRI_MIN;
1266 if (pri < PRI_MAX) pri = PRI_MAX;
1267 next_pri = pri + PRI_BIAS;
1268
1269#endif
1230 1270
1231void 1271void
1232flush () 1272flush ()
1233 PROTOTYPE: 1273 PROTOTYPE:
1234 CODE: 1274 CODE:
… …
1288 PROTOTYPE: 1328 PROTOTYPE:
1289 CODE: 1329 CODE:
1290 req_cancel (req); 1330 req_cancel (req);
1291 1331
1292void 1332void
1293cb (aio_req req, SV *callback=&PL_sv_undef) 1333cb (aio_req_ornot req, SV *callback=&PL_sv_undef)
1294 CODE: 1334 CODE:
1295 SvREFCNT_dec (req->callback); 1335 SvREFCNT_dec (req->callback);
1296 req->callback = newSVsv (callback); 1336 req->callback = newSVsv (callback);
1297 1337
1298MODULE = IO::AIO PACKAGE = IO::AIO::GRP 1338MODULE = IO::AIO PACKAGE = IO::AIO::GRP

Diff Legend

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