ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/Linux-AIO/AIO.xs
(Generate patch)

Comparing Linux-AIO/AIO.xs (file contents):
Revision 1.17 by root, Thu May 6 12:16:48 2004 UTC vs.
Revision 1.18 by root, Thu May 6 15:05:57 2004 UTC

… …
28 28
29typedef struct { 29typedef struct {
30 char stack[STACKSIZE]; 30 char stack[STACKSIZE];
31} aio_thread; 31} aio_thread;
32 32
33typedef struct { 33typedef struct aio_cb {
34 struct aio_cb *next;
35
34 int type; 36 int type;
35 aio_thread *thread; 37 aio_thread *thread;
36 38
37 int fd; 39 int fd;
38 off_t offset; 40 off_t offset;
… …
51 53
52static int started; 54static int started;
53static int nreqs; 55static int nreqs;
54static int reqpipe[2], respipe[2]; 56static int reqpipe[2], respipe[2];
55 57
58static aio_req qs, qe; /* queue start, queue end */
59
56static int aio_proc(void *arg); 60static int aio_proc(void *arg);
57 61
58static void 62static void
59start_thread(void) 63start_thread (void)
60{ 64{
61 aio_thread *thr; 65 aio_thread *thr;
62 66
63 New (0, thr, 1, aio_thread); 67 New (0, thr, 1, aio_thread);
64 68
… …
70 else 74 else
71 Safefree (thr); 75 Safefree (thr);
72} 76}
73 77
74static void 78static void
79send_reqs (void)
80{
81 /* this write is atomic */
82 while (qs && write (reqpipe[1], &qs, sizeof qs) == sizeof qs)
83 {
84 qs = qs->next;
85 if (!qs) qe = 0;
86 }
87}
88
89static void
90send_req (aio_req req)
91{
92 nreqs++;
93 req->next = 0;
94
95 if (qe)
96 qe->next = req;
97 else
98 qe = qs = req;
99
100 send_reqs ();
101}
102
103static void
75end_thread(void) 104end_thread (void)
76{ 105{
77 aio_req req; 106 aio_req req;
78 New (0, req, 1, aio_cb); 107 New (0, req, 1, aio_cb);
79 req->type = REQ_QUIT; 108 req->type = REQ_QUIT;
80 write (reqpipe[1], &req, sizeof (aio_req));
81}
82 109
83static void 110 send_req (req);
84send_req (aio_req req)
85{
86 nreqs++;
87 write (reqpipe[1], &req, sizeof (aio_req));
88} 111}
89 112
90static void 113static void
91read_write (pTHX_ 114read_write (pTHX_
92 int dowrite, int fd, off_t offset, size_t length, 115 int dowrite, int fd, off_t offset, size_t length,
93 SV *data, STRLEN dataoffset, SV*callback) 116 SV *data, STRLEN dataoffset, SV *callback)
94{ 117{
95 aio_req req; 118 aio_req req;
96 STRLEN svlen; 119 STRLEN svlen;
97 char *svptr = SvPV (data, svlen); 120 char *svptr = SvPV (data, svlen);
98 121
… …
198 } 221 }
199 222
200 Safefree (req); 223 Safefree (req);
201 } 224 }
202 225
226 if (qs)
227 {
228 printf ("outstanding %p %d\n", qs, nreqs);
229 send_reqs ();
230 }
231
203 return count; 232 return count;
204} 233}
205 234
206static sigset_t fullsigset; 235static sigset_t fullsigset;
207 236
208#undef errno 237#undef errno
209#include <asm/unistd.h> 238#include <asm/unistd.h>
210 239
211static int 240static int
212aio_proc(void *thr_arg) 241aio_proc (void *thr_arg)
213{ 242{
214 aio_thread *thr = thr_arg; 243 aio_thread *thr = thr_arg;
215 aio_req req; 244 aio_req req;
216 int errno; 245 int errno;
217 246
… …
236 while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req)) 265 while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
237 { 266 {
238 req->thread = thr; 267 req->thread = thr;
239 errno = 0; /* strictly unnecessary */ 268 errno = 0; /* strictly unnecessary */
240 269
241 if (req->type == REQ_READ) 270 switch (req->type)
242 req->result = pread64 (req->fd, req->dataptr, req->length, req->offset & 0xffffffff, req->offset >> 32);
243 else if (req->type == REQ_WRITE)
244 req->result = pwrite64(req->fd, req->dataptr, req->length, req->offset & 0xffffffff, req->offset >> 32);
245 else if (req->type == REQ_OPEN)
246 req->result = open (req->dataptr, req->fd, req->mode);
247 else if (req->type == REQ_CLOSE)
248 req->result = close (req->fd);
249 else if (req->type == REQ_STAT)
250 req->result = stat64 (req->dataptr, req->statdata);
251 else if (req->type == REQ_LSTAT)
252 req->result = lstat64 (req->dataptr, req->statdata);
253 else if (req->type == REQ_FSTAT)
254 req->result = fstat64 (req->fd, req->statdata);
255 else
256 { 271 {
272 case REQ_READ: req->result = pread64 (req->fd, req->dataptr, req->length, req->offset & 0xffffffff, req->offset >> 32); break;
273 case REQ_WRITE: req->result = pwrite64(req->fd, req->dataptr, req->length, req->offset & 0xffffffff, req->offset >> 32); break;
274 case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break;
275 case REQ_CLOSE: req->result = close (req->fd); break;
276 case REQ_STAT: req->result = stat64 (req->dataptr, req->statdata); break;
277 case REQ_LSTAT: req->result = lstat64 (req->dataptr, req->statdata); break;
278 case REQ_FSTAT: req->result = fstat64 (req->fd, req->statdata); break;
279
280 case REQ_QUIT:
281 default:
257 write (respipe[1], (void *)&req, sizeof (req)); 282 write (respipe[1], (void *)&req, sizeof (req));
258 break; 283 return 0;
259 } 284 }
260 285
261 req->errorno = errno; 286 req->errorno = errno;
262 write (respipe[1], (void *)&req, sizeof (req)); 287 write (respipe[1], (void *)&req, sizeof (req));
263 } 288 }
… …
275 sigdelset (&fullsigset, SIGABRT); 300 sigdelset (&fullsigset, SIGABRT);
276 sigdelset (&fullsigset, SIGINT); 301 sigdelset (&fullsigset, SIGINT);
277 302
278 if (pipe (reqpipe) || pipe (respipe)) 303 if (pipe (reqpipe) || pipe (respipe))
279 croak ("unable to initialize request or result pipe"); 304 croak ("unable to initialize request or result pipe");
305
306 if (fcntl (reqpipe[1], F_SETFL, O_NONBLOCK))
307 croak ("cannot set result pipe to nonblocking mode");
280 308
281 if (fcntl (respipe[0], F_SETFL, O_NONBLOCK)) 309 if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))
282 croak ("cannot set result pipe to nonblocking mode"); 310 croak ("cannot set result pipe to nonblocking mode");
283} 311}
284 312

Diff Legend

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