r19338 MediaWiki - Code Review archive

Repository:	MediaWiki
Revision:	< r19337‎ \| r19338 \| r19339 >
Date:	14:17, 16 January 2007
Author:	river
Status:	old
Tags:
Comment:	sendfile can work on solaris sendfile checking should be less platform-specific, was broken on non-linux platforms with sys/sendfile.h flowio.cc:79: warning: int format, pid_t arg (arg 4)
Modified paths:	/trunk/loreley/src/include/config.h (modified) (history) /trunk/loreley/src/include/flowio.h (modified) (history) /trunk/loreley/src/include/net.h (modified) (history) /trunk/loreley/src/loreley/Makefile.in (modified) (history) /trunk/loreley/src/loreley/flowio.cc (modified) (history) /trunk/loreley/src/loreley/net.cc (modified) (history) /trunk/loreley/src/loreley/net_solaris.cc (added) (history)

Diff [purge]

Index: trunk/loreley/src/include/flowio.h
—	—	@@ -150,7 +150,7 @@
151	151	_socket->clearbacks();
152	152	}
153	153
154		~~-#ifdef HAVE_SYS_SENDFILE_H~~
	154	+#ifdef DO_SENDFILE
155	155	virtual bool dio_supported(dio_source s) const {
156	156	return s == dio_source_fd;
157	157	}
Index: trunk/loreley/src/include/config.h
—	—	@@ -78,7 +78,7 @@
79	79	size_t max_entity_size;
80	80	string cache_master;
81	81	bool htcp_sigrequired;
82		~~- int backend_timeo;~~
	82	+ time_t backend_timeo;
83	83
84	84	vector<cachedir> cachedirs;
85	85	vector<pair<string, string> > stats_hosts;
Index: trunk/loreley/src/include/net.h
—	—	@@ -22,6 +22,11 @@
23	23
24	24	#include "loreley.h"
25	25
	26	+#if defined(HAVE_SENDFILE) && (defined(__linux__) \|\| (defined(__SVR4) && defined(__sun)))
	27	+# define DO_SENDFILE
	28	+# define SOLARIS_SENDFILE
	29	+#endif
	30	+
26	31	extern bool wnet_exit;
27	32	struct event_queue;
28	33
—	—	@@ -255,7 +260,7 @@
256	261	int write (char const *buf, size_t count) {
257	262	return ::write(_s, buf, count);
258	263	}
259		~~-#if defined(HAVE_SENDFILE) && defined(__linux__)~~
	264	+#ifdef DO_SENDFILE
260	265	int sendfile (int to, off_t *off, size_t n) {
261	266	return ::sendfile(_s, to, off, n);
262	267	}
Index: trunk/loreley/src/loreley/Makefile.in
—	—	@@ -32,6 +32,7 @@
33	33	ifname_to_address.cc \
34	34	log.cc \
35	35	net.cc \
	36	+ net_solaris.cc \
36	37	preprocessor.cc \
37	38	radix.cc \
38	39	loreley.cc \
Index: trunk/loreley/src/loreley/flowio.cc
—	—	@@ -75,8 +75,8 @@
76	76	return ret;
77	77	}
78	78
79		~~- snprintf(path, sizeof(path), "/dev/shm/loreley.diobuf.%d.%d.%d",~~
80		~~- getpid(), (int) pthread_self(), rand());~~
	79	+ snprintf(path, sizeof(path), "/dev/shm/loreley.diobuf.%lu.%lu.%d",
	80	+ (unsigned long) getpid(), (unsigned long) pthread_self(), rand());
81	81	if ((_diofd = open(path, O_CREAT \| O_EXCL \| O_RDWR, 0600)) == -1) {
82	82	wlog.warn(format("opening diobuf %s: %s")
83	83	% path % strerror(errno));
—	—	@@ -224,13 +224,17 @@
225	225	}
226	226	}
227	227
228		~~-#if defined(HAVE_SENDFILE) && defined(__linux__)~~
	228	+#if defined(DO_SENDFILE)
