[ViewVC] Diff of: cvs/Linux-AIO/AIO.xs

Comparing Linux-AIO/AIO.xs (file contents):
Revision 1.5 by root, Tue Aug 14 18:06:37 2001 UTC vs.
Revision 1.20 by root, Sun Jul 18 10:55:34 2004 UTC

…		…
		1	#define PERL_NO_GET_CONTEXT
		2
1	#include "EXTERN.h"	3	#include "EXTERN.h"
2	#include "perl.h"	4	#include "perl.h"
3	#include "XSUB.h"	5	#include "XSUB.h"
4		6
5	#include <sys/types.h>	7	#include <sys/types.h>
		8	#include <sys/stat.h>
6	#include <unistd.h>	9	#include <unistd.h>
7	#include <fcntl.h>	10	#include <fcntl.h>
		11	#include <signal.h>
8	#include <sched.h>	12	#include <sched.h>
9		13
		14	typedef void InputStream; / hack, but 5.6.1 is simply toooo old ;) */
		15	typedef void OutputStream; / hack, but 5.6.1 is simply toooo old ;) */
		16	typedef void InOutStream; / hack, but 5.6.1 is simply toooo old ;) */
		17
		18	#ifndef __NR_pread64
		19	# define __NR_pread64 __NR_pread
		20	#endif
		21	#ifndef __NR_pwrite64
		22	# define __NR_pwrite64 __NR_pwrite
		23	#endif
		24
10	#define STACKSIZE 1024 /* yeah */	25	#define STACKSIZE 1024 /* yeah */
11		26
12	#define REQ_QUIT 0	27	enum {
13	#define REQ_READ 1	28	REQ_QUIT,
14	#define REQ_WRITE 2	29	REQ_OPEN, REQ_CLOSE, REQ_READ, REQ_WRITE,
		30	REQ_STAT, REQ_LSTAT, REQ_FSTAT, REQ_UNLINK
		31	};
15		32
16	typedef struct {	33	typedef struct {
17	char stack[STACKSIZE];	34	char stack[STACKSIZE];
18	} aio_thread;	35	} aio_thread;
19		36
20	typedef struct {	37	typedef struct aio_cb {
		38	struct aio_cb *next;
		39
21	int type;	40	int type;
22	aio_thread *thread;	41	aio_thread *thread;
23		42
24	/* read/write */
25	int fd;	43	int fd;
26	off_t offset;	44	off_t offset;
27	size_t length;	45	size_t length;
28	ssize_t result;	46	ssize_t result;
		47	mode_t mode; /* open */
29	int errorno;	48	int errorno;
30
31	SV data, callback;	49	SV data, callback;
32	void *dataptr;	50	void *dataptr;
33	STRLEN dataoffset;	51	STRLEN dataoffset;
		52
		53	struct stat64 *statdata;
34	} aio_cb;	54	} aio_cb;
35		55
36	typedef aio_cb *aio_req;	56	typedef aio_cb *aio_req;
37		57
38	static int started;	58	static int started;
39	static int nreqs;	59	static int nreqs;
40	static int reqpipe[2], respipe[2];	60	static int reqpipe[2], respipe[2];
41		61
		62	static aio_req qs, qe; /* queue start, queue end */
		63
42	static int aio_proc(void *arg);	64	static int aio_proc(void *arg);
43		65
44	static void	66	static void
45	start_thread(void)	67	start_thread (void)
46	{	68	{
47	aio_thread *thr;	69	aio_thread *thr;
48		70
49	New (0, thr, 1, aio_thread);	71	New (0, thr, 1, aio_thread);
50		72
51	if (clone (aio_proc,	73	if (clone (aio_proc,
52	&(thr->stack[STACKSIZE]),	74	&(thr->stack[STACKSIZE]),
53	CLONE_VM\|CLONE_FS\|CLONE_FILES\|CLONE_SIGHAND,	75	CLONE_VM\|CLONE_FS\|CLONE_FILES,
54	thr) >= 0)	76	thr) >= 0)
55	started++;	77	started++;
56	else	78	else
57	Safefree (thr);	79	Safefree (thr);
58	}	80	}
59		81
60	static void	82	static void
		83	send_reqs (void)
		84	{
		85	/* this write is atomic */
		86	while (qs && write (reqpipe[1], &qs, sizeof qs) == sizeof qs)
		87	{
		88	qs = qs->next;
		89	if (!qs) qe = 0;
		90	}
		91	}
		92
		93	static void
		94	send_req (aio_req req)
		95	{
		96	nreqs++;
		97	req->next = 0;
		98
		99	if (qe)
		100	qe->next = req;
		101	else
		102	qe = qs = req;
		103
		104	send_reqs ();
		105	}
		106
		107	static void
