X-Git-Url: http://git.tremily.us/?a=blobdiff_plain;f=irkerd;h=5f90fdcdff0c775d63399eb3710d202cb44aad23;hb=a4fc71a6cf88963987efd7e2491ab64eeac11cd8;hp=ffa5733ca0007fe56587f044eb89dea41e3ea89c;hpb=cdc6dceb5f192ef4d21a7344d43a3b76d035b922;p=irker.git diff --git a/irkerd b/irkerd index ffa5733..5f90fdc 100755 --- a/irkerd +++ b/irkerd @@ -19,7 +19,7 @@ Design and code by Eric S. Raymond . See the project resource page at . Requires Python 2.6 or 2.5 with the simplejson library installed, and -the irc client library at version >= 2.0.2: see +the irc client library at version >= 3.4 which requires 2.6: see http://pypi.python.org/pypi/irc/ """ @@ -38,32 +38,13 @@ UNSEEN_TTL = 60 # Time to live, seconds since first request CHANNEL_MAX = 18 # Max channels open per socket (default) ANTI_FLOOD_DELAY = 0.5 # Anti-flood delay after transmissions, seconds ANTI_BUZZ_DELAY = 0.09 # Anti-buzz delay after queue-empty check +CONNECTION_MAX = 200 # To avoid hitting a thread limit # No user-serviceable parts below this line -version = "1.9" +version = "1.15" -# This black magic imports support for green threads (coroutines), -# then has kinky sex with the import library internals, replacing -# "threading" with a coroutine-using imposter. Threads then become -# ultra-light-weight and cooperatively scheduled. -try: - import eventlet - eventlet.monkey_patch() - green_threads = True - # With greenlets we don't worry about thread exhaustion, only the - # file descriptor limit (typically 1024 on modern Unixes). Thus we - # can handle a lot more concurrent sessions and generate less - # join/leave spam under heavy load. - CONNECTION_MAX = 1000 -except ImportError: - # Threads are more expensive if we have to use OS-level ones - # rather than greenlets. We need to avoid pushing thread limits - # as well as fd limits. See security.txt for discussion. - CONNECTION_MAX = 200 - green_threads = False - -import sys, getopt, urlparse, time, random, socket +import sys, getopt, urlparse, time, random, socket, signal import threading, Queue, SocketServer import irc.client, logging try: @@ -88,19 +69,23 @@ except ImportError: # even if the queue is nonempty but efforts to connect have failed for # a long time. # -# There are multiple threads. One accepts incoming traffic from all servers. -# Each Connection also has a consumer thread and a thread-safe message queue. -# The program main appends messages to queues as JSON requests are received; -# the consumer threads try to ship them to servers. When a socket write -# stalls, it only blocks an individual consumer thread; if it stalls long -# enough, the session will be timed out. +# There are multiple threads. One accepts incoming traffic from all +# servers. Each Connection also has a consumer thread and a +# thread-safe message queue. The program main appends messages to +# queues as JSON requests are received; the consumer threads try to +# ship them to servers. When a socket write stalls, it only blocks an +# individual consumer thread; if it stalls long enough, the session +# will be timed out. This solves the biggest problem with a +# single-threaded implementation, which is that you can't count on a +# single stalled write not hanging all other traffic - you're at the +# mercy of the length of the buffers in the TCP/IP layer. # # Message delivery is thus not reliable in the face of network stalls, # but this was considered acceptable because IRC (notoriously) has the # same problem - there is little point in reliable delivery to a relay # that is down or unreliable. # -# This code uses only NICK, JOIN, MODE, and PRIVMSG. It is strictly +# This code uses only NICK, JOIN, PART, MODE, and PRIVMSG. It is strictly # compliant to RFC1459, except for the interpretation and use of the # DEAF and CHANLIMIT and (obsolete) MAXCHANNELS features. CHANLIMIT # is as described in the Internet RFC draft @@ -194,7 +179,7 @@ class Connection: break elif xmit_timeout or ping_timeout: self.irker.debug(1, "timing out connection to %s at %s (ping_timeout=%s, xmit_timeout=%s)" % (self.servername, time.asctime(), ping_timeout, xmit_timeout)) - with self.irker.library_lock: + with self.irker.irc.mutex: self.connection.context = None self.connection.quit("transmission timeout") self.connection = None @@ -215,7 +200,7 @@ class Connection: break elif not self.connection: # Queue is nonempty but server isn't connected. - with self.irker.library_lock: + with self.irker.irc.mutex: self.connection = self.irker.irc.server() self.connection.context = self # Try to avoid colliding with other instances @@ -229,9 +214,12 @@ class Connection: nickname=self.nickname(), username="irker", ircname="irker relaying client") + if hasattr(self.connection, "buffer"): + self.connection.buffer.errors = 'replace' self.status = "handshaking" self.irker.debug(1, "XMIT_TTL bump (%s connection) at %s" % (self.servername, time.asctime())) self.last_xmit = time.time() + self.last_ping = time.time() except irc.client.ServerConnectionError: self.status = "disconnected" elif self.status == "handshaking": @@ -252,17 +240,16 @@ class Connection: self.status = "expired" break elif self.status == "ready": - with self.irker.library_lock: - (channel, message) = self.queue.get() - if channel not in self.channels_joined: - self.connection.join(channel) - self.irker.debug(1, "joining %s on %s." % (channel, self.servername)) - for segment in message.split("\n"): - self.connection.privmsg(channel, segment) - time.sleep(ANTI_FLOOD_DELAY) - self.last_xmit = self.channels_joined[channel] = time.time() - self.irker.debug(1, "XMIT_TTL bump (%s transmission) at %s" % (self.servername, time.asctime())) - self.queue.task_done() + (channel, message) = self.queue.get() + if channel not in self.channels_joined: + self.connection.join(channel) + self.irker.debug(1, "joining %s on %s." % (channel, self.servername)) + for segment in message.split("\n"): + self.connection.privmsg(channel, segment) + time.sleep(ANTI_FLOOD_DELAY) + self.last_xmit = self.channels_joined[channel] = time.time() + self.irker.debug(1, "XMIT_TTL bump (%s transmission) at %s" % (self.servername, time.asctime())) + self.queue.task_done() except: (exc_type, _exc_value, exc_traceback) = sys.exc_info() self.irker.logerr("exception %s in thread for %s" % \ @@ -271,6 +258,9 @@ class Connection: # when we need to be able to for debugging purposes. if debuglvl > 0: raise exc_type, _exc_value, exc_traceback + else: + # Maybe this should have its own status? + self.status = "expired" def live(self): "Should this connection not be scavenged?" return self.status != "expired" @@ -282,6 +272,9 @@ class Connection: if self.channel_limits: match_count = 0 for already in self.channels_joined: + # This obscure code is because the RFCs allow separate limits + # by channel type (indicated by the first character of the name) + # a feature that is almost never actually used. if already[0] == channel[0]: match_count += 1 return match_count < self.channel_limits.get(channel[0], CHANNEL_MAX) @@ -291,6 +284,8 @@ class Connection: class Target(): "Represent a transmission target." def __init__(self, url): + # Pre-2.6 Pythons don't recognize irc: as a valid URL prefix. + url = url.replace("irc://", "http://") parsed = urlparse.urlparse(url) irchost, _, ircport = parsed.netloc.partition(':') if not ircport: @@ -302,6 +297,9 @@ class Target(): # got kicked, and irkerd crashed because the server returned # "#channel" in the notification that our kick handler saw. self.channel = parsed.path.lstrip('/').lower() + # This deals with a tweak in recent versions of urlparse. + if parsed.fragment: + self.channel += "#" + parsed.fragment isnick = self.channel.endswith(",isnick") if isnick: self.channel = self.channel[:-7] @@ -375,8 +373,7 @@ class Irker: self.irc.add_global_handler("featurelist", self._handle_features) self.irc.add_global_handler("disconnect", self._handle_disconnect) self.irc.add_global_handler("kick", self._handle_kick) - self.library_lock = threading.Lock() - thread = threading.Thread(target=self._process_forever) + thread = threading.Thread(target=self.irc.process_forever) thread.setDaemon(True) self.irc._thread = thread thread.start() @@ -388,12 +385,6 @@ class Irker: "Debugging information." if self.debuglevel >= level: sys.stderr.write("irkerd: %s\n" % errmsg) - def _process_forever(self): - "IRC library process_forever with mutex." - self.debug(1, "process_forever()") - while True: - with self.library_lock: - self.irc.process_once(ANTI_BUZZ_DELAY) def _handle_ping(self, connection, _event): "PING arrived, bump the last-received time for the connection." if connection.context: @@ -410,7 +401,11 @@ class Irker: "Determine if and how we can set deaf mode." if connection.context: cxt = connection.context - for lump in event.arguments(): + arguments = event.arguments + # irclib 5.0 compatibility, because the maintainer is a fool + if callable(arguments): + arguments = arguments() + for lump in arguments: if lump.startswith("DEAF="): connection.mode(cxt.nickname(), "+"+lump[5:]) elif lump.startswith("MAXCHANNELS="): @@ -439,9 +434,14 @@ class Irker: connection.context.handle_disconnect() def _handle_kick(self, connection, event): "Server hung up the connection." - self.debug(1, "irker has been kicked from %s on %s" % (event.target(), connection.server)) + target = event.target + # irclib 5.0 compatibility, because the maintainer continues to be a + # fool. + if callable(target): + target = target() + self.debug(1, "irker has been kicked from %s on %s" % (target, connection.server)) if connection.context: - connection.context.handle_kick(event.target()) + connection.context.handle_kick(target) def handle(self, line): "Perform a JSON relay request." try: @@ -532,8 +532,7 @@ if __name__ == '__main__': server.setDaemon(True) server.start() try: - while True: - time.sleep(10) + signal.pause() except KeyboardInterrupt: raise SystemExit(1) except socket.error, e: