[ViewVC] Diff of: cvs/gvpe/src/iom.C

Comparing gvpe/src/iom.C (file contents):
Revision 1.3 by pcg, Fri Mar 21 21:21:02 2003 UTC vs.
Revision 1.28 by pcg, Mon Jan 10 19:21:07 2005 UTC

…		…
1	/*	1	/*
2	iom.C -- I/O multiplexor	2	iom.C -- generic I/O multiplexer
		3	Copyright (C) 2003, 2004 Marc Lehmann <pcg@goof.com>
3		4
4	This program is free software; you can redistribute it and/or modify	5	This program is free software; you can redistribute it and/or modify
5	it under the terms of the GNU General Public License as published by	6	it under the terms of the GNU General Public License as published by
6	the Free Software Foundation; either version 2 of the License, or	7	the Free Software Foundation; either version 2 of the License, or
7	(at your option) any later version.	8	(at your option) any later version.
14	You should have received a copy of the GNU General Public License	15	You should have received a copy of the GNU General Public License
15	along with this program; if not, write to the Free Software	16	along with this program; if not, write to the Free Software
16	Foundation, Inc. 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA	17	Foundation, Inc. 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17	*/	18	*/
18		19
19	#include "config.h"	20	#include "iom.h"
		21
		22	#include <cstdio>
		23	#include <cstdlib>
		24	#include <cerrno>
20		25
21	#include <sys/time.h>	26	#include <sys/time.h>
22		27
23	#include <algorithm>	28	#include <assert.h>
		29
		30	#if 1 // older unices need these includes for select (2)
		31	# include <unistd.h>
		32	# include <sys/types.h>
		33	# include <time.h>
		34	#endif
		35
		36	// for IOM_SIG
		37	#if IOM_SIG
24	#include <functional>	38	# include <csignal>
		39	# include <fcntl.h>
		40	#endif
25		41
26	#include "slog.h"	42	// if the BSDs would at least be marginally POSIX-compatible.. sigh
27	#include "iom.h"	43	// until that happens, sys/select.h must come last
		44	#include <sys/select.h>
28		45
29	inline bool lowest_first (const time_watcher a, const time_watcher b)	46	// TSTAMP_MAX must still fit into a positive struct timeval
30	{	47	#define TSTAMP_MAX (double)(1UL<<31)
31	return a->at > b->at;	48
32	}	49	#define TIMEVAL timeval
		50	#define TV_FRAC tv_usec
		51	#define TV_MULT 1000000L
		52
		53	#if IOM_IO
		54	static io_manager_vec<io_watcher> iow;
		55	#endif
		56	#if IOM_CHECK
		57	static io_manager_vec<check_watcher> cw;
		58	#endif
		59	#if IOM_TIME
		60	static io_manager_vec<time_watcher> tw;
		61	#endif
		62	#if IOM_IDLE
		63	static io_manager_vec<idle_watcher> iw;
		64	#endif
		65	#if IOM_SIG
		66	static int sigpipe[2]; // signal signalling pipe
		67	static sigset_t sigs;
		68	struct sig_vec : io_manager_vec<sig_watcher> {
		69	int pending;
		70	sig_vec ()
		71	: pending (false)
		72	{ }
		73	};
		74	static vector<sig_vec *> sw;
		75	#endif
		76
		77	// this is a dummy time watcher to ensure that the first
		78	// time watcher is _always_ valid, this gets rid of a lot
		79	// of null-pointer-checks
		80	// (must come _before_ iom is being defined)
		81	static struct tw0 : time_watcher
		82	{
		83	void cb (time_watcher &w)
		84	{
		85	// should never get called
		86	// reached end-of-time, or tstamp has a bogus definition,
		87	// or compiler initialisation order broken, or something else :)
		88	abort ();
		89	}
		90
		91	tw0 ()
		92	: time_watcher (this, &tw0::cb)
		93	{ }
		94	} tw0;