61	end_thread(void)	108	end_thread (void)
62	{	109	{
63	aio_req req;	110	aio_req req;
64	New (0, req, 1, aio_cb);	111	New (0, req, 1, aio_cb);
65	req->type = REQ_QUIT;	112	req->type = REQ_QUIT;
66	write (reqpipe[1], &req, sizeof (aio_req));	113
		114	send_req (req);
67	}	115	}
68		116
69	static void	117	static void
		118	read_write (pTHX_
70	read_write (pTHX_ int dowrite, int fd, off_t offset, size_t length,	119	int dowrite, int fd, off_t offset, size_t length,
71	SV data, STRLEN dataoffset, SVcallback)	120	SV data, STRLEN dataoffset, SV callback)
72	{	121	{
73	aio_req req;	122	aio_req req;
74	STRLEN svlen;	123	STRLEN svlen;
75	char *svptr = SvPV (data, svlen);	124	char *svptr = SvPV (data, svlen);
		125
		126	SvUPGRADE (data, SVt_PV);
		127	SvPOK_on (data);
76		128
77	if (dataoffset < 0)	129	if (dataoffset < 0)
78	dataoffset += svlen;	130	dataoffset += svlen;
79		131
80	if (dataoffset < 0 \|\| dataoffset > svlen)	132	if (dataoffset < 0 \|\| dataoffset > svlen)
93	}	145	}
94		146
95	if (length < 0)	147	if (length < 0)
96	croak ("length must not be negative");	148	croak ("length must not be negative");
97		149
98	New (0, req, 1, aio_cb);	150	Newz (0, req, 1, aio_cb);
99		151
100	if (!req)	152	if (!req)
101	croak ("out of memory during aio_req allocation");	153	croak ("out of memory during aio_req allocation");
102		154
103	req->type = dowrite ? REQ_WRITE : REQ_READ;	155	req->type = dowrite ? REQ_WRITE : REQ_READ;
104	req->fd = fd;	156	req->fd = fd;
105	req->offset = offset;	157	req->offset = offset;
106	req->length = length;	158	req->length = length;
107	req->data = SvREFCNT_inc (data);	159	req->data = SvREFCNT_inc (data);
108	req->dataptr = svptr + dataoffset;	160	req->dataptr = (char *)svptr + dataoffset;
109	req->callback = SvREFCNT_inc (callback);	161	req->callback = SvREFCNT_inc (callback);
110		162
111	nreqs++;	163	send_req (req);
112	write (reqpipe[1], &req, sizeof (aio_req));
113	}	164	}
114		165
115	static int	166	static int
116	poll_cb (pTHX)	167	poll_cb (pTHX)
117	{	168	{
…		…
130	{	181	{
131	int errorno = errno;	182	int errorno = errno;
132	errno = req->errorno;	183	errno = req->errorno;
133		184
134	if (req->type == REQ_READ)	185	if (req->type == REQ_READ)
135	SvCUR_set (req->data, req->result > 0	186	SvCUR_set (req->data, req->dataoffset
136	? req->dataoffset + req->result	187	+ req->result > 0 ? req->result : 0);
137	: req->dataoffset);	188
		189	if (req->data)
		190	SvREFCNT_dec (req->data);
		191
		192	if (req->type == REQ_STAT \|\| req->type == REQ_LSTAT \|\| req->type == REQ_FSTAT)
		193	{
		194	PL_laststype = req->type == REQ_LSTAT ? OP_LSTAT : OP_STAT;
		195	PL_laststatval = req->result;
		196	PL_statcache.st_dev = req->statdata->st_dev;
		197	PL_statcache.st_ino = req->statdata->st_ino;
		198	PL_statcache.st_mode = req->statdata->st_mode;
		199	PL_statcache.st_nlink = req->statdata->st_nlink;
		200	PL_statcache.st_uid = req->statdata->st_uid;
		201	PL_statcache.st_gid = req->statdata->st_gid;
		202	PL_statcache.st_rdev = req->statdata->st_rdev;
		203	PL_statcache.st_size = req->statdata->st_size;
		204	PL_statcache.st_atime = req->statdata->st_atime;
		205	PL_statcache.st_mtime = req->statdata->st_mtime;
		206	PL_statcache.st_ctime = req->statdata->st_ctime;
		207	PL_statcache.st_blksize = req->statdata->st_blksize;
		208	PL_statcache.st_blocks = req->statdata->st_blocks;
		209
		210	Safefree (req->statdata);
		211	}
138		212
139	PUSHMARK (SP);	213	PUSHMARK (SP);
140	XPUSHs (sv_2mortal (newSViv (req->result)));	214	XPUSHs (sv_2mortal (newSViv (req->result)));
141	PUTBACK;	215	PUTBACK;
142	call_sv (req->callback, G_VOID);	216	call_sv (req->callback, G_VOID);
143	SPAGAIN;	217	SPAGAIN;
144		218
145	SvREFCNT_dec (req->data);	219	if (req->callback)
146	SvREFCNT_dec (req->callback);	220	SvREFCNT_dec (req->callback);
147		221
148	errno = errorno;	222	errno = errorno;
149	nreqs--;	223	nreqs--;
150	count++;	224	count++;
151	}	225	}
152		226
153	Safefree (req);	227	Safefree (req);
154	}	228	}
155		229
		230	if (qs)
		231	send_reqs ();
		232
