<commit_before>#!/usr/bin/env python
"""
Implementation of the minusconf protocol. See http://code.google.com/p/minusconf/ for details.
Apache License 2.0, see the LICENSE file for details.
"""

import struct
import socket
import threading
import time

_PORT = 6376
_ADDRESS_4 = '172.16.17.32'
_ADDRESS_6 = 'fd00:a516:7c1b:17cd:6d81:2137:bd2a:2c5b'
_ADDRESSES = [_ADDRESS_4]
_FALLBACK_ADDRS = ['::1', '127.0.0.1']
if socket.has_ipv6:
	_ADDRESSES.append(_ADDRESS_6)
_CHARSET = 'UTF-8'

try:
	if bytes != str: # Python 3+
		_compat_bytes = lambda bytestr: bytes(bytestr, 'charmap')
	else: # 2.6+
		_compat_bytes = str
except: # <2.6
	_compat_bytes = str

_MAGIC = _compat_bytes('\xad\xc3\xe6\xe7')
_OPCODE_QUERY = _compat_bytes('\x01')
_OPCODE_EMPTY = _compat_bytes('\x02')
_OPCODE_ADVERTISEMENT = _compat_bytes('\x65')
_OPCODE_ERROR = _compat_bytes('\x6f')
_STRING_TERMINATOR = _compat_bytes('\x00')

_TTL = None
_MAX_PACKET_SIZE = 2048 # Biggest packet size this implementation will accept"""
_SEEKER_TIMEOUT = 2.0 # Timeout for seeks in s

class MinusconfError(Exception):
	def __init__(self, msg=''):
		super(MinusconfError, self).__init__()
		self.msg = msg
	
	def send(self, sock, to):
		_send_packet(sock, to, _OPCODE_ERROR, _encode_string(self.msg))

class _ImmutableStruct(object):
	""" Helper structure for immutable objects """
	
	def __setattr__(self, *args):
		raise TypeError("This structure is immutable")
	__delattr__ = __setattr__
	
	def __init__(self, values):
		for (k,v) in values.items():
			super(_ImmutableStruct, self).__setattr__(k, v)
	
	def __eq__(self, other):
		return self.__dict__ == other.__dict__
	
	def __ne__(self, other):
		return self.__dict__ != other.__dict__
	
	def __lt__(self, other):
		return self.__dict__ < other.__dict__
	
	def __le__(self, other):
		return self.__dict__ <= other.__dict__
	
	def __gt__(self, other):
		return self.__dict__ > other.__dict__
	
	def __ge__(self, other):
		return self.__dict__ >= other.__dict__
	
	def __hash__(self):
		return hash(sum((hash(i) for i in self.__dict__.items())))

class Service(_ImmutableStruct):
	""" Helper structure for a service."""
	
	def __init__(self, stype, port, sname='', location=''):
		super(Service, self).__init__({'stype': stype, 'port':port, 'sname': sname, 'location':location})
	
	def matches_query(self, stype, sname):
		return _string_match(stype, self.stype) and _string_match(sname, self.sname)
	
	def __str__(self):
		res = self.stype + ' service at '
		if self.sname != '': res += self.sname + ' '
		res += self.location + ':' + self.port
		
		return res
	
	def __repr__(self):
		return ('Service(' +
			repr(self.stype) + ', ' +
			repr(self.port) + ', ' +
			repr(self.sname) + ', ' +
			repr(self.location) + ')')

class ServiceAt(_ImmutableStruct):
	""" A service returned by an advertiser"""
	
	def __init__(self, aname, stype, sname, location, port, addr):
		super(ServiceAt, self).__init__(
			{'aname': aname, 'stype': stype, 'sname': sname, 'location': location, 'port': port, 'addr':addr}
		)
	
	def matches_query_at(self, aname, stype, sname):
		return _string_match(stype, self.stype) and _string_match(sname, self.sname) and _string_match(aname, self.aname)
	
	@property
	def effective_location(self):
		return self.location if self.location != "" else self.addr
	
	def __str__(self):
		return (
			self.stype + ' service at ' +
			((self.sname + ' ') if self.sname != '' else '') +
			self.location + ':' + self.port +
			' (advertiser "' + self.aname + '" at ' + self.addr + ')'
			)
	
	def __repr__(self):
		return ('ServiceAt(' +
			repr(self.aname) + ', ' +
			repr(self.stype) + ', ' +
			repr(self.sname) + ', ' +
			repr(self.location) + ', ' +
			repr(self.port) + ', ' +
			repr(self.addr) + ')')