229	229	sink_result
230	230	socket_sink::dio_ready(int fd, off_t off, size_t len, ssize_t &discard)
231	231	{
232	232	ssize_t wrote;
233	233	WDEBUG(format("dio_ready: starting off %d") % off);
	234	+#ifdef SOLARIS_SENDFILE /* also linux */
234	235	if ((wrote = _socket->sendfile(fd, &off, len)) == -1) {
	236	+#else
	237	+# error shouldn't get here, no sendfile
	238	+#endif
235	239	if (errno == EAGAIN) {
236	240	_sink_spigot->sp_cork();
237	241	if (!_reg) {
Index: trunk/loreley/src/loreley/net_solaris.cc
—	—	@@ -0,0 +1,780 @@
	2	+/* Loreley: Lightweight HTTP reverse-proxy. */
	3	+/* net_solaris: Solaris-specific networking. */
	4	+/* Copyright (c) 2005, 2006 River Tarnell <river@attenuate.org>. */
	5	+/*
	6	+ * Permission is granted to anyone to use this software for any purpose,
	7	+ * including commercial applications, and to alter it and redistribute it
	8	+ * freely. This software is provided 'as-is', without any express or implied
	9	+ * warranty.
	10	+ */
	11	+
	12	+/* @(#) $Id$ */
	13	+
	14	+#include "stdinc.h"
	15	+#ifdef __INTEL_COMPILER
	16	+# pragma hdrstop
	17	+#endif
	18	+
	19	+#ifdef SOLARIS_IO
	20	+
	21	+#include <port.h>
	22	+
	23	+namespace sfun {
	24	+ using ::bind; /* because of conflict with boost::bind from util.h */
	25	+};
	26	+
	27	+using std::deque;
	28	+using std::signal;
	29	+using std::multimap;
	30	+
	31	+#include "loreley.h"
	32	+#include "net.h"
	33	+#include "config.h"
	34	+#include "log.h"
	35	+#include "http.h"
	36	+
	37	+#define RDBUF_INC 8192 /* buffer in 8 KiB incrs */
	38	+
	39	+/* see ifname_to_address.cc */
	40	+int ifname_to_address(int, sockaddr_in , char const );
	41	+unsigned int if_nametoindex_wrap(const char *);
	42	+
	43	+enum evtype_t {
	44	+ evtype_timer,
	45	+ evtype_event
	46	+};
	47	+
	48	+struct event_impl {
	49	+ void schedule(int64_t);
	50	+
	51	+ int64_t ev_when;
	52	+ event_queue *ev_queue;
	53	+ function<void (void)> ev_func;
	54	+};
	55	+
	56	+struct ev_pending {
	57	+ ev_pending(::event *ev, int64_t to, evtype_t type)
	58	+ : ep_event(ev)
	59	+ , ep_timeout(to)
	60	+ , ep_type(type) {}
	61	+
	62	+ ::event *ep_event;
	63	+ int64_t ep_timeout;
	64	+ evtype_t ep_type;
	65	+};
	66	+
	67	+pthread_t io_loop_thread;
	68	+
	69	+struct event_queue {
	70	+ event_queue(int p) : portfd(p) {}
	71	+ int portfd;
	72	+};
	73	+
	74	+tss<event_queue *> ev_queue;
	75	+
	76	+ioloop_t *ioloop;
	77	+
	78	+char current_time_str[30];
	79	+char current_time_short[30];
	80	+time_t current_time;
	81	+
	82	+static void secondly_sched(void);
	83	+
	84	+bool wnet_exit;
	85	+vector<wsocket *> awaks;
	86	+size_t cawak;
	87	+
	88	+void
	89	+wnet_add_accept_wakeup(wsocket *s)
	90	+{
	91	+ awaks.push_back(s);
	92	+}
	93	+
	94	+net::event secondly_ev;
	95	+
	96	+static void
	97	+secondly_update(void)
	98	+{
	99	+ WDEBUG("secondly_update");
	100	+ wnet_set_time();
	101	+ secondly_sched();
	102	+}
	103	+
	104	+static void
	105	+secondly_sched(void)
	106	+{
	107	+ secondly_ev.schedule(secondly_update, 1000);
	108	+}
	109	+
	110	+pthread_cond_t iot_ready;
	111	+pthread_mutex_t iot_ready_m;
	112	+
	113	+pthread_t io_thread;
	114	+
	115	+void *
	116	+io_start(void *)
	117	+{
	118	+#if 0
	119	+ signal_set(&ev_sigusr2, SIGUSR2, usr2_handler, NULL);
	120	+ signal_add(&ev_sigusr2, NULL);
	121	+#endif
	122	+ io_loop_thread = pthread_self();
	123	+ pthread_mutex_lock(&iot_ready_m);
	124	+ pthread_cond_signal(&iot_ready);
	125	+ pthread_mutex_unlock(&iot_ready_m);
	126	+
	127	+ ioloop->run();
	128	+ return NULL;
	129	+}
	130	+
	131	+ioloop_t::ioloop_t(void)
	132	+{
	133	+ prepare();
	134	+}
	135	+
	136	+void
	137	+ioloop_t::prepare(void)
	138	+{
	139	+size_t i;
	140	+
	141	+ pthread_mutex_init(&iot_ready_m, NULL);
	142	+ pthread_cond_init(&iot_ready, NULL);
	143	+
	144	+ pthread_mutex_lock(&iot_ready_m);
	145	+ pthread_create(&io_thread, NULL, io_start, NULL);
	146	+ pthread_cond_wait(&iot_ready, &iot_ready_m);
	147	+ pthread_mutex_unlock(&iot_ready_m);
	148	+
	149	+ wlog.notice(format("maximum number of open files: %d")
	150	+ % getdtablesize());
	151	+
	152	+ signal(SIGPIPE, SIG_IGN);
	153	+
	154	+ for (i = 0; i < listeners.size(); ++i) {
	155	+ listener *lns = listeners[i];
	156	+
	157	+ try {
	158	+ lns->sock->reuseaddr(true);
	159	+ lns->sock->bind();
	160	+ lns->sock->listen();
	161	+ } catch (socket_error &e) {
	162	+ wlog.error(format("creating listener %s: %s")
	163	+ % lns->sock->straddr() % e.what());
	164	+ exit(8);
	165	+ }
	166	+
	167	+ lns->sock->readback(bind(&ioloop_t::_accept, this, _1, _2), -1);
	168	+ }
	169	+ wlog.notice("net: initialised, using Solaris I/O");
	170	+ secondly_sched();
	171	+}
	172	+
	173	+void
	174	+ioloop_t::_accept(wsocket *s, int)
	175	+{
	176	+ wsocket *newe;
	177	+static rate_limited_logger enfile_log(60, ll_warn, "accept error: %s");
	178	+
	179	+ if ((newe = s->accept("HTTP client", prio_norm)) == NULL) {
	180	+ if (errno == ENFILE \|\| errno == EMFILE)
	181	+ enfile_log.log(strerror(errno));
	182	+ else
	183	+ wlog.warn(format("accept error: %s") % strerror(errno));
	184	+ s->readback(bind(&ioloop_t::_accept, this, _1, _2), -1);
	185	+ return;
	186	+ }
	187	+
	188	+ s->readback(bind(&ioloop_t::_accept, this, _1, _2), -1);
	189	+
	190	+ newe->nonblocking(true);
	191	+
	192	+ if (cawak == awaks.size())
	193	+ cawak = 0;
	194	+char buf[sizeof(wsocket ) 2];
	195	+ memcpy(buf, &newe, sizeof(newe));
	196	+ memcpy(buf + sizeof(newe), &s, sizeof(s));
	197	+ WDEBUG(format("_accept, lsnr=%d") % s);
	198	+
	199	+ if (awaks[cawak]->write(buf, sizeof(wsocket ) 2) < 0) {
	200	+ wlog.error(format("writing to thread wakeup socket: %s")
	201	+ % strerror(errno));
	202	+ exit(1);
	203	+ }
	204	+ cawak++;
	205	+ return;
	206	+}
	207	+
	208	+void
	209	+wnet_set_time(void)
	210	+{
	211	+struct tm *now;
	212	+ time_t old = current_time;
	213	+ size_t n;
	214	+
	215	+ current_time = time(NULL);
	216	+ if (current_time == old)
	217	+ return;
	218	+
	219	+ now = gmtime(&current_time);
	220	+
	221	+ n = strftime(current_time_str, sizeof(current_time_str), "%a, %d %b %Y %H:%M:%S GMT", now);
	222	+ assert(n);
	223	+ n = strftime(current_time_short, sizeof(current_time_short), "%Y-%m-%d %H:%M:%S", now);
	224	+ assert(n);
	225	+}
	226	+
	227	+namespace net {
	228	+
	229	+event::event(void)
	230	+{
	231	+ impl = new event_impl;
	232	+}
	233	+
	234	+event::~event(void)
	235	+{
	236	+ delete impl;
	237	+};
	238	+
	239	+void
	240	+event::schedule(function<void (void)> f, int64_t t)
	241	+{
	242	+ impl->ev_func = f;
	243	+ impl->schedule(t);
	244	+}
	245	+
	246	+#if 0
	247	+void
	248	+timer_callback(int, short, void *d)
	249	+{
	250	+event_impl ev = (event_impl )d;
	251	+ HOLDING(ev_lock);
	252	+ ev->ev_queue->add(NULL, ev, 0);
	253	+}
	254	+#endif
	255	+
	256	+void
	257	+socket::_register(int what, int64_t to, socket::call_type handler)
	258	+{
	259	+ _ev_flags = 0;
	260	+ _queue = (event_queue *)ev_queue;
	261	+
	262	+ WDEBUG(format("_register: %s%son %d (%s), queue %p")
	263	+ % ((what & FDE_READ) ? "read " : "")
	264	+ % ((what & FDE_WRITE) ? "write " : "")
	265	+ % _s % _desc % _queue);
	266	+
	267	+ if (what & FDE_READ)
	268	+ _read_handler = handler;
	269	+ if (what & FDE_WRITE)
	270	+ _write_handler = handler;
	271	+
	272	+ port_associate(_queue->portfd, PORT_SOURCE_FD, _s,
	273	+ (what & FDE_READ ? POLLRDNORM : 0) \|
	274	+ (what & FDE_WRITE ? POLLWRNORM : 0), this);
	275	+}
	276	+
	277	+address::address(void)
	278	+{
	279	+ memset(&_addr, 0, sizeof(_addr));
	280	+ _addrlen = 0;
	281	+ _fam = AF_UNSPEC;
	282	+ _stype = _prot = 0;
	283	+}
	284	+
	285	+address::address(sockaddr *sa, socklen_t len)
	286	+{
	287	+ memcpy(&_addr, sa, len);
	288	+ _addrlen = len;
	289	+ _stype = _prot = 0;
	290	+ _fam = ((sockaddr_storage *)sa)->ss_family;
	291	+}
	292	+
	293	+address::address(addrinfo *ai)
	294	+{
	295	+ memcpy(&_addr, ai->ai_addr, ai->ai_addrlen);
	296	+ _addrlen = ai->ai_addrlen;
	297	+ _fam = ai->ai_family;
	298	+ _stype = ai->ai_socktype;
	299	+ _prot = ai->ai_protocol;
	300	+}
	301	+
	302	+socket *
	303	+address::makesocket(char const *desc, sprio p) const
	304	+{
	305	+ return new socket(*this, desc, p);
	306	+}
	307	+
	308	+address::address(address const &o)
	309	+ : _addrlen(o._addrlen)
	310	+ , _fam(o._fam)
	311	+ , _stype(o._stype)
	312	+ , _prot(o._prot) {
	313	+ memcpy(&_addr, &o._addr, _addrlen);
	314	+}
	315	+
	316	+address &
	317	+address::operator= (address const &o)
	318	+{
	319	+ _addrlen = o._addrlen;
	320	+ _fam = o._fam;
	321	+ _stype = o._stype;
	322	+ _prot = o._prot;
	323	+ memcpy(&_addr, &o._addr, _addrlen);
	324	+ return *this;
	325	+}
	326	+
	327	+string const &
	328	+address::straddr(bool lng) const
	329	+{
	330	+char res[NI_MAXHOST];
	331	+int i;
	332	+ if (!lng) {
	333	+ if (_shortaddr.empty()) {
	334	+ if ((i = getnameinfo(sockaddr_cast<sockaddr const *>(&_addr),
	335	+ _addrlen, res, sizeof(res), NULL, 0, NI_NUMERICHOST)) != 0)
	336	+ throw resolution_error(i);
	337	+ _shortaddr = res;
	338	+ }
	339	+ return _shortaddr;
	340	+ }
	341	+
	342	+ if (_straddr.empty()) {
	343	+ char port[NI_MAXSERV];
	344	+ if ((i = getnameinfo(sockaddr_cast<sockaddr const *>(&_addr),
	345	+ _addrlen, res, sizeof(res), port, sizeof(port),
	346	+ NI_NUMERICHOST \| NI_NUMERICSERV)) != 0)
	347	+ throw resolution_error(i);
	348	+ _straddr = str(format("[%s]:%s") % res % port);
	349	+ }
	350	+ return _straddr;
	351	+}
	352	+
	353	+addrlist *
	354	+addrlist::resolve(string const &addr, string const &port,
	355	+ enum socktype socktype, int family)
	356	+{
	357	+addrinfo hints, res, ai;
	358	+int r;
	359	+ memset(&hints, 0, sizeof(hints));
	360	+ hints.ai_socktype = (int) socktype;
	361	+ if (family != AF_UNSPEC)
	362	+ hints.ai_family = family;
	363	+
	364	+ if ((r = getaddrinfo(addr.c_str(),
	365	+ port.c_str(), &hints, &res)) != 0)
	366	+ throw resolution_error(r);
	367	+
	368	+addrlist *al = new addrlist;
	369	+ for (ai = res; ai; ai = ai->ai_next)
	370	+ al->_addrs.push_back(address(ai));
	371	+
	372	+ freeaddrinfo(res);
	373	+ return al;
	374	+}
	375	+
	376	+address
	377	+addrlist::first(string const &addr, int port,
	378	+ enum socktype socktype, int family)
	379	+{
	380	+ return first(addr, lexical_cast<string>(port), socktype, family);
	381	+}
	382	+
	383	+addrlist *
	384	+addrlist::resolve(string const &addr, int port,
	385	+ enum socktype socktype, int family)
	386	+{
	387	+ return resolve(addr, lexical_cast<string>(port), socktype, family);
	388	+}
	389	+
	390	+address
	391	+addrlist::first(string const &addr, string const &port,
	392	+ enum socktype socktype, int family)
	393	+{
	394	+addrlist *r = addrlist::resolve(addr, port, socktype, family);
	395	+address res;
	396	+ res = *r->begin();
	397	+ delete r;
	398	+ return res;
	399	+}
	400	+
	401	+addrlist::~addrlist(void)
	402	+{
	403	+}
	404	+
	405	+addrlist::iterator
	406	+addrlist::begin(void) const
	407	+{
	408	+ return _addrs.begin();
	409	+}
	410	+
	411	+addrlist::iterator
	412	+addrlist::end(void) const
	413	+{
	414	+ return _addrs.end();
	415	+}
	416	+
	417	+socket *
	418	+addrlist::makesocket(char const *desc, sprio p) const
	419	+{
	420	+iterator it = _addrs.begin(), end = _addrs.end();
	421	+ for (; it != end; ++it) {
	422	+ socket *ns;
	423	+ if ((ns = it->makesocket(desc, p)) != NULL)
	424	+ return ns;
	425	+ }
	426	+ throw socket_error();
	427	+}
	428	+
	429	+socket *
	430	+socket::create(string const &addr, int port,
	431	+ enum socktype socktype, char const *desc, sprio p, int family)
	432	+{
	433	+ return create(addr, lexical_cast<string>(port), socktype, desc, p, family);
	434	+}
	435	+
	436	+socket *
	437	+socket::create(string const &addr, string const &port,
	438	+ enum socktype socktype, char const *desc, sprio p, int family)
	439	+{
	440	+addrlist *al = addrlist::resolve(addr, port, socktype, family);
	441	+ return al->makesocket(desc, p);
	442	+}
	443	+
	444	+pair<socket , socket >
	445	+socket::socketpair(enum socktype st)
	446	+{
	447	+socket s1 = NULL, s2 = NULL;
	448	+int sv[2];
	449	+ if (::socketpair(AF_UNIX, (int) st, 0, sv) == -1)
	450	+ throw socket_error();
	451	+ s1 = new socket(sv[0], net::address(), "socketpair", prio_norm);
	452	+ try {
	453	+ s2 = new socket(sv[1], net::address(), "socketpair", prio_norm);
	454	+ } catch (...) {
	455	+ delete s1;
	456	+ throw;
	457	+ }
	458	+ return make_pair(s1, s2);
	459	+}
	460	+
	461	+connect_status
	462	+socket::connect(void)
	463	+{
	464	+ if (::connect(_s, _addr.addr(), _addr.length()) == -1)
	465	+ if (errno == EINPROGRESS)
	466	+ return connect_later;
	467	+ else
	468	+ throw socket_error();
	469	+ return connect_okay;
	470	+}
	471	+
	472	+socket *
	473	+socket::accept(char const *desc, sprio p)
	474	+{
	475	+int ns;
	476	+sockaddr_storage addr;
	477	+socklen_t addrlen = sizeof(addr);
	478	+ if ((ns = ::accept(_s, (sockaddr *)&addr, &addrlen)) == -1)
	479	+ return NULL;
	480	+
	481	+ /*
	482	+ * If TCP_CORK is not supported, disable Nagle's algorithm on the
	483	+ * socket to prevent delays in HTTP keepalive (at the expense of
	484	+ * sending more packets).
	485	+ */
	486	+#if !defined(TCP_CORK) && defined(TCP_NODELAY)
	487	+int one = 1;
	488	+ setsockopt(ns, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
	489	+#endif
	490	+
	491	+ return new socket(ns, net::address((sockaddr *)&addr, addrlen), desc, p);
	492	+}
	493	+
	494	+int
	495	+socket::recvfrom(char *buf, size_t count, net::address &addr)
	496	+{
	497	+sockaddr_storage saddr;
	498	+socklen_t addrlen = sizeof(addr);
	499	+int i;
	500	+ if ((i = ::recvfrom(_s, buf, count, 0, (sockaddr *)&saddr, &addrlen)) < 0)
	501	+ return i;
	502	+ WDEBUG(format("recvfrom: fam=%d") % saddr.ss_family);
	503	+ addr = net::address((sockaddr *)&saddr, addrlen);
	504	+ return i;
	505	+}
	506	+
	507	+int
	508	+socket::sendto(char const *buf, size_t count, net::address const &addr)
	509	+{
	510	+ return ::sendto(_s, buf, count, 0, addr.addr(), addr.length());
	511	+}
	512	+
	513	+void
	514	+socket::nonblocking(bool v)
	515	+{
	516	+int val;
	517	+ val = fcntl(_s, F_GETFL, 0);
	518	+ if (val == -1)
	519	+ throw socket_error();
	520	+ if (v)
	521	+ val \|= O_NONBLOCK;
	522	+ else val &= ~O_NONBLOCK;
	523	+
	524	+ if (fcntl(_s, F_SETFL, val) == -1)
	525	+ throw socket_error();
	526	+}
	527	+
	528	+void
	529	+socket::reuseaddr(bool v)
	530	+{
	531	+int i = v;
	532	+int len = sizeof(i);
	533	+ setopt(SOL_SOCKET, SO_REUSEADDR, &i, len);
	534	+}
	535	+
	536	+void
	537	+socket::cork(void)
	538	+{
	539	+#ifdef TCP_CORK
	540	+int one = 1;
	541	+ setopt(IPPROTO_TCP, TCP_CORK, &one, sizeof(one));
	542	+#endif
	543	+}
	544	+
	545	+void
	546	+socket::uncork(void)
	547	+{
	548	+#ifdef TCP_CORK
	549	+int zero = 0;
	550	+ setopt(IPPROTO_TCP, TCP_CORK, &zero, sizeof(zero));
	551	+#endif
	552	+}
	553	+
	554	+int
	555	+socket::getopt(int level, int what, void addr, socklen_t len) const
	556	+{
	557	+int i;
	558	+ if ((i = getsockopt(_s, level, what, addr, len)) == -1)
	559	+ throw socket_error();
	560	+ return i;
	561	+}
	562	+
	563	+int
	564	+socket::setopt(int level, int what, void *addr, socklen_t len)
	565	+{
	566	+int i;
	567	+ if ((i = setsockopt(_s, level, what, addr, len)) == -1)
	568	+ throw socket_error();
	569	+ return i;
	570	+}
	571	+
	572	+int
	573	+socket::error(void) const
	574	+{
	575	+int error = 0;
	576	+socklen_t len = sizeof(error);
	577	+ try {
	578	+ getopt(SOL_SOCKET, SO_ERROR, &error, &len);
	579	+ return error;
	580	+ } catch (socket_error &) {
	581	+ return 0;
	582	+ }
	583	+}
	584	+
	585	+socket::socket(int s, net::address const &a, char const *desc, sprio p)
	586	+ : _addr(a)
	587	+ , _desc(desc)
	588	+ , _prio(p)
	589	+{
	590	+ _s = s;
	591	+}
	592	+
	593	+socket::socket(net::address const &a, char const *desc, sprio p)
	594	+ : _addr(a)
	595	+ , _desc(desc)
	596	+ , _prio(p)
	597	+{
	598	+ _s = ::socket(_addr.family(), _addr.socktype(), _addr.protocol());
	599	+ if (_s == -1)
	600	+ throw socket_error();
	601	+}
	602	+
	603	+void
	604	+socket::bind(void)
	605	+{
	606	+int one = 1;
	607	+ setopt(SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));
	608	+ if (sfun::bind(_s, _addr.addr(), _addr.length()) == -1)
	609	+ throw socket_error();
	610	+}
	611	+
	612	+void
	613	+socket::listen(int bl)
	614	+{
	615	+ if (::listen(_s, bl) == -1)
	616	+ throw socket_error();
	617	+}
	618	+
	619	+socket::~socket(void)
	620	+{
	621	+ WDEBUG("closing socket");
	622	+ port_dissociate(_queue->portfd, PORT_SOURCE_FD, _s);
	623	+ close(_s);
	624	+}
	625	+
	626	+void
	627	+socket::clearbacks(void)
	628	+{
	629	+ port_dissociate(_queue->portfd, PORT_SOURCE_FD, _s);
	630	+}
	631	+
	632	+void
	633	+socket::mcast_join(string const &ifname)
	634	+{
	635	+ switch (_addr.family()) {
	636	+ case AF_INET: {
	637	+ struct address ifaddr = address::from_ifname(_s, ifname);
	638	+ sockaddr_in inbind = (sockaddr_in )_addr.addr();
	639	+ sockaddr_in inif = (sockaddr_in )ifaddr.addr();
	640	+ ip_mreq mr;
	641	+ memset(&mr, 0, sizeof(mr));
	642	+ mr.imr_multiaddr.s_addr = inbind->sin_addr.s_addr;
	643	+ mr.imr_interface.s_addr = inif->sin_addr.s_addr;
	644	+ WDEBUG(format("NET: %s joins mcast on if %s")
	645	+ % straddr() % ifaddr.straddr());
	646	+ setopt(IPPROTO_IP, IP_ADD_MEMBERSHIP, &mr, sizeof(mr));
	647	+ break;
	648	+ }
	649	+
	650	+ case AF_INET6: {
	651	+#ifdef IPV6_ADD_MEMBERSHIP
	652	+ u_int ifindex = address::ifname_to_index(ifname);
	653	+ sockaddr_in6 inbind = (sockaddr_in6 )_addr.addr();
	654	+ ipv6_mreq mr;
	655	+ memset(&mr, 0, sizeof(mr));
	656	+ memcpy(&mr.ipv6mr_multiaddr, &inbind->sin6_addr,
	657	+ sizeof(mr.ipv6mr_multiaddr));
	658	+ mr.ipv6mr_interface = ifindex;
	659	+ setopt(IPPROTO_IPV6, IPV6_ADD_MEMBERSHIP, &mr, sizeof(mr));
	660	+#else
	661	+ wlog.warn("IPv6 multicast not supported on this platform");
	662	+#endif
	663	+ break;
	664	+ }
	665	+
	666	+ default:
	667	+ throw socket_error("multicast join not applicable for this socket type");
	668	+ }
	669	+}
	670	+
	671	+u_int
	672	+address::ifname_to_index(string const &ifname)
	673	+{
	674	+u_int ret = if_nametoindex_wrap(ifname.c_str());
	675	+ if (ret == 0)
	676	+ throw socket_error("named interface does not exist");
	677	+ return ret;
	678	+}
	679	+
	680	+address
	681	+address::from_ifname(int s, string const &ifname)
	682	+{
	683	+sockaddr_in addr;
	684	+ if (ifname_to_address(s, &addr, ifname.c_str()) < 0)
	685	+ throw socket_error();
	686	+address ret((sockaddr *)&addr, sizeof(sockaddr_in));
	687	+ return ret;
	688	+}
	689	+
	690	+} // namespace net
	691	+
	692	+void
	693	+make_event_base(void)
	694	+{
	695	+//static lockable meb_lock;
	696	+// if (evb == NULL) {
	697	+// HOLDING(meb_lock);
	698	+// evb = (event_base *)event_init();
	699	+// event_base_priority_init(evb, prio_max);
	700	+// signal_set(&ev_sigint, SIGINT, sig_exit, NULL);
	701	+// signal_add(&ev_sigint, NULL);
	702	+// signal_set(&ev_sigterm, SIGTERM, sig_exit, NULL);
	703	+// signal_add(&ev_sigterm, NULL);
	704	+// }
	705	+ io_loop_thread = pthread_self();
	706	+ ev_queue = new event_queue * (new event_queue(port_create()));
	707	+}
	708	+
	709	+void
	710	+sig_exit(int sig, short what, void *d)
	711	+{
	712	+ wnet_exit = true;
	713	+}
	714	+
	715	+void
	716	+ioloop_t::run(void)
	717	+{
	718	+ for (;;)
	719	+ sleep(INT_MAX);
	720	+#if 0
	721	+size_t i;
	722	+ for (i = 0; i < listeners.size(); ++i)
	723	+ delete listeners[i];
	724	+#endif
	725	+}
	726	+
	727	+void
	728	+ioloop_t::thread_run(void)
	729	+{
	730	+event_queue eq = ev_queue;
	731	+port_event_t ev;
	732	+net::socket *sk;
	733	+
	734	+ while (port_get(eq->portfd, &ev, NULL)) {
	735	+ WDEBUG(format("[%d] thread_run: waiting for event, eq=%p")
	736	+ % pthread_self() % eq);
	737	+
	738	+ switch (ev.portev_source) {
	739	+ case PORT_SOURCE_FD:
	740	+ sk = static_cast<net::socket *>(ev.portev_user);
	741	+ WDEBUG(format("[%d] thread_run: got event on %s")
	742	+ % pthread_self() % sk->_desc);
	743	+ if (ev.portev_events & POLLRDNORM)
	744	+ sk->_read_handler(sk, false);
	745	+ if (ev.portev_events & POLLWRNORM)
	746	+ sk->_write_handler(sk, false);
	747	+ break;
	748	+
	749	+#if 0
	750	+ if ((*it)->ee_flags & EV_TIMEOUT) {
	751	+ if ((*it)->ee_sock->_ev_flags & EV_READ) {
	752	+ (it)->ee_sock->_read_handler((it)->ee_sock, true);
	753	+ } else if ((*it)->ee_sock->_ev_flags & EV_WRITE) {
	754	+ (it)->ee_sock->_write_handler((it)->ee_sock, true);
	755	+ }
	756	+ }
	757	+#endif
	758	+#if 0
	759	+ WDEBUG("event thread");
	760	+ (*it)->ee_event->ev_func();
	761	+#endif
	762	+ }
	763	+ }
	764	+}
	765	+
	766	+
	767	+void
	768	+event_impl::schedule(int64_t when)
	769	+{
	770	+#if 0
	771	+ HOLDING(ev_lock);
	772	+ WDEBUG(format("schedule, when=%d") % when);
	773	+ this->ev_when = when;
	774	+ this->ev_queue = (event_queue *)::ev_queue;
	775	+ evtimer_set(&ev_event, timer_callback, this);
	776	+ ev_pending_list.insert(make_pair(0, ev_pending(&ev_event, ev_when, evtype_timer)));
	777	+ pthread_kill(io_loop_thread, SIGUSR2);
	778	+#endif
	779	+}
	780	+
	781	+#endif
Index: trunk/loreley/src/loreley/net.cc
—	—	@@ -15,6 +15,8 @@
16	16	# pragma hdrstop
17	17	#endif
18	18
	19	+#ifndef SOLARIS_IO
	20	+
19	21	namespace sfun {
20	22	using ::bind; /* because of conflict with boost::bind from util.h */
21	23	};
—	—	@@ -923,3 +925,5 @@
924	926	ev_pending_list.insert(make_pair(0, ev_pending(&ev_event, ev_when, evtype_timer)));
925	927	pthread_kill(io_loop_thread, SIGUSR2);
926	928	}
	929	+
	930	+#endif /* !SOLARIS_IO */