156	return count;	233	return count;
157	}	234	}
		235
		236	static sigset_t fullsigset;
158		237
159	#undef errno	238	#undef errno
160	#include <asm/unistd.h>	239	#include <asm/unistd.h>
		240	#include <sys/prctl.h>
161		241
162	static int	242	static int
163	aio_proc(void *thr_arg)	243	aio_proc (void *thr_arg)
164	{	244	{
165	aio_thread *thr = thr_arg;	245	aio_thread *thr = thr_arg;
166	int sig;	246	aio_req req;
167	int errno;	247	int errno;
168	aio_req req;
169		248
		249	/* this is very much x86 and kernel-specific :(:(:( */
170	/* we rely on gcc's ability to create closures. */	250	/* we rely on gcc's ability to create closures. */
171	_syscall3(int,lseek,int,fd,off_t,offset,int,whence);
172	_syscall3(int,read,int,fd,char *,buf,off_t,count);	251	_syscall3(int,read,int,fd,char *,buf,size_t,count)
173	_syscall3(int,write,int,fd,char *,buf,off_t,count);	252	_syscall3(int,write,int,fd,char *,buf,size_t,count)
174		253
175	/* first get rid of any signals */	254	_syscall3(int,open,char *,pathname,int,flags,mode_t,mode)
176	for (sig = 1; sig < _NSIG; sig++)	255	_syscall1(int,close,int,fd)
177	signal (sig, SIG_DFL);
178		256
179	signal (SIGPIPE, SIG_IGN);	257	_syscall5(int,pread64,int,fd,char *,buf,size_t,count,unsigned int,offset_lo,unsigned int,offset_hi)
180		258	_syscall5(int,pwrite64,int,fd,char *,buf,size_t,count,unsigned int,offset_lo,unsigned int,offset_hi)
		259
		260	_syscall2(int,stat64, const char , filename, struct stat64 , buf)
		261	_syscall2(int,lstat64, const char , filename, struct stat64 , buf)
		262	_syscall2(int,fstat64, int, fd, struct stat64 *, buf)
		263
		264	_syscall1(int,unlink, char *, filename);
		265
		266	sigprocmask (SIG_SETMASK, &fullsigset, 0);
		267	prctl (PR_SET_PDEATHSIG, SIGKILL);
		268
181	/* then loop */	269	/* then loop */
182	while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))	270	while (read (reqpipe[0], (void *)&req, sizeof (req)) == sizeof (req))
183	{	271	{
184	req->thread = thr;	272	req->thread = thr;
		273	errno = 0; /* strictly unnecessary */
185		274
186	if (req->type == REQ_READ \|\| req->type == REQ_WRITE)	275	switch (req->type)
187	{	276	{
		277	case REQ_READ: req->result = pread64 (req->fd, req->dataptr, req->length, req->offset & 0xffffffff, req->offset >> 32); break;
		278	case REQ_WRITE: req->result = pwrite64(req->fd, req->dataptr, req->length, req->offset & 0xffffffff, req->offset >> 32); break;
		279	case REQ_OPEN: req->result = open (req->dataptr, req->fd, req->mode); break;
		280	case REQ_CLOSE: req->result = close (req->fd); break;
		281	case REQ_STAT: req->result = stat64 (req->dataptr, req->statdata); break;
		282	case REQ_LSTAT: req->result = lstat64 (req->dataptr, req->statdata); break;
		283	case REQ_FSTAT: req->result = fstat64 (req->fd, req->statdata); break;
		284	case REQ_UNLINK: req->result = unlink (req->dataptr); break;
		285
		286	case REQ_QUIT:
		287	default:
		288	write (respipe[1], (void *)&req, sizeof (req));
188	errno = 0;	289	return 0;
189
190	if (lseek (req->fd, req->offset, SEEK_SET) == req->offset)
191	{
192	if (req->type == REQ_READ)
193	req->result = read (req->fd, req->dataptr, req->length);
194	else
195	req->result = write(req->fd, req->dataptr, req->length);
196	}
197
198	req->errorno = errno;
199	}	290	}
200	else
201	{
202	write (respipe[1], (void *)&req, sizeof (req));
203	break;
204	}
205		291
		292	req->errorno = errno;
206	write (respipe[1], (void *)&req, sizeof (req));	293	write (respipe[1], (void *)&req, sizeof (req));
207	}	294	}
208		295
209	return 0;	296	return 0;
210	}	297	}
211		298
212	MODULE = Linux::AIO PACKAGE = Linux::AIO	299	MODULE = Linux::AIO PACKAGE = Linux::AIO
213		300
214	BOOT:	301	BOOT:
215	{	302	{
		303	sigfillset (&fullsigset);
		304	sigdelset (&fullsigset, SIGTERM);
		305	sigdelset (&fullsigset, SIGQUIT);
		306	sigdelset (&fullsigset, SIGABRT);
		307	sigdelset (&fullsigset, SIGINT);
		308
216	if (pipe (reqpipe) \|\| pipe (respipe))	309	if (pipe (reqpipe) \|\| pipe (respipe))
217	croak ("unable to initialize request or result pipe");	310	croak ("unable to initialize request or result pipe");
		311
		312	if (fcntl (reqpipe[1], F_SETFL, O_NONBLOCK))
		313	croak ("cannot set result pipe to nonblocking mode");
218		314
219	if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))	315	if (fcntl (respipe[0], F_SETFL, O_NONBLOCK))
220	croak ("cannot set result pipe to nonblocking mode");	316	croak ("cannot set result pipe to nonblocking mode");
221	}	317	}
222		318
…		…
238	{	334	{
239	end_thread ();	335	end_thread ();
240	cur--;	336	cur--;
241	}	337	}
242		338
243	poll_cb ();
244	while (started > nthreads)	339	while (started > nthreads)
245	{	340	{
246	sched_yield ();	341	fd_set rfd;
		342	FD_ZERO(&rfd);
		343	FD_SET(respipe[0], &rfd);
		344
		345	select (respipe[0] + 1, &rfd, 0, 0, 0);
247	poll_cb ();	346	poll_cb (aTHX);
248	}	347	}
249		348
250	void	349	void
		350	aio_open(pathname,flags,mode,callback)
		351	SV * pathname
		352	int flags
		353	int mode
		354	SV * callback
		355	PROTOTYPE: $$$$
		356	CODE:
		357	aio_req req;
		358
		359	Newz (0, req, 1, aio_cb);
		360
		361	if (!req)
		362	croak ("out of memory during aio_req allocation");
		363
		364	req->type = REQ_OPEN;
		365	req->data = newSVsv (pathname);
		366	req->dataptr = SvPV_nolen (req->data);
		367	req->fd = flags;
		368	req->mode = mode;
		369	req->callback = SvREFCNT_inc (callback);
		370
		371	send_req (req);
		372
		373	void
		374	aio_close(fh,callback)
		375	InputStream fh
		376	SV * callback
		377	PROTOTYPE: $$
		378	CODE:
		379	aio_req req;
		380
		381	Newz (0, req, 1, aio_cb);
		382
		383	if (!req)
		384	croak ("out of memory during aio_req allocation");
		385
		386	req->type = REQ_CLOSE;
		387	req->fd = PerlIO_fileno (fh);
		388	req->callback = SvREFCNT_inc (callback);
		389
		390	send_req (req);
		391
		392	void
251	aio_read(fh,offset,length,data,dataoffset,callback)	393	aio_read(fh,offset,length,data,dataoffset,callback)
252	PerlIO * fh	394	InputStream fh
253	unsigned long offset	395	UV offset
254	long length	396	IV length
255	SV * data	397	SV * data
256	STRLEN dataoffset	398	IV dataoffset
257	SV * callback	399	SV * callback
258	PROTOTYPE: $$$$$$	400	PROTOTYPE: $$$$$$
259	ALIAS:	401	CODE:
260	aio_write = 1
261	CODE:
262	sv_upgrade (data, SVt_PV);
263	read_write (aTHX_ ix, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);	402	read_write (aTHX_ 0, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
		403
		404	void
		405	aio_write(fh,offset,length,data,dataoffset,callback)
		406	OutputStream fh
		407	UV offset
		408	IV length
		409	SV * data
		410	IV dataoffset
		411	SV * callback
		412	PROTOTYPE: $$$$$$
		413	CODE:
		414	read_write (aTHX_ 1, PerlIO_fileno (fh), offset, length, data, dataoffset, callback);
		415
		416	void
		417	aio_stat(fh_or_path,callback)
		418	SV * fh_or_path
		419	SV * callback
		420	PROTOTYPE: $$
		421	ALIAS:
		422	aio_lstat = 1
		423	CODE:
		424	aio_req req;
		425
		426	Newz (0, req, 1, aio_cb);
		427
		428	if (!req)
		429	croak ("out of memory during aio_req allocation");
		430
		431	New (0, req->statdata, 1, struct stat64);
		432
		433	if (!req->statdata)
		434	croak ("out of memory during aio_req->statdata allocation");
		435
		436	if (SvPOK (fh_or_path))
		437	{
		438	req->type = ix ? REQ_LSTAT : REQ_STAT;
		439	req->data = newSVsv (fh_or_path);
		440	req->dataptr = SvPV_nolen (req->data);
		441	}
		442	else
		443	{
		444	req->type = REQ_FSTAT;
		445	req->fd = PerlIO_fileno (IoIFP (sv_2io (fh_or_path)));
		446	}
		447
		448	req->callback = SvREFCNT_inc (callback);
		449
		450	send_req (req);
		451
		452	void
		453	aio_unlink(pathname,callback)
		454	SV * pathname
		455	SV * callback
		456	PROTOTYPE: $$
		457	CODE:
		458	aio_req req;
		459
		460	Newz (0, req, 1, aio_cb);
		461
		462	if (!req)
		463	croak ("out of memory during aio_req allocation");
		464
		465	req->type = REQ_UNLINK;
		466	req->data = newSVsv (pathname);
		467	req->dataptr = SvPV_nolen (req->data);
		468	req->callback = SvREFCNT_inc (callback);
		469
		470	send_req (req);
264		471
265	int	472	int
266	poll_fileno()	473	poll_fileno()
267	PROTOTYPE:	474	PROTOTYPE:
268	CODE:	475	CODE:
269	RETVAL = respipe[0];	476	RETVAL = respipe[0];
270	OUTPUT:	477	OUTPUT:
271	RETVAL	478	RETVAL
272		479
273	int	480	int
274	poll_cb()	481	poll_cb(...)
275	PROTOTYPE:	482	PROTOTYPE:
276	CODE:	483	CODE:
277	RETVAL = poll_cb (aTHX);	484	RETVAL = poll_cb (aTHX);
278	OUTPUT:	485	OUTPUT:
279	RETVAL	486	RETVAL

Diff Legend

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

Comparing Linux-AIO/AIO.xs (file contents): Revision 1.5 by root, Tue Aug 14 18:06:37 2001 UTC vs. Revision 1.20 by root, Sun Jul 18 10:55:34 2004 UTC

Diff Legend

Comparing Linux-AIO/AIO.xs (file contents):
Revision 1.5 by root, Tue Aug 14 18:06:37 2001 UTC vs.
Revision 1.20 by root, Sun Jul 18 10:55:34 2004 UTC