class Advertiser(object):
	""" Generic implementation of a -conf advertiser. You will probably want to use one of the subclasses.
	If ignore_unavailable is set, unsupported addresses (typically IPv6) are silently ignored
	"""
	
	def __init__(self, services=[], aname=None, ignore_unavailable=True):
		super(Advertiser, self).__init__()
		
		self.services = services
		self.aname = aname if aname != None else socket.gethostname()
		self.port = _PORT
		self.addresses = _ADDRESSES
		self.ignore_unavailable = ignore_unavailable
	
	def run(self):
		sock = self._init_sock()
		
		while True:
			self._read_and_handle(sock)
	
	def _init_sock(self):
		sock = _find_sock()
		sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
		sock.bind(('', self.port))
		
		addrs = _resolve_addrs(self.addresses, None, self.ignore_unavailable, (sock.family,))
		
		for fam,to,orig_fam,orig_addr in addrs:
			try:
				_multicast_join_group(sock, orig_fam, orig_addr)
			except socket.error:
				if not self.ignore_unavailable:
					raise
		
		return sock
	
	def _read_and_handle(self, sock):
		try:
			opcode, data, sender = _parse_packet(sock)
			
			if opcode == _OPCODE_QUERY:
				self.__handle_query(sock, sender, data)
			elif opcode == _OPCODE_EMPTY:
				#print("handling empty packet")
				pass
			else:
				MinusconfError('Invalid or unsupported opcode ' + struct.unpack('!B', opcode)).send(sock, sender)
		# Comment the following lines out for proper error handling (Python 2.6+)
		#except MinusconfError as mce:
		#	mce.send(sock, sender)
		#except ValueError as ve:
		#	MinusconfError(str(ve)).send(sock, sender)
		# Silent error handling
		except (MinusconfError, ValueError):
			pass
	
	def services_matching(self, stype, sname):
		return filter(lambda svc: svc.matches_query(stype, sname), self.services)
	
	def __handle_query(self, sock, sender, qrydata):
		qaname,p = _decode_string(qrydata, 0)
		qstype,p = _decode_string(qrydata, p)
		qsname,p = _decode_string(qrydata, p)
		
		if _string_match(qaname, self.aname):
			for svc in self.services_matching(qstype, qsname):
				rply = (
					_encode_string(self.aname) +
					_encode_string(svc.stype) +
					_encode_string(svc.sname) +
					_encode_string(svc.location) +
					_encode_string(svc.port)
					)
				
				_send_packet(sock, sender, _OPCODE_ADVERTISEMENT, rply)

class ConcurrentAdvertiser(Advertiser):
	def __init__(self, services=[], aname=None, ignore_unavailable=True):
		super(ConcurrentAdvertiser, self).__init__(services, aname, ignore_unavailable)
		
		# _caev_started, _caev_should_stop, and _caev_stopped have to be set to events by subclasses
	
	def start_blocking(self):
		""" Start the advertiser in a new thread, but wait until it is ready """
		
		self._caev_started.clear()
		self.start()
		self._caev_started.wait()
	
	def run(self):
		self._caev_should_stop.clear()
		self._caev_stopped.clear()
		
		try:
			sock = self._init_sock()
		finally:
			self._caev_started.set()
		
		while not self._caev_should_stop.is_set():
			self._read_and_handle(sock)
		
		self._caev_stopped.set()
	
	def stop(self):
		self._caev_should_stop.set()
		sock = _find_sock()
		localhost = '::1' if sock.family == socket.AF_INET6 else '127.0.0.1'
		sock.sendto(_MAGIC + _OPCODE_EMPTY + b'stop', 0, (localhost, self.port))
	
	def stop_blocking(self):
		self.stop()
		self._caev_stopped.wait()

class ThreadAdvertiser(ConcurrentAdvertiser, threading.Thread):
	""" Run the advertiser in a separate thread.
	
	This is what most applications want, but must not be used multiple times on the same port lest the OS can't distinguish which socket should receive messages.
	"""
	
	def __init__(self, services=[], aname=None, ignore_unavailable=True, daemon=True):
		ConcurrentAdvertiser.__init__(self, services, aname, ignore_unavailable)
		threading.Thread.__init__(self)
		
		self.setDaemon(daemon)
		
		self._caev_started = self._createEvent()
		self._caev_should_stop = self._createEvent()
		self._caev_stopped = self._createEvent()
	
	@staticmethod
	def _createEvent():
		res = threading.Event()
		
		if not hasattr(res, 'is_set'): # Python<2.6
			res.is_set = res.isSet
		
		return res

