<commit_before>from __future__ import absolute_import
import binascii
import contextlib
import logging
import os
import struct
import types
import weakref
import datetime
from collections import namedtuple


_use_gevent = False
def use_gevent():
	_use_gevent = True

class _sched_proxy(object):
	def __getattr__(self, name):
		_sched = None
		if _use_gevent:
			_sched = __import__("sched_gevent", globals(), level=1)
		else:
			_sched = __import__("sched_basic", globals(), level=1)
		if name in "subprocess socket Queue Lock Event spawn serve_forever".split():
			return getattr(_sched, name)
		raise AttributeError("No such attribute")

sched = _sched_proxy()

def default_wrapper(func):
	def wrap(*args, **kwargs):
		socket = sched.socket
		try:
			return func(*args, **kwargs)
		except socket.timeout:
			return None
		except socket.error as e:
			if e.errno in (os.errno.EAGAIN, os.errno.ECONNRESET, os.errno.EBADF):
				return b""
			elif e.errno in (os.errno.EINTR,):
				return None
			raise
		except KeyboardInterrupt:
			return b""
	return wrap


class ReadWrapper(object):
	def __init__(self, channel, read_wrap):
		self.channel = channel
		self.read_wrap = read_wrap
	
	def __enter__(self):
		self.installed_wrapper = self.channel.read_wrap
		self.channel.read_wrap = self
		return self.channel
	
	def __exit__(self, *args, **kwargs):
		self.channel.read_wrap = self.installed_wrapper
	
	def __call__(self, func):
		def wrap(*args, **kwargs):
			if self.channel.closed:
				return b""
			return self.read_wrap(func)(*args, **kwargs)
		return wrap


class Channel(object):
	'''
	Openflow abstract connection class
	
	This is not only for TCP but also for UDP.
	This is the reason that the name is not "Connection" but "Channel".
	You can subclass this to have instance members, of which lifecycle is 
	the same with channel.
	'''
	def __init__(self, *args, **kwargs):
		self._socket = kwargs.pop("socket", None) # dedicated socket
		self._sendto = kwargs.pop("sendto", None) # only if channel prefers sendto()
		self.reader = kwargs.pop("reader", None)
		self.read_wrap = kwargs.pop("read_wrap", default_wrapper)
		self.remote_address = kwargs.pop("remote_address", None)
		self.local_address = kwargs.pop("local_address", None)
		if self._socket:
			if self.remote_address is None:
				self.remote_address = self._socket.getpeername()
			if self.local_address is None:
				self.local_address = self._socket.getsockname()
			if hasattr(self._socket, "settimeout") and self._socket.gettimeout() == None:
				self._socket.settimeout(0.5)
	
	def attach(self, stream, **kwargs):
		self._socket = stream
		if hasattr(self._socket, "settimeout") and self._socket.gettimeout() == None:
			self._socket.settimeout(0.5)
		self.remote_address = stream.getpeername()
		self.local_address = stream.getsockname()
	
	@property
	def closed(self):
		# This is not self._socket.closed because in some use cases, 
		# self._socket is not available, for example with gevent.server.DatagramServer
		return self.remote_address is None
	
	def close(self):
		if self._socket:
			self._socket.close()
		
		if self.remote_address is not None:
			self.remote_address = None
	
	def send(self, message, **kwargs):
		if self._sendto:
			self._sendto(message, self.remote_address)
		elif self._socket:
			self._socket.send(message)
		else:
			raise ValueError("socket or sendto is required")
	
	def _recv(self, num):
		if self.reader:
			reader = self.reader
		else:
			reader = self._socket.recv
		return ReadWrapper(self, self.read_wrap)(reader)(num)


class Error(Exception):
	pass


class ChannelClose(Error):
	pass


class OpenflowBaseChannel(Channel):
	version = None # The negotiated version
	accept_versions = [4,] # defaults to openflow 1.3
	
	def __init__(self, *args, **kwargs):
		super(OpenflowBaseChannel, self).__init__(*args, **kwargs)
		self.buffer = b""
	
	def __iter__(self):
		while True:
			ret = self.recv()
			if ret:
				yield ret
			else:
				break
	
	def recv(self):
		required_len = 8
		while len(self.buffer) < required_len:
			tmp = super(OpenflowBaseChannel, self)._recv(8192)
			if tmp is None:
				continue
			elif len(tmp)==0:
				return tmp
			self.buffer += tmp
		
		p = struct.unpack_from("!BBHI", self.buffer)
		required_len = p[2]
		
		while len(self.buffer) < required_len:
			tmp = super(OpenflowBaseChannel, self)._recv(8192)
			if tmp is None:
				continue
			elif len(tmp)==0:
				return tmp
			self.buffer += tmp
		
		ret = self.buffer[0:required_len]
		self.buffer = self.buffer[required_len:]
		return ret