33		95
34	tstamp NOW;	96	tstamp NOW;
35		97
36	io_manager iom;	98	#if IOM_TIME
37		99	tstamp io_manager::now ()
38	void time_watcher::set (tstamp when)
39	{	100	{
		101	struct timeval tv;
		102
		103	gettimeofday (&tv, 0);
		104	return (tstamp)tv.tv_sec + (tstamp)tv.tv_usec / 1000000.;
		105	}
		106
		107	void io_manager::set_now ()
		108	{
		109	NOW = now ();
		110	}
		111	#endif
		112
		113	static bool iom_valid;
		114
		115	// used for initialisation only
		116	static struct init {
		117	init ()
		118	{
		119	#if IOM_SIG
		120	sigemptyset (&sigs);
		121
		122	if (pipe (sigpipe))
		123	{
		124	perror ("io_manager: unable to create signal pipe, aborting.");
		125	abort ();
		126	}
		127
		128	fcntl (sigpipe[0], F_SETFL, O_NONBLOCK); fcntl (sigpipe[0], F_SETFD, FD_CLOEXEC);
		129	fcntl (sigpipe[1], F_SETFL, O_NONBLOCK); fcntl (sigpipe[1], F_SETFD, FD_CLOEXEC);
		130	#endif
		131
		132	iom_valid = true;
		133
		134	#if IOM_TIME
		135	io_manager::set_now ();
		136
		137	tw0.start (TSTAMP_MAX);
		138	#endif
		139	}
		140
		141	static void required ();
		142	} init;
		143
		144	void
		145	init::required ()
		146	{
		147	if (!iom_valid)
		148	{
		149	write (2, "io_manager: early registration attempt, aborting.\n",
		150	sizeof ("io_manager: early registration attempt, aborting.\n") - 1);
		151	abort ();
		152	}
		153	}
		154
		155	template<class watcher>
		156	void io_manager::reg (watcher &w, io_manager_vec<watcher> &queue)
		157	{
		158	init::required ();
		159
		160	if (!w.active)
		161	{
		162	queue.push_back (&w);
		163	w.active = queue.size ();
		164	}
		165	}
		166
		167	template<class watcher>
		168	void io_manager::unreg (watcher &w, io_manager_vec<watcher> &queue)
		169	{
		170	if (!iom_valid)
		171	return;
		172
		173	if (w.active)
		174	{
		175	queue [w.active - 1] = 0;
		176	w.active = 0;
		177	}
		178	}
		179
		180	#if IOM_TIME
		181	void time_watcher::trigger ()
		182	{
		183	call (*this);
40	iom.unreg (this);	184	io_manager::reg (*this);
41	at = when;
42	iom.reg (this);
43	}	185	}
44		186
45	void io_manager::reg (int fd, short events, io_watcher *w)	187	void io_manager::reg (time_watcher &w) { io_manager::reg (w, tw); }
46	{	188	void io_manager::unreg (time_watcher &w) { io_manager::unreg (w, tw); }
47	pollfd pfd;	189	#endif
48		190
49	pfd.fd = fd;	191	#if IOM_IO
50	pfd.events = events;	192	void io_manager::reg (io_watcher &w) { io_manager::reg (w, iow); }
		193	void io_manager::unreg (io_watcher &w) { io_manager::unreg (w, iow); }
		194	#endif
51		195
52	pfs.push_back (pfd);	196	#if IOM_CHECK
53	iow.push_back (w);	197	void io_manager::reg (check_watcher &w) { io_manager::reg (w, cw); }
54	}	198	void io_manager::unreg (check_watcher &w) { io_manager::unreg (w, cw); }
		199	#endif
55		200
		201	#if IOM_IDLE
		202	void io_manager::reg (idle_watcher &w) { io_manager::reg (w, iw); }
		203	void io_manager::unreg (idle_watcher &w) { io_manager::unreg (w, iw); }
		204	#endif
		205
		206	#if IOM_SIG
		207	static void
		208	sighandler (int signum)
		209	{
		210	sw [signum - 1]->pending = true;
		211
		212	// we use a pipe for signal notifications, as most current
		213	// OSes (Linux...) do not implement pselect correctly. ugh.
		214	char ch = signum; // actual content not used
		215	write (sigpipe[1], &ch, 1);
		216	}
		217
56	void io_manager::unreg (io_watcher *w)	218	void io_manager::reg (sig_watcher &w)
57	{	219	{
58	unsigned int sz = iow.size ();	220	assert (0 < w.signum);
59	unsigned int i = find (iow.begin (), iow.end (), w) - iow.begin ();
60		221
		222	sw.reserve (w.signum);
		223
		224	while (sw.size () < w.signum) // pathetic
		225	sw.push_back (0);
		226
		227	sig_vec *&sv = sw[w.signum - 1];
		228
61	if (i != sz)	229	if (!sv)
62	{	230	{
63	if (sz == 1)	231	sv = new sig_vec;
		232
		233	sigaddset (&sigs, w.signum);
		234	sigprocmask (SIG_BLOCK, &sigs, NULL);
		235
		236	struct sigaction sa;
		237	sa.sa_handler = sighandler;
		238	sigfillset (&sa.sa_mask);
		239	sa.sa_flags = SA_RESTART;
		240
		241	if (sigaction (w.signum, &sa, 0))
64	{	242	{
		243	perror ("io_manager: error while installing signal handler, ignoring.");
65	pfs.clear ();	244	abort ();
66	iow.clear ();
67	}	245	}
68	else if (i == sz - 1)	246
		247	}
		248
		249	io_manager::reg (w, *sv);
		250	}
		251
		252	void io_manager::unreg (sig_watcher &w)
		253	{
		254	if (!w.active)
		255	return;
		256
		257	assert (0 < w.signum && w.signum <= sw.size ());
		258
		259	io_manager::unreg (w, *sw[w.signum - 1]);
		260	}
		261
		262	void sig_watcher::start (int signum)
		263	{
		264	stop ();
		265	this->signum = signum;
		266	io_manager::reg (*this);
		267	}
		268	#endif
		269
		270	void io_manager::loop ()
		271	{
		272	init::required ();
		273
		274	#if IOM_TIME
		275	set_now ();
		276	#endif
		277
		278	for (;;)
		279	{
		280
		281	#if IOM_TIME
		282	// call pending time watchers
		283	{
		284	bool activity;
		285
		286	do
		287	{
		288	activity = false;
		289
		290	for (int i = tw.size (); i--; )
		291	if (!tw[i])
		292	tw.erase_unordered (i);
		293	else if (tw[i]->at <= NOW)
		294	{
		295	time_watcher &w = *tw[i];
		296
		297	unreg (w);
		298	w.call (w);
		299
		300	activity = true;
		301	}
		302	}
		303	while (activity);
		304	}
		305	#endif
		306
		307	#if IOM_CHECK
		308	// call all check watchers
		309	for (int i = cw.size (); i--; )
		310	if (!cw[i])
		311	cw.erase_unordered (i);
		312	else
		313	cw[i]->call (*cw[i]);
		314	#endif
		315
		316	struct TIMEVAL *to = 0;
		317	struct TIMEVAL tval;
		318
		319	#if IOM_IDLE
		320	if (iw.size ())
69	{	321	{
70	iow.pop_back ();	322	tval.tv_sec = 0;
71	pfs.pop_back ();	323	tval.TV_FRAC = 0;
		324	to = &tval;
72	}	325	}
73	else	326	else
		327	#endif
74	{	328	{
75	iow[i] = iow[sz - 1]; iow.pop_back ();	329	#if IOM_TIME
76	pfs[i] = pfs[sz - 1]; pfs.pop_back ();	330	// find earliest active watcher
		331	time_watcher *next = tw[0]; // the first time-watcher must exist at ALL times
		332
		333	for (io_manager_vec<time_watcher>::const_iterator i = tw.end (); i-- > tw.begin (); )
		334	if (i && (i)->at < next->at)
		335	next = *i;
		336
		337	if (next->at > NOW && next != tw[0])
		338	{
		339	double diff = next->at - NOW;
		340	tval.tv_sec = (int)diff;
		341	tval.TV_FRAC = (int) ((diff - tval.tv_sec) * TV_MULT);
		342	to = &tval;
		343	}
77	}	344	}
78	}	345	#endif
79	}
80		346
81	void io_manager::reg (time_watcher *w)	347	#if IOM_IO \|\| IOM_SIG
82	{	348	fd_set rfd, wfd;
83	tw.push_back (w);
84	push_heap (tw.begin (), tw.end (), lowest_first);
85	}
86		349
87	void io_manager::unreg (time_watcher *w)	350	FD_ZERO (&rfd);
88	{	351	FD_ZERO (&wfd);
89	unsigned int sz = tw.size ();
90	unsigned int i = find (tw.begin (), tw.end (), w) - tw.begin ();
91		352
92	if (i != sz)	353	int fds = 0;
93	{	354
94	if (sz == 1)	355	# if IOM_IO
95	tw.clear ();	356	for (io_manager_vec<io_watcher>::const_iterator i = iow.end (); i-- > iow.begin (); )
96	else	357	if (*i)
		358	{
		359	if ((i)->events & EVENT_READ ) FD_SET ((i)->fd, &rfd);
		360	if ((i)->events & EVENT_WRITE) FD_SET ((i)->fd, &wfd);
		361
		362	if ((i)->fd >= fds) fds = (i)->fd + 1;
		363	}
		364	# endif
		365
		366	if (!to && !fds) //TODO: also check idle_watchers and check_watchers?
		367	break; // no events
		368
		369	# if IOM_SIG
		370	FD_SET (sigpipe[0], &rfd);
		371	if (sigpipe[0] >= fds) fds = sigpipe[0] + 1;
		372	# endif
		373
		374	# if IOM_SIG
		375	// there is no race, as we use a pipe for signals, so select
		376	// will return if a signal is caught.
		377	sigprocmask (SIG_UNBLOCK, &sigs, NULL);
		378	# endif
		379	fds = select (fds, &rfd, &wfd, NULL, to);
		380	# if IOM_SIG
		381	sigprocmask (SIG_BLOCK, &sigs, NULL);
		382	# endif
		383
		384	# if IOM_TIME
		385	set_now ();
		386	# endif
		387
		388	if (fds > 0)
97	{	389	{
98	if (i != sz - 1)	390	# if IOM_SIG
99	tw[i] = tw[sz - 1];	391	if (FD_ISSET (sigpipe[0], &rfd))
		392	{
		393	char ch;
100		394
101	tw.pop_back ();	395	while (read (sigpipe[0], &ch, 1) > 0)
102	make_heap (tw.begin (), tw.end (), lowest_first);	396	;
		397
		398	for (vector<sig_vec *>::iterator svp = sw.end (); svp-- > sw.begin (); )
		399	if (svp && (svp)->pending)
		400	{
		401	sig_vec &sv = **svp;
		402	for (int i = sv.size (); i--; )
		403	if (!sv[i])
		404	sv.erase_unordered (i);
		405	else
		406	sv[i]->call (*sv[i]);
		407
		408	sv.pending = false;
		409	}
		410	}
		411	# endif
		412
		413	# if IOM_IO
		414	for (int i = iow.size (); i--; )
		415	if (!iow[i])
		416	iow.erase_unordered (i);
		417	else
		418	{
		419	io_watcher &w = *iow[i];
		420	short revents = w.events;
		421
		422	if (!FD_ISSET (w.fd, &rfd)) revents &= ~EVENT_READ;
		423	if (!FD_ISSET (w.fd, &wfd)) revents &= ~EVENT_WRITE;
		424
		425	if (revents)
		426	w.call (w, revents);
		427	}
		428	#endif
103	}	429	}
104	}	430	else if (fds < 0 && errno != EINTR)
105	}
106
107	inline void set_now (void)
108	{
109	struct timeval tv;
110
111	gettimeofday (&tv, 0);
112
113	NOW = (tstamp)tv.tv_sec + (tstamp)tv.tv_usec / 1000000;
114	}
115
116	void io_manager::loop ()
117	{
118	set_now ();
119
120	while (!(iow.empty () && tw.empty ()))
121	{
122	int timeout = tw.empty ()
123	? 3600 * 1000 // wake up at least every hour
124	: (int) ((tw[0]->at - NOW) * 1000);
125
126	printf ("s%d t%d #%d\n", pfs.size (), timeout, tw.size ());
127
128	if (timeout >= 0)
129	{	431	{
130	int fds = poll (&pfs[0], pfs.size (), timeout);	432	perror ("io_manager: fatal error while waiting for I/O or time event, aborting.");
131
132	set_now ();	433	abort ();
133
134	for (unsigned int i = iow.size (); fds && i--; )
135	if (pfs[i].revents)
136	{
137	--fds;
138	iow[i]->call (pfs[i].revents);
139	}
140	}	434	}
		435	#if IOM_IDLE
		436	else
		437	for (int i = iw.size (); i--; )
		438	if (!iw[i])
		439	iw.erase_unordered (i);
		440	else
		441	iw[i]->call (*iw[i]);
		442	#endif
141		443
142	while (!tw.empty () && tw[0]->at <= NOW)	444	#elif IOM_TIME
143	{	445	if (!to)
144	pop_heap (tw.begin (), tw.end (), lowest_first);	446	break;
145	(*(tw.end () - 1))->trigger ();
146	push_heap (tw.begin (), tw.end (), lowest_first);
147	}
148	}
149	}
150		447
151	io_manager::io_manager ()	448	select (0, 0, 0, 0, &to);
152	{
153	set_now ();	449	set_now ();
		450	#else
		451	break;
		452	#endif
		453	}
154	}	454	}
155		455
156	io_manager::~io_manager ()
157	{
158	//
159	}
160

Diff Legend

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

Comparing gvpe/src/iom.C (file contents): Revision 1.3 by pcg, Fri Mar 21 21:21:02 2003 UTC vs. Revision 1.28 by pcg, Mon Jan 10 19:21:07 2005 UTC

Diff Legend

Comparing gvpe/src/iom.C (file contents):
Revision 1.3 by pcg, Fri Mar 21 21:21:02 2003 UTC vs.
Revision 1.28 by pcg, Mon Jan 10 19:21:07 2005 UTC