try:
	import multiprocessing
	
	class MultiprocessingAdvertiser(ConcurrentAdvertiser, multiprocessing.Process):
		"""
		multiprocessing is only available for Python 2.6+.
		See http://code.google.com/p/python-multiprocessing/ for a backport.
		"""
		def __init__(self, services=[], aname=None, ignore_unavailable=True, daemon=True):
			ConcurrentAdvertiser.__init__(self, services, aname, ignore_unavailable)
			multiprocessing.Process.__init__(self)
			
			self.daemon = daemon
			
			manager = multiprocessing.Manager()
			self._caev_started = manager.Event()
			self._caev_should_stop = manager.Event()
			self._caev_stopped = manager.Event()
			
			self.services = manager.list(services)
except ImportError:
	pass


class Seeker(threading.Thread):
	""" find_callback is called with (this_seeker,found_service_at)
	error_callback is called with (this seeker, sender, error message) """
	def __init__(self, servicetype='', advertisername='', servicename='', timeout=_SEEKER_TIMEOUT, port=_PORT, addresses=_ADDRESSES, fallback_addrs=_FALLBACK_ADDRS, find_callback=None, error_callback=None, daemonized=True, ignore_senderrors=True):
		super(Seeker, self).__init__()
		
		self.timeout = timeout
		self.port = port
		self.addresses = addresses
		self.fallback_addrs = fallback_addrs
		self.find_callback = find_callback
		self.error_callback = error_callback
		self.setDaemon(daemonized)
		self.ignore_senderrors = ignore_senderrors
		self.reset(servicetype, advertisername, servicename)
	
	def reset(self, servicetype='', advertisername='', servicename=''):
		self.servicetype = servicetype
		self.advertisername = advertisername
		self.servicename = servicename
		
		self.results = set()<commit_msg>Fix multiprocessing events for python<2.6<commit_after>#!/usr/bin/env python
"""
Implementation of the minusconf protocol. See http://code.google.com/p/minusconf/ for details.
Apache License 2.0, see the LICENSE file for details.
"""

import struct
import socket
import threading
import time

_PORT = 6376
_ADDRESS_4 = '172.16.17.32'
_ADDRESS_6 = 'fd00:a516:7c1b:17cd:6d81:2137:bd2a:2c5b'
_ADDRESSES = [_ADDRESS_4]
_FALLBACK_ADDRS = ['::1', '127.0.0.1']
if socket.has_ipv6:
	_ADDRESSES.append(_ADDRESS_6)
_CHARSET = 'UTF-8'

try:
	if bytes != str: # Python 3+
		_compat_bytes = lambda bytestr: bytes(bytestr, 'charmap')
	else: # 2.6+
		_compat_bytes = str
except: # <2.6
	_compat_bytes = str

_MAGIC = _compat_bytes('\xad\xc3\xe6\xe7')
_OPCODE_QUERY = _compat_bytes('\x01')
_OPCODE_EMPTY = _compat_bytes('\x02')
_OPCODE_ADVERTISEMENT = _compat_bytes('\x65')
_OPCODE_ERROR = _compat_bytes('\x6f')
_STRING_TERMINATOR = _compat_bytes('\x00')

_TTL = None
_MAX_PACKET_SIZE = 2048 # Biggest packet size this implementation will accept"""
_SEEKER_TIMEOUT = 2.0 # Timeout for seeks in s

class MinusconfError(Exception):
	def __init__(self, msg=''):
		super(MinusconfError, self).__init__()
		self.msg = msg
	
	def send(self, sock, to):
		_send_packet(sock, to, _OPCODE_ERROR, _encode_string(self.msg))

class _ImmutableStruct(object):
	""" Helper structure for immutable objects """
	
	def __setattr__(self, *args):
		raise TypeError("This structure is immutable")
	__delattr__ = __setattr__
	
	def __init__(self, values):
		for (k,v) in values.items():
			super(_ImmutableStruct, self).__setattr__(k, v)
	
	def __eq__(self, other):
		return self.__dict__ == other.__dict__
	
	def __ne__(self, other):
		return self.__dict__ != other.__dict__
	
	def __lt__(self, other):
		return self.__dict__ < other.__dict__
	
	def __le__(self, other):
		return self.__dict__ <= other.__dict__
	
	def __gt__(self, other):
		return self.__dict__ > other.__dict__
	
	def __ge__(self, other):
		return self.__dict__ >= other.__dict__
	
	def __hash__(self):
		return hash(sum((hash(i) for i in self.__dict__.items())))