class LoggingChannel(OpenflowBaseChannel):
	channel_log_name = "channel"
	send_log_name = "send"
	recv_log_name = "recv"
	remote = ""
	
	def __init__(self, *args, **kwargs):
		super(LoggingChannel, self).__init__(*args, **kwargs)
		if self.remote_address:
			self.remote = " from %s" % self.remote_address[0]
		logging.getLogger(self.channel_log_name).info("%s connect%s" % (self, self.remote))
	
	def send(self, message, **kwargs):
		logging.getLogger(self.send_log_name).debug("%s %s" % (self, binascii.b2a_hex(message)))
		return super(LoggingChannel, self).send(message, **kwargs)
	
	def recv(self):
		message = super(LoggingChannel, self).recv()
		if message: # ignore b"" and None
			logging.getLogger(self.recv_log_name).debug("%s %s" % (self, binascii.b2a_hex(message)))
		return message
	
	def close(self):
		if not self.closed:
			super(LoggingChannel, self).close()
			logging.getLogger(self.channel_log_name).info("%s close%s" % (self, self.remote))


class OpenflowChannel(OpenflowBaseChannel):
	_start = None
	
	def attach(self, stream, **kwargs):
		super(OpenflowBaseChannel, self).attach(stream, **kwargs)
		if kwargs.get("autostart", True):
			self.start()
	
	def start(self):
		if self._start is None:
			self.send(hello(self.accept_versions))
			self._start = True
	
	def recv(self):
		message = super(OpenflowChannel, self).recv()
		if message:
			(version, oftype, length, xid) = parse_ofp_header(message)
			if oftype==0: # HELLO
				accept_versions = ofp_version_normalize(self.accept_versions)
				if not accept_versions:
					accept_versions = set([1,])
				cross_versions = parse_hello(message) & accept_versions
				if cross_versions:
					self.version = max(cross_versions)
				else:
					ascii_txt = "Accept versions: %s" % ["- 1.0 1.1 1.2 1.3 1.4".split()[x] for x in list(accept_versions)]
					self.send(struct.pack("!BBHIHH", max(accept_versions), 1,
						struct.calcsize("!BBHIHH")+len(ascii_txt), hms_xid(),
						0, 0) + ascii_txt.encode("ASCII"))
					raise ChannelClose(ascii_txt)
		return message


def parse_ofp_header(message):
	'''
	@return (version, oftype, message_len, xid)
	'''
	return struct.unpack_from("!BBHI", message)


def ofp_header_only(oftype, version=1, xid=None):
	if xid is None:
		xid = hms_xid()
	return struct.pack("!BBHI", version, oftype, 8, xid)


def hms_xid():
	'''Xid looks readable datetime like format when logged as int.'''
	now = datetime.datetime.now()
	candidate = int(("%02d"*3+"%04d") % (now.hour, now.minute, now.second, now.microsecond/100))
	if hasattr(hms_xid, "dedup"):
		if hms_xid.dedup >= candidate:
			candidate = hms_xid.dedup+1
	setattr(hms_xid, "dedup", candidate)
	return candidate


def ofp_version_normalize(versions):
	if isinstance(versions, list) or isinstance(versions, tuple) or isinstance(versions, set):
		vset = set()
		for version in versions:
			if isinstance(version, float):
				version = [1.0, 1.1, 1.2, 1.3, 1.4].index(version) + 1
			assert isinstance(version, int), "unknown version %s" % version
			vset.add(version)
		return vset
	elif versions is None:
		return set()
	assert False, "unknown versions %s" % versions


def hello(versions, **kwargs):
	xid = kwargs.get("xid", hms_xid())
	if versions:
		vset = ofp_version_normalize(versions)
	else:
		vset = set((1,))
	version = max(vset)
	
	if version < 4:
		return struct.pack("!BBHI", version, 0, 8, xid)
	else:
		units = [0,]*(1 + version//32)
		for v in vset:
			units[v//32] |= 1<<(v%32)
		
		versionbitmap_length = 4 + len(units)*4
		fmt = "!BBHIHH%dI%dx" % (len(units), 8*((len(units)-1)%2))
		return struct.pack(fmt, version, 0, struct.calcsize(fmt), xid, # HELLO
			1, versionbitmap_length, *units) # VERSIONBITMAP


def parse_hello(message):
	(version, oftype, length, xid) = parse_ofp_header(message)
	assert oftype==0 # HELLO
	versions = set()
	if length == 8:
		versions.add(version)
	else:
		(subtype, sublength) = struct.unpack_from("!HH", message, offset=8)
		assert subtype == 1 # VERSIONBITMAP
		units = struct.unpack_from("!%dI" % (sublength/4 - 1), message, offset=12)
		for idx,unit in zip(range(len(units)),units):
			for s in range(32):
				if unit&(1<<s):
					versions.add(idx*32 + s)
	return versions


class OpenflowServerChannel(OpenflowChannel):
	def loop(self):
		try:
			for message in self:
				if not message:
					break
				
				self.handle_proxy(self.handle)(message, self)
		except ChannelClose:
			self.close()
	
	def handle_proxy(self, handle):
		return handle
	
	def handle(self, message, channel):
		logging.getLogger(__name__).warn("check MRO")
		pass

class AutoEchoChannel(OpenflowServerChannel):
	'''
	AuthEchoChannel steals ECHO_REQUEST and automatically send echo response.
	'''
	def handle_proxy(self, handle):
		def intercept(message, channel):
			if message:
				(version, oftype, length, xid) = parse_ofp_header(message)
				if oftype==2: # ECHO
					self.send(struct.pack("!BBHI", self.version, 3, length, xid)+message[8:])
				else:
					super(AutoEchoChannel, self).handle_proxy(handle)(message, channel)
		return intercept


class WeakCallbackCaller(object):
	@property
	def callback(self):
		if self.cbref:
			return self.cbref()


class Barrier(WeakCallbackCaller):
	def __init__(self, xid, message_handler=None):
		if message_handler:
			self.cbref = weakref.ref(message_handler)
		else:
			self.cbref = None
		self.xid = xid


class Chunk(WeakCallbackCaller):
	def __init__(self, message_handler):
		if message_handler:
			self.cbref = weakref.ref(message_handler)
		else:
			self.cbref = None


class ControllerChannel(OpenflowServerChannel, WeakCallbackCaller):
	datapath = None
	auxiliary = None
	cbref = None
	
	def __init__(self, *args, **kwargs):
		super(ControllerChannel, self).__init__(*args, **kwargs)
		self.seq_lock = sched.Lock()
		self.seq = []
	
	def send(self, message, **kwargs):
		with self.seq_lock:
			return self.locked_send(message, **kwargs)
	
	def locked_send(self, message, **kwargs):
		message_handler = kwargs.get("callback") # callable object
		if message_handler is None:
			pass
		else:
			assert isinstance(message_handler, object)
			assert callable(message_handler)
		
		(version, oftype, length, xid) = parse_ofp_header(message)
		if (oftype==18 and version==1) or (oftype==20 and version!=1): # OFPT_BARRIER_REQUEST
			self.seq.append(Barrier(xid, message_handler))
		elif self.seq:
			seq_last = self.seq[-1]
			if isinstance(seq_last, Chunk):
				if seq_last.callback != message_handler:
					bxid = hms_xid()
					if self.version==1:
						msg = ofp_header_only(18, version=1, xid=bxid) # OFPT_BARRIER_REQUEST=18 (v1.0)
					else:
						msg = ofp_header_only(20, version=self.version, xid=bxid) # OFPT_BARRIER_REQUEST=20 (v1.1--v1.4)
					
					self.seq.append(Barrier(bxid))
					self.seq.append(Chunk(message_handler))
					super(ControllerChannel, self).send(msg)
			elif isinstance(seq_last, Barrier):
				self.seq.append(Chunk(message_handler))
			else:
				assert False, "seq element must be Chunk or Barrier"
		else:
			if self.callback != message_handler:
				self.seq.append(Chunk(message_handler))
			if message_handler:
				self.cbfunc = weakref.ref(message_handler)
		
		super(ControllerChannel, self).send(message)
	
	def recv(self):
		message = super(ControllerChannel, self).recv()
		if message:
			(version, oftype, length, xid) = parse_ofp_header(message)
			if oftype==6: # FEATURES_REPLY
				if self.version < 4:
					(self.datapath,) = struct.unpack_from("!Q", message, offset=8) # v1.0--v1.2
				else:
					(self.datapath,_1,_2,self.auxiliary) = struct.unpack_from("!QIBB", message, offset=8) # v1.3--v1.4
		return message
	
	def handle_proxy(self, handle):
		def intercept(message, channel):
			(version, oftype, length, xid) = parse_ofp_header(message)
			
			if hasattr(self, "handle_async") and oftype in (10,11,12):
				# bypass method call for async message
				return super(ControllerChannel, self).handle_proxy(self.handle_async)(message, channel)
			
			with self.seq_lock:
				if self.seq:
					if (oftype==19 and version==1) or (oftype==21 and version!=1): # is barrier
						chunk_drop = False
						for e in self.seq:
							if isinstance(e, Barrier):
								if e.xid == xid:
									self.seq = self.seq[self.seq.index(e)+1:]
									if e.callback:
										return e.callback(message, self)
									return True
								else:
									assert False, "missing barrier(xid=%x) before barrier(xid=%x)" % (e.xid, xid)
							elif isinstance(e, Chunk):
								assert chunk_drop==False, "dropping multiple chunks at a time"
								chunk_drop = True
						assert False, "got unknown barrier xid=%x" % xid
					else:
						e = self.seq[0]
						if isinstance(e, Chunk):
							if e.callback:
								return e.callback(message, self)
			
			if self.callback:
				return self.callback(message, self)
			else:
				return super(ControllerChannel, self).handle_proxy(handle)(message, channel)
			
			logging.getLogger(__name__).warn("No callback found for handling message %s" % binascii.b2a_hex(message))
		return intercept


class RateLimit(object):
	def __init__(self, size):
		self.size = size
		
		self.cold_lock = sched.Lock()
		self.cold = []
		
		self.loop_lock = sched.Lock()
	
	def spawn(self, func, *args, **kwargs):
		with self.cold_lock:
			self.cold.append((func, args, kwargs))
		
		sched.spawn(self.loop)
	
	def loop(self):
		with self.loop_lock:
			while len(self.cold) > 0:
				hot_lock = sched.Lock()
				hot = []
				children = {}
				while len(hot) < self.size and len(self.cold) > 0:
					task = None
					with self.cold_lock:
						task = self.cold.pop(0)
					
					if task:
						(func, args, kwargs) = task
						def proxy():
							func(*args, **kwargs)
							with hot_lock:
								hot.remove(task)
						hot.append(task)
						children[id(task)] = sched.spawn(proxy)
					
					for task_id,job in tuple(children.items()):
						running = False
						with hot_lock:
							if task_id in [id(task) for task in hot]:
								running = True
						
						if running:
							job.join(0.5)
						else:
							chilren.pop(task)
						
						break


class ParallelChannel(OpenflowServerChannel):
	# mixin for parent channel
	socket_dir = None
	async_rate = 0
	
	def __init__(self, *args, **kwargs):
		super(ParallelChannel, self).__init__(*args, **kwargs)
		self.close_lock = sched.Lock()
		self.async_pool = RateLimit(self.async_rate)
	
	def close(self):
		with self.close_lock:
			super(ParallelChannel, self).close()
	
	def handle_proxy(self, handle):
		def intercept(message, channel):
			def proxy(message, channel):
				try:
					handle(message, channel)
				except ChannelClose:
					logging.getLogger(__name__).info("closing", exc_info=True)
					channel.close()
				except:
					logging.getLogger(__name__).error("handle error", exc_info=True)
					channel.close()
			
			rated_call = False
			if self.async_rate:
				(version, oftype, length, xid) = parse_ofp_header(message)
				if oftype in (10, 11, 12):
					rated_call = True
			
			if rated_call:
				self.async_pool.spawn(proxy, message, channel)
			else:
				sched.spawn(proxy, message, channel)
		return super(ParallelChannel, self).handle_proxy(intercept)
	
	def socket_path(self, path):
		if self.socket_dir:
			path = os.path.join(self.socket_dir, path)
		return os.path.abspath(path)
	
	def helper_path(self, suffix):
		old = self.socket_path("unknown-%x.%s" % (id(self), suffix))
		if self.datapath:
			new = self.socket_path("%x-%x.%s" % (self.datapath, id(self), suffix))
			try:
				os.rename(old, new)
			except OSError:
				pass
			return new
		return old
	
	def override_required(self, *args, **kwargs):
		raise Error("Concrete MixIn required")


def bound_socket(info, socktype):
	socket = sched.socket
	if isinstance(info, socket.socket):
		return info
	elif isinstance(info, tuple) or isinstance(info, list):
		infos = [o for o in socket.getaddrinfo(*info) if o[1]==socktype or o[1]==0]
		(family, socktype, proto, canonname, sockaddr) = infos[0]
		s = socket.socket(family, socktype)
		s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
		s.bind(sockaddr)
		return s
	elif isinstance(info, str):
		s = socket.socket(socket.AF_UNIX, socktype)
		s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
		s.bind(info)
		return s
	else:
		raise ValueError("unexpected %s" % info)


def stream_socket(info):
	return bound_socket(info, sched.socket.SOCK_STREAM)


def dgram_socket(info):
	return bound_socket(info, sched.socket.SOCK_DGRAM)


class StreamServer(object):
	channel_cls = None
	def __init__(self, bound_sock, **kwargs):
		self.accepting = False
		self.sock = stream_socket(bound_sock)
		self.channels_lock = sched.Lock()
		self.channels = set()
		self.server_address = self.sock.getsockname()
	
	def start(self):
		self.accepting = True
		sock = self.sock
		sock.settimeout(0.5)
		sock.listen(10)
		sched.spawn(self.run)
	
	def run(self):
		try:
			while self.accepting:
				try:
					s = self.sock.accept()
				except sched.socket.timeout:
					continue
			
				ch = self.channel_cls(socket=s[0], remote_address=s[1], read_wrap=self.read_wrap)
				ch.start()
				sched.spawn(self._loop_runner, ch)
		finally:
			self.sock.close()
	
	def _loop_runner(self, ch):
		with self.channels_lock:
			self.channels.add(ch)
		ch.loop()
		ch.close()
		with self.channels_lock:
			self.channels.remove(ch)
	
	def read_wrap(self, func):
		def wrap(*args, **kwargs):
			if self.accepting==False:
				return b""
			return default_wrapper(func)(*args, **kwargs)
		return wrap
	<commit_msg>Channel.__init__ might fail in some use-case<commit_after>from __future__ import absolute_import
import binascii
import contextlib
import logging
import os
import struct
import types
import weakref
import datetime
from collections import namedtuple


_use_gevent = False
def use_gevent():
	_use_gevent = True

class _sched_proxy(object):
	def __getattr__(self, name):
		_sched = None
		if _use_gevent:
			_sched = __import__("sched_gevent", globals(), level=1)
		else:
			_sched = __import__("sched_basic", globals(), level=1)
		if name in "subprocess socket Queue Lock Event spawn serve_forever".split():
			return getattr(_sched, name)
		raise AttributeError("No such attribute")

sched = _sched_proxy()

def default_wrapper(func):
	def wrap(*args, **kwargs):
		socket = sched.socket
		try:
			return func(*args, **kwargs)
		except socket.timeout:
			return None
		except socket.error as e:
			if e.errno in (os.errno.EAGAIN, os.errno.ECONNRESET, os.errno.EBADF):
				return b""
			elif e.errno in (os.errno.EINTR,):
				return None
			raise
		except KeyboardInterrupt:
			return b""
	return wrap


class ReadWrapper(object):
	def __init__(self, channel, read_wrap):
		self.channel = channel
		self.read_wrap = read_wrap
	
	def __enter__(self):
		self.installed_wrapper = self.channel.read_wrap
		self.channel.read_wrap = self
		return self.channel
	
	def __exit__(self, *args, **kwargs):
		self.channel.read_wrap = self.installed_wrapper
	
	def __call__(self, func):
		def wrap(*args, **kwargs):
			if self.channel.closed:
				return b""
			return self.read_wrap(func)(*args, **kwargs)
		return wrap


class Channel(object):
	'''
	Openflow abstract connection class
	
	This is not only for TCP but also for UDP.
	This is the reason that the name is not "Connection" but "Channel".
	You can subclass this to have instance members, of which lifecycle is 
	the same with channel.
	'''
	def __init__(self, *args, **kwargs):
		self._socket = kwargs.pop("socket", None) # dedicated socket
		self._sendto = kwargs.pop("sendto", None) # only if channel prefers sendto()
		self.reader = kwargs.pop("reader", None)
		self.read_wrap = kwargs.pop("read_wrap", default_wrapper)
		self.remote_address = kwargs.pop("remote_address", None)
		self.local_address = kwargs.pop("local_address", None)
		if self._socket:
			if self.remote_address is None:
				self.remote_address = self._socket.getpeername()
			if self.local_address is None:
				self.local_address = self._socket.getsockname()
			if hasattr(self._socket, "settimeout") and self._socket.gettimeout() == None:
				self._socket.settimeout(0.5)
	
	def attach(self, stream, **kwargs):
		self._socket = stream
		if hasattr(self._socket, "settimeout") and self._socket.gettimeout() == None:
			self._socket.settimeout(0.5)
		self.remote_address = stream.getpeername()
		self.local_address = stream.getsockname()
	
	@property
	def closed(self):
		# This is not self._socket.closed because in some use cases, 
		# self._socket is not available, for example with gevent.server.DatagramServer
		return self.remote_address is None
	
	def close(self):
		if self._socket:
			self._socket.close()
		
		if self.remote_address is not None:
			self.remote_address = None
	
	def send(self, message, **kwargs):
		if self._sendto:
			self._sendto(message, self.remote_address)
		elif self._socket:
			self._socket.send(message)
		else:
			raise ValueError("socket or sendto is required")
	
	def _recv(self, num):
		if self.reader:
			reader = self.reader
		else:
			reader = self._socket.recv
		return ReadWrapper(self, self.read_wrap)(reader)(num)


class Error(Exception):
	pass


class ChannelClose(Error):
	pass


class OpenflowBaseChannel(Channel):
	version = None # The negotiated version
	accept_versions = [4,] # defaults to openflow 1.3
	
	def __init__(self, *args, **kwargs):
		super(OpenflowBaseChannel, self).__init__(*args, **kwargs)
		self.buffer = b""
	
	def __iter__(self):
		while True:
			ret = self.recv()
			if ret:
				yield ret
			else:
				break
	
	def recv(self):
		required_len = 8
		while len(self.buffer) < required_len:
			tmp = super(OpenflowBaseChannel, self)._recv(8192)
			if tmp is None:
				continue
			elif len(tmp)==0:
				return tmp
			self.buffer += tmp
		
		p = struct.unpack_from("!BBHI", self.buffer)
		required_len = p[2]
		
		while len(self.buffer) < required_len:
			tmp = super(OpenflowBaseChannel, self)._recv(8192)
			if tmp is None:
				continue
			elif len(tmp)==0:
				return tmp
			self.buffer += tmp
		
		ret = self.buffer[0:required_len]
		self.buffer = self.buffer[required_len:]
		return ret


class LoggingChannel(OpenflowBaseChannel):
	channel_log_name = "channel"
	send_log_name = "send"
	recv_log_name = "recv"
	remote = ""
	
	def __init__(self, *args, **kwargs):
		super(LoggingChannel, self).__init__(*args, **kwargs)
		if self.remote_address:
			self.remote = " from %s" % self.remote_address[0]
		logging.getLogger(self.channel_log_name).info("%s connect%s" % (self, self.remote))
	
	def send(self, message, **kwargs):
		logging.getLogger(self.send_log_name).debug("%s %s" % (self, binascii.b2a_hex(message)))
		return super(LoggingChannel, self).send(message, **kwargs)
	
	def recv(self):
		message = super(LoggingChannel, self).recv()
		if message: # ignore b"" and None
			logging.getLogger(self.recv_log_name).debug("%s %s" % (self, binascii.b2a_hex(message)))
		return message
	
	def close(self):
		if not self.closed:
			super(LoggingChannel, self).close()
			logging.getLogger(self.channel_log_name).info("%s close%s" % (self, self.remote))


class OpenflowChannel(OpenflowBaseChannel):
	_start = None
	
	def attach(self, stream, **kwargs):
		super(OpenflowBaseChannel, self).attach(stream, **kwargs)
		if kwargs.get("autostart", True):
			self.start()
	
	def start(self):
		if self._start is None:
			self.send(hello(self.accept_versions))
			self._start = True
	
	def recv(self):
		message = super(OpenflowChannel, self).recv()
		if message:
			(version, oftype, length, xid) = parse_ofp_header(message)
			if oftype==0: # HELLO
				accept_versions = ofp_version_normalize(self.accept_versions)
				if not accept_versions:
					accept_versions = set([1,])
				cross_versions = parse_hello(message) & accept_versions
				if cross_versions:
					self.version = max(cross_versions)
				else:
					ascii_txt = "Accept versions: %s" % ["- 1.0 1.1 1.2 1.3 1.4".split()[x] for x in list(accept_versions)]
					self.send(struct.pack("!BBHIHH", max(accept_versions), 1,
						struct.calcsize("!BBHIHH")+len(ascii_txt), hms_xid(),
						0, 0) + ascii_txt.encode("ASCII"))
					raise ChannelClose(ascii_txt)
		return message


def parse_ofp_header(message):
	'''
	@return (version, oftype, message_len, xid)
	'''
	return struct.unpack_from("!BBHI", message)


def ofp_header_only(oftype, version=1, xid=None):
	if xid is None:
		xid = hms_xid()
	return struct.pack("!BBHI", version, oftype, 8, xid)


def hms_xid():
	'''Xid looks readable datetime like format when logged as int.'''
	now = datetime.datetime.now()
	candidate = int(("%02d"*3+"%04d") % (now.hour, now.minute, now.second, now.microsecond/100))
	if hasattr(hms_xid, "dedup"):
		if hms_xid.dedup >= candidate:
			candidate = hms_xid.dedup+1
	setattr(hms_xid, "dedup", candidate)
	return candidate


def ofp_version_normalize(versions):
	if isinstance(versions, list) or isinstance(versions, tuple) or isinstance(versions, set):
		vset = set()
		for version in versions:
			if isinstance(version, float):
				version = [1.0, 1.1, 1.2, 1.3, 1.4].index(version) + 1
			assert isinstance(version, int), "unknown version %s" % version
			vset.add(version)
		return vset
	elif versions is None:
		return set()
	assert False, "unknown versions %s" % versions


def hello(versions, **kwargs):
	xid = kwargs.get("xid", hms_xid())
	if versions:
		vset = ofp_version_normalize(versions)
	else:
		vset = set((1,))
	version = max(vset)
	
	if version < 4:
		return struct.pack("!BBHI", version, 0, 8, xid)
	else:
		units = [0,]*(1 + version//32)
		for v in vset:
			units[v//32] |= 1<<(v%32)
		
		versionbitmap_length = 4 + len(units)*4
		fmt = "!BBHIHH%dI%dx" % (len(units), 8*((len(units)-1)%2))
		return struct.pack(fmt, version, 0, struct.calcsize(fmt), xid, # HELLO
			1, versionbitmap_length, *units) # VERSIONBITMAP


def parse_hello(message):
	(version, oftype, length, xid) = parse_ofp_header(message)
	assert oftype==0 # HELLO
	versions = set()
	if length == 8:
		versions.add(version)
	else:
		(subtype, sublength) = struct.unpack_from("!HH", message, offset=8)
		assert subtype == 1 # VERSIONBITMAP
		units = struct.unpack_from("!%dI" % (sublength/4 - 1), message, offset=12)
		for idx,unit in zip(range(len(units)),units):
			for s in range(32):
				if unit&(1<<s):
					versions.add(idx*32 + s)
	return versions


class OpenflowServerChannel(OpenflowChannel):
	def loop(self):
		try:
			for message in self:
				if not message:
					break
				
				self.handle_proxy(self.handle)(message, self)
		except ChannelClose:
			self.close()
	
	def handle_proxy(self, handle):
		return handle
	
	def handle(self, message, channel):
		logging.getLogger(__name__).warn("check MRO")
		pass

class AutoEchoChannel(OpenflowServerChannel):
	'''
	AuthEchoChannel steals ECHO_REQUEST and automatically send echo response.
	'''
	def handle_proxy(self, handle):
		def intercept(message, channel):
			if message:
				(version, oftype, length, xid) = parse_ofp_header(message)
				if oftype==2: # ECHO
					self.send(struct.pack("!BBHI", self.version, 3, length, xid)+message[8:])
				else:
					super(AutoEchoChannel, self).handle_proxy(handle)(message, channel)
		return intercept


class WeakCallbackCaller(object):
	@property
	def callback(self):
		if self.cbref:
			return self.cbref()


class Barrier(WeakCallbackCaller):
	def __init__(self, xid, message_handler=None):
		if message_handler:
			self.cbref = weakref.ref(message_handler)
		else:
			self.cbref = None
		self.xid = xid


class Chunk(WeakCallbackCaller):
	def __init__(self, message_handler):
		if message_handler:
			self.cbref = weakref.ref(message_handler)
		else:
			self.cbref = None


class ControllerChannel(OpenflowServerChannel, WeakCallbackCaller):
	datapath = None
	auxiliary = None
	cbref = None
	
	def __init__(self, *args, **kwargs):
		super(ControllerChannel, self).__init__(*args, **kwargs)
		self.seq_lock = sched.Lock()
		self.seq = []
	
	def send(self, message, **kwargs):
		with self.seq_lock:
			return self.locked_send(message, **kwargs)
	
	def locked_send(self, message, **kwargs):
		message_handler = kwargs.get("callback") # callable object
		if message_handler is None:
			pass
		else:
			assert isinstance(message_handler, object)
			assert callable(message_handler)
		
		(version, oftype, length, xid) = parse_ofp_header(message)
		if (oftype==18 and version==1) or (oftype==20 and version!=1): # OFPT_BARRIER_REQUEST
			self.seq.append(Barrier(xid, message_handler))
		elif self.seq:
			seq_last = self.seq[-1]
			if isinstance(seq_last, Chunk):
				if seq_last.callback != message_handler:
					bxid = hms_xid()
					if self.version==1:
						msg = ofp_header_only(18, version=1, xid=bxid) # OFPT_BARRIER_REQUEST=18 (v1.0)
					else:
						msg = ofp_header_only(20, version=self.version, xid=bxid) # OFPT_BARRIER_REQUEST=20 (v1.1--v1.4)
					
					self.seq.append(Barrier(bxid))
					self.seq.append(Chunk(message_handler))
					super(ControllerChannel, self).send(msg)
			elif isinstance(seq_last, Barrier):
				self.seq.append(Chunk(message_handler))
			else:
				assert False, "seq element must be Chunk or Barrier"
		else:
			if self.callback != message_handler:
				self.seq.append(Chunk(message_handler))
			if message_handler:
				self.cbfunc = weakref.ref(message_handler)
		
		super(ControllerChannel, self).send(message)
	
	def recv(self):
		message = super(ControllerChannel, self).recv()
		if message:
			(version, oftype, length, xid) = parse_ofp_header(message)
			if oftype==6: # FEATURES_REPLY
				if self.version < 4:
					(self.datapath,) = struct.unpack_from("!Q", message, offset=8) # v1.0--v1.2
				else:
					(self.datapath,_1,_2,self.auxiliary) = struct.unpack_from("!QIBB", message, offset=8) # v1.3--v1.4
		return message
	
	def handle_proxy(self, handle):
		def intercept(message, channel):
			(version, oftype, length, xid) = parse_ofp_header(message)
			
			if hasattr(self, "handle_async") and oftype in (10,11,12):
				# bypass method call for async message
				return super(ControllerChannel, self).handle_proxy(self.handle_async)(message, channel)
			
			with self.seq_lock:
				if self.seq:
					if (oftype==19 and version==1) or (oftype==21 and version!=1): # is barrier
						chunk_drop = False
						for e in self.seq:
							if isinstance(e, Barrier):
								if e.xid == xid:
									self.seq = self.seq[self.seq.index(e)+1:]
									if e.callback:
										return e.callback(message, self)
									return True
								else:
									assert False, "missing barrier(xid=%x) before barrier(xid=%x)" % (e.xid, xid)
							elif isinstance(e, Chunk):
								assert chunk_drop==False, "dropping multiple chunks at a time"
								chunk_drop = True
						assert False, "got unknown barrier xid=%x" % xid
					else:
						e = self.seq[0]
						if isinstance(e, Chunk):
							if e.callback:
								return e.callback(message, self)
			
			if self.callback:
				return self.callback(message, self)
			else:
				return super(ControllerChannel, self).handle_proxy(handle)(message, channel)
			
			logging.getLogger(__name__).warn("No callback found for handling message %s" % binascii.b2a_hex(message))
		return intercept


class RateLimit(object):
	def __init__(self, size):
		self.size = size
		
		self.cold_lock = sched.Lock()
		self.cold = []
		
		self.loop_lock = sched.Lock()
	
	def spawn(self, func, *args, **kwargs):
		with self.cold_lock:
			self.cold.append((func, args, kwargs))
		
		sched.spawn(self.loop)
	
	def loop(self):
		with self.loop_lock:
			while len(self.cold) > 0:
				hot_lock = sched.Lock()
				hot = []
				children = {}
				while len(hot) < self.size and len(self.cold) > 0:
					task = None
					with self.cold_lock:
						task = self.cold.pop(0)
					
					if task:
						(func, args, kwargs) = task
						def proxy():
							func(*args, **kwargs)
							with hot_lock:
								hot.remove(task)
						hot.append(task)
						children[id(task)] = sched.spawn(proxy)
					
					for task_id,job in tuple(children.items()):
						running = False
						with hot_lock:
							if task_id in [id(task) for task in hot]:
								running = True
						
						if running:
							job.join(0.5)
						else:
							chilren.pop(task)
						
						break


class ParallelChannel(OpenflowServerChannel):
	# mixin for parent channel
	socket_dir = None
	async_rate = 0
	
	def __init__(self, *args, **kwargs):
		super(ParallelChannel, self).__init__(*args, **kwargs)
		self.close_lock = sched.Lock()
		self.async_pool = RateLimit(self.async_rate)
	
	def close(self):
		with self.close_lock:
			super(ParallelChannel, self).close()
	
	def handle_proxy(self, handle):
		def intercept(message, channel):
			def proxy(message, channel):
				try:
					handle(message, channel)
				except ChannelClose:
					logging.getLogger(__name__).info("closing", exc_info=True)
					channel.close()
				except:
					logging.getLogger(__name__).error("handle error", exc_info=True)
					channel.close()
			
			rated_call = False
			if self.async_rate:
				(version, oftype, length, xid) = parse_ofp_header(message)
				if oftype in (10, 11, 12):
					rated_call = True
			
			if rated_call:
				self.async_pool.spawn(proxy, message, channel)
			else:
				sched.spawn(proxy, message, channel)
		return super(ParallelChannel, self).handle_proxy(intercept)
	
	def socket_path(self, path):
		if self.socket_dir:
			path = os.path.join(self.socket_dir, path)
		return os.path.abspath(path)
	
	def helper_path(self, suffix):
		old = self.socket_path("unknown-%x.%s" % (id(self), suffix))
		if self.datapath:
			new = self.socket_path("%x-%x.%s" % (self.datapath, id(self), suffix))
			try:
				os.rename(old, new)
			except OSError:
				pass
			return new
		return old
	
	def override_required(self, *args, **kwargs):
		raise Error("Concrete MixIn required")


def bound_socket(info, socktype):
	socket = sched.socket
	if isinstance(info, socket.socket):
		return info
	elif isinstance(info, tuple) or isinstance(info, list):
		infos = [o for o in socket.getaddrinfo(*info) if o[1]==socktype or o[1]==0]
		(family, socktype, proto, canonname, sockaddr) = infos[0]
		s = socket.socket(family, socktype)
		s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
		s.bind(sockaddr)
		return s
	elif isinstance(info, str):
		s = socket.socket(socket.AF_UNIX, socktype)
		s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
		s.bind(info)
		return s
	else:
		raise ValueError("unexpected %s" % info)


def stream_socket(info):
	return bound_socket(info, sched.socket.SOCK_STREAM)


def dgram_socket(info):
	return bound_socket(info, sched.socket.SOCK_DGRAM)


class StreamServer(object):
	channel_cls = None
	def __init__(self, bound_sock, **kwargs):
		self.accepting = False
		self.sock = stream_socket(bound_sock)
		self.channels_lock = sched.Lock()
		self.channels = set()
		self.server_address = self.sock.getsockname()
	
	def start(self):
		self.accepting = True
		sock = self.sock
		sock.settimeout(0.5)
		sock.listen(10)
		sched.spawn(self.run)
	
	def run(self):
		try:
			while self.accepting:
				try:
					s = self.sock.accept()
				except sched.socket.timeout:
					continue
				try:
					ch = self.channel_cls(socket=s[0], remote_address=s[1], read_wrap=self.read_wrap)
					ch.start()
				except Exception as e:
					logging.getLogger(__name__).error("Channel setup failed for %s %s" % (s[1], e), exc_info=True)
					s[0].close()
					continue
				
				sched.spawn(self._loop_runner, ch)
		finally:
			self.sock.close()
	
	def _loop_runner(self, ch):
		with self.channels_lock:
			self.channels.add(ch)
		ch.loop()
		ch.close()
		with self.channels_lock:
			self.channels.remove(ch)
	
	def read_wrap(self, func):
		def wrap(*args, **kwargs):
			if self.accepting==False:
				return b""
			return default_wrapper(func)(*args, **kwargs)
		return wrap
	
{'fraction_non_alphanumeric': 0.12755440185573844, 'fraction_numerical': 0.009665304319010273, 'mean_word_length': 3.8738896366083444, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '174951', 'n_tokens_mistral': 14725, 'n_tokens_neox': 12380, 'n_words': 2995}