class Service(_ImmutableStruct):
	""" Helper structure for a service."""
	
	def __init__(self, stype, port, sname='', location=''):
		super(Service, self).__init__({'stype': stype, 'port':port, 'sname': sname, 'location':location})
	
	def matches_query(self, stype, sname):
		return _string_match(stype, self.stype) and _string_match(sname, self.sname)
	
	def __str__(self):
		res = self.stype + ' service at '
		if self.sname != '': res += self.sname + ' '
		res += self.location + ':' + self.port
		
		return res
	
	def __repr__(self):
		return ('Service(' +
			repr(self.stype) + ', ' +
			repr(self.port) + ', ' +
			repr(self.sname) + ', ' +
			repr(self.location) + ')')

class ServiceAt(_ImmutableStruct):
	""" A service returned by an advertiser"""
	
	def __init__(self, aname, stype, sname, location, port, addr):
		super(ServiceAt, self).__init__(
			{'aname': aname, 'stype': stype, 'sname': sname, 'location': location, 'port': port, 'addr':addr}
		)
	
	def matches_query_at(self, aname, stype, sname):
		return _string_match(stype, self.stype) and _string_match(sname, self.sname) and _string_match(aname, self.aname)
	
	@property
	def effective_location(self):
		return self.location if self.location != "" else self.addr
	
	def __str__(self):
		return (
			self.stype + ' service at ' +
			((self.sname + ' ') if self.sname != '' else '') +
			self.location + ':' + self.port +
			' (advertiser "' + self.aname + '" at ' + self.addr + ')'
			)
	
	def __repr__(self):
		return ('ServiceAt(' +
			repr(self.aname) + ', ' +
			repr(self.stype) + ', ' +
			repr(self.sname) + ', ' +
			repr(self.location) + ', ' +
			repr(self.port) + ', ' +
			repr(self.addr) + ')')

class Advertiser(object):
	""" Generic implementation of a -conf advertiser. You will probably want to use one of the subclasses.
	If ignore_unavailable is set, unsupported addresses (typically IPv6) are silently ignored
	"""
	
	def __init__(self, services=[], aname=None, ignore_unavailable=True):
		super(Advertiser, self).__init__()
		
		self.services = services
		self.aname = aname if aname != None else socket.gethostname()
		self.port = _PORT
		self.addresses = _ADDRESSES
		self.ignore_unavailable = ignore_unavailable
	
	def run(self):
		sock = self._init_sock()
		
		while True:
			self._read_and_handle(sock)
	
	def _init_sock(self):
		sock = _find_sock()
		sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
		sock.bind(('', self.port))
		
		addrs = _resolve_addrs(self.addresses, None, self.ignore_unavailable, (sock.family,))
		
		for fam,to,orig_fam,orig_addr in addrs:
			try:
				_multicast_join_group(sock, orig_fam, orig_addr)
			except socket.error:
				if not self.ignore_unavailable:
					raise
		
		return sock
	
	def _read_and_handle(self, sock):
		try:
			opcode, data, sender = _parse_packet(sock)
			
			if opcode == _OPCODE_QUERY:
				self.__handle_query(sock, sender, data)
			elif opcode == _OPCODE_EMPTY:
				#print("handling empty packet")
				pass
			else:
				MinusconfError('Invalid or unsupported opcode ' + struct.unpack('!B', opcode)).send(sock, sender)
		# Comment the following lines out for proper error handling (Python 2.6+)
		#except MinusconfError as mce:
		#	mce.send(sock, sender)
		#except ValueError as ve:
		#	MinusconfError(str(ve)).send(sock, sender)
		# Silent error handling
		except (MinusconfError, ValueError):
			pass
	
	def services_matching(self, stype, sname):
		return filter(lambda svc: svc.matches_query(stype, sname), self.services)
	
	def __handle_query(self, sock, sender, qrydata):
		qaname,p = _decode_string(qrydata, 0)
		qstype,p = _decode_string(qrydata, p)
		qsname,p = _decode_string(qrydata, p)
		
		if _string_match(qaname, self.aname):
			for svc in self.services_matching(qstype, qsname):
				rply = (
					_encode_string(self.aname) +
					_encode_string(svc.stype) +
					_encode_string(svc.sname) +
					_encode_string(svc.location) +
					_encode_string(svc.port)
					)
				
				_send_packet(sock, sender, _OPCODE_ADVERTISEMENT, rply)

class ConcurrentAdvertiser(Advertiser):
	def __init__(self, services=[], aname=None, ignore_unavailable=True):
		super(ConcurrentAdvertiser, self).__init__(services, aname, ignore_unavailable)
		
		# _cav_started, _cav_should_stop, and _cav_stopped have to be set to events by subclasses
	
	def start_blocking(self):
		""" Start the advertiser in a new thread, but wait until it is ready """
		
		self._cav_started.clear()
		self.start()
		self._cav_started.wait()
	
	def run(self):
		self._cav_should_stop.clear()
		self._cav_stopped.clear()
		
		try:
			sock = self._init_sock()
		finally:
			self._cav_started.set()
		
		while not self._cav_should_stop.is_set():
			self._read_and_handle(sock)
		
		self._cav_stopped.set()
	
	def stop(self):
		self._cav_should_stop.set()
		sock = _find_sock()
		localhost = '::1' if sock.family == socket.AF_INET6 else '127.0.0.1'
		_send_packet(sock, (localhost, self.port), _OPCODE_EMPTY, _encode_string('stop'))
	
	def stop_blocking(self):
		self.stop()
		self._cav_stopped.wait()

class ThreadAdvertiser(ConcurrentAdvertiser, threading.Thread):
	""" Run the advertiser in a separate thread.
	
	This is what most applications want, but must not be used multiple times on the same port lest the OS can't distinguish which socket should receive messages.
	"""
	
	def __init__(self, services=[], aname=None, ignore_unavailable=True, daemon=True):
		ConcurrentAdvertiser.__init__(self, services, aname, ignore_unavailable)
		threading.Thread.__init__(self)
		
		self.setDaemon(daemon)
		
		self._cav_started = self._createEvent()
		self._cav_should_stop = self._createEvent()
		self._cav_stopped = self._createEvent()
	
	@staticmethod
	def _createEvent():
		res = threading.Event()
		
		if not hasattr(res, 'is_set'): # Python<2.6
			res.is_set = res.isSet
		
		return res

try:
	import multiprocessing
	
	class MultiprocessingAdvertiser(ConcurrentAdvertiser, multiprocessing.Process):
		"""
		multiprocessing is only available for Python 2.6+.
		See http://code.google.com/p/python-multiprocessing/ for a backport.
		"""
		def __init__(self, services=[], aname=None, ignore_unavailable=True, daemon=True):
			ConcurrentAdvertiser.__init__(self, services, aname, ignore_unavailable)
			multiprocessing.Process.__init__(self)
			
			self.daemon = daemon
			self._mpa_manager = multiprocessing.Manager()
			
			self._cav_started = multiprocessing.Event()
			self._cav_should_stop = multiprocessing.Event()
			self._cav_stopped = multiprocessing.Event()
			
			self.services = self._mpa_manager.list(services)
	
except ImportError:
	pass


class Seeker(threading.Thread):
	""" find_callback is called with (this_seeker,found_service_at)
	error_callback is called with (this seeker, sender, error message) """
	def __init__(self, servicetype='', advertisername='', servicename='', timeout=_SEEKER_TIMEOUT, port=_PORT, addresses=_ADDRESSES, fallback_addrs=_FALLBACK_ADDRS, find_callback=None, error_callback=None, daemonized=True, ignore_senderrors=True):
		super(Seeker, self).__init__()
		
		self.timeout = timeout
		self.port = port
		self.addresses = addresses
		self.fallback_addrs = fallback_addrs
		self.find_callback = find_callback
		self.error_callback = error_callback
		self.setDaemon(daemonized)
		self.ignore_senderrors = ignore_senderrors
		self.reset(servicetype, advertisername, servicename)
	
	def reset(self, servicetype='', advertisername='', servicename=''):
		self.servicetype = servicetype
		self.advertisername = advertisername
		self.servicename = servicename
		
		self.results = set()
{'fraction_non_alphanumeric': 0.12577373211963588, 'fraction_numerical': 0.009206762028608583, 'mean_word_length': 4.499427917620137, 'pattern_counts': {'":': 0, '<': 12, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 6, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1141649', 'n_tokens_mistral': 7566, 'n_tokens_neox': 6673, 'n_words': 1508}