<commit_before>#!/usr/bin/env python
# encoding: utf-8
"""
peer.py

Created by <NAME> on 2009-08-25.
Copyright (c) 2009-2011 Exa Networks. All rights reserved.
"""

import sys
import time
import traceback

from bgp.message              import Failure
from bgp.message.nop          import NOP
from bgp.message.open         import Open,Capabilities
from bgp.message.update       import Update
from bgp.message.keepalive    import KeepAlive
from bgp.message.notification import Notification, Notify, NotConnected
from bgp.network.protocol     import Protocol

from bgp.log import Logger,LazyFormat
logger = Logger()

# As we can not know if this is our first start or not, this flag is used to
# always make the program act like it was recovering from a failure
# If set to FALSE, no EOR and OPEN Flags set for Restart will be set in the
# OPEN Graceful Restart Capability
FORCE_GRACEFUL = True

# Present a File like interface to socket.socket

class Peer (object):
	# debug hold/keepalive timers
	debug_trace = True			# debug traceback on unexpected exception
	update_time = 3

	def __init__ (self,neighbor,supervisor):
		self.supervisor = supervisor
		self.neighbor = neighbor
		# The next restart neighbor definition
		self._neighbor = None
		self.bgp = None
		# We may have new update to transmit to our peers, so we need to check
		self._updates = False

		self._loop = None
		self.open = None

		# The peer message should be processed
		self._running = False
		# The peer should restart after a stop
		self._restart = True
		# The peer was restarted (to know what kind of open to send for graceful restart)
		self._restarted = FORCE_GRACEFUL
		self._reset_skip()
		
		# The routes we have parsed from our neighbour
		self._received_routes = []
		
		self._route_parsed = 0L
		self._now = time.time()
		self._next_info = self._now + self.update_time

	def _reset_skip (self):
		# We are currently not skipping connection attempts
		self._skip_time = 0
		# when we can not connect to a peer how many time (in loop) should we back-off
		self._next_skip = 0

	def _more_skip (self):
		self._skip_time = time.time() + self._next_skip
		self._next_skip = int(1+ self._next_skip*1.2)
		if self._next_skip > 60:
			self._next_skip = 60

	def me (self,message):
		return "Peer %15s ASN %-7s %s" % (self.neighbor.peer_address,self.neighbor.peer_as,message)

	def stop (self):
		self._running = False
		self._restart = False
		self._restarted = False
		self._reset_skip()

	def reload (self,routes):
		self._updates = True
		self.neighbor.set_routes(routes)
		self._reset_skip()

	def restart (self,restart_neighbor=None):
		# we want to tear down the session and re-establish it
		self._running = False
		self._restart = True
		self._restarted = True
		self._neighbor = restart_neighbor
		self._reset_skip()

	def run (self):
		if self._loop:
			try:
				if self._skip_time > time.time():
					return None
				else:
					return self._loop.next()
			except StopIteration:
				self._loop = None
		elif self._restart:
			# If we are restarting, and the neighbor definition is different, update the neighbor
			if self._neighbor:
				self.neighbor = self._neighbor
				self._neighbor = None
			self._running = True
			self._loop = self._run()
		else:
			self.bgp.close()
			self.supervisor.unschedule(self)

	def _run (self,max_wait_open=10.0):
		try:
			self.bgp = Protocol(self)
			self.bgp.connect()

			self._reset_skip()

			_open = self.bgp.new_open(self._restarted)
			logger.message(self.me('>> %s' % _open))
			yield None

			start = time.time()
			while True:
				self.open = self.bgp.read_open(_open,self.neighbor.peer_address.ip)
				# OPEN or NOP
				if self.open.TYPE == Open.TYPE:
					logger.message(self.me('<< %s' % self.open))
					yield None
					break
				if time.time() - start > max_wait_open: 
					logger.message(self.me('Waited for an OPEN for too long - killing the session'))
					raise Notify(1,1,'The client took over %s seconds to send the OPEN, closing' % str(max_wait_open))
				yield None

			message = self.bgp.new_keepalive(force=True)
			logger.message(self.me('>> KEEPALIVE (OPENCONFIRM)'))
			yield True

			while True:
				message = self.bgp.read_keepalive()
				# KEEPALIVE or NOP
				if message.TYPE == KeepAlive.TYPE:
					logger.message(self.me('<< KEEPALIVE (ESTABLISHED)'))
					break
				yield None

			count = 0
			for count in self.bgp.new_announce():
				yield True
			if count:
				logger.message(self.me('>> %d UPDATE(s)' % count))

			eor = False
			if	self.neighbor.graceful_restart and \
				self.open.capabilities.announced(Capabilities.MULTIPROTOCOL_EXTENSIONS) and \
				self.open.capabilities.announced(Capabilities.GRACEFUL_RESTART):

				families = []
				for family in self.open.capabilities[Capabilities.GRACEFUL_RESTART].families():
					if family in self.neighbor.families:
						families.append(family)
				self.bgp.new_eors(families)
				if families:
					eor = True
					logger.message(self.me('>> EOR %s' % ', '.join(['%s %s' % (str(afi),str(safi)) for (afi,safi) in families])))
			
			if not eor:
				# If we are not sending an EOR, send a keepalive as soon as when finished
				# So the other routers knows that we have no (more) routes to send ...
				# (is that behaviour documented somewhere ??)
				c,k = self.bgp.new_keepalive(True)
				if k: logger.message(self.me('>> KEEPALIVE (no more UPDATE and no EOR)'))

			while self._running:
				self._now = time.time()
				if self._now > self._next_info:
					self._next_info = self._now + self.update_time
					display_update = True
				else:
					display_update = False

				c,k = self.bgp.new_keepalive()
				if k: logger.message(self.me('>> KEEPALIVE'))

				if display_update:
					logger.timers(self.me('Sending Timer %d second(s) left' % c))

				message = self.bgp.read_message()
				# let's read if we have keepalive before doing the timer check
				c = self.bgp.check_keepalive()

				if display_update:
					logger.timers(self.me('Receive Timer %d second(s) left' % c))

				if message.TYPE == KeepAlive.TYPE:
					logger.message(self.me('<< KEEPALIVE'))
				if message.TYPE == Update.TYPE:
					self._received_routes.extend(message.routes)
					if message.routes:
						logger.message(self.me('<< UPDATE'))
						self._route_parsed += len(message.routes)
						if self._route_parsed:
							for route in message.routes:
								logger.routes(LazyFormat(self.me(''),str,route))
							if display_update:
								logger.supervisor(self.me('processed %d routes' % self._route_parsed))
					else:
						logger.message(self.me('<< UPDATE (not parsed)'))
				if message.TYPE not in (KeepAlive.TYPE,Update.TYPE,NOP.TYPE):
					 logger.message(self.me('<< %d' % ord(message.TYPE)))

				if self._updates:
					self._updates = False

					count = 0
					for count in self.bgp.new_update():
						yield True
					logger.message(self.me('>> UPDATE (%d)' % count))

				yield None
			
			if self.neighbor.graceful_restart and self.open.capabilities.announced(Capabilities.GRACEFUL_RESTART):
				logger.warning('Closing the connection without notification')
				self.bgp.close()
				return

			# User closing the connection
			raise Notify(6,3)
		except NotConnected, e:
			logger.warning('we can not connect to the peer %s' % str(e))
			self._more_skip()
			try:
				self.bgp.close()
			except Failure:
				pass
			return
		except Notify,e:
			logger.message(self.me('Sending Notification (%d,%d) [%s]  %s' % (e.code,e.subcode,str(e),e.data)))
			try:
				self.bgp.new_notification(e)
				self.bgp.close()
			except Failure:
				pass
			return
		except Notification, e:
			logger.message(self.me('Received Notification (%d,%d) from peer %s' % (e.code,e.subcode,str(e))))
			self.bgp.close()
			return
		except Failure, e:
			logger.warning(self.me(str(e)),'connection')
			self._more_skip()
			self.bgp.close()
			return
		except Exception, e:
			logger.warning(self.me('UNHANDLED EXCEPTION'))
			self._more_skip()
			if self.debug_trace:
				# should really go to syslog
				traceback.print_exc(file=sys.stdout)
				raise
			else:
				logger.warning(self.me(str(e)))
			if self.bgp: self.bgp.close()
			return

	def received_routes (self):
		if self._received_routes:
			route = self._received_routes.pop(0)
			yield str(route)
			del(route)
		else:
			raise StopIteration()
			<commit_msg>trying to reduce memory usage<commit_after>#!/usr/bin/env python
# encoding: utf-8
"""
peer.py

Created by <NAME> on 2009-08-25.
Copyright (c) 2009-2011 Exa Networks. All rights reserved.
"""

import sys
import time
import traceback

from bgp.message              import Failure
from bgp.message.nop          import NOP
from bgp.message.open         import Open,Capabilities
from bgp.message.update       import Update
from bgp.message.keepalive    import KeepAlive
from bgp.message.notification import Notification, Notify, NotConnected
from bgp.network.protocol     import Protocol

from bgp.log import Logger,LazyFormat
logger = Logger()

# As we can not know if this is our first start or not, this flag is used to
# always make the program act like it was recovering from a failure
# If set to FALSE, no EOR and OPEN Flags set for Restart will be set in the
# OPEN Graceful Restart Capability
FORCE_GRACEFUL = True

# Present a File like interface to socket.socket

class Peer (object):
	# debug hold/keepalive timers
	debug_trace = True			# debug traceback on unexpected exception
	update_time = 3

	def __init__ (self,neighbor,supervisor):
		self.supervisor = supervisor
		self.neighbor = neighbor
		# The next restart neighbor definition
		self._neighbor = None
		self.bgp = None
		# We may have new update to transmit to our peers, so we need to check
		self._updates = False

		self._loop = None
		self.open = None

		# The peer message should be processed
		self._running = False
		# The peer should restart after a stop
		self._restart = True
		# The peer was restarted (to know what kind of open to send for graceful restart)
		self._restarted = FORCE_GRACEFUL
		self._reset_skip()
		
		# The routes we have parsed from our neighbour
		self._received_routes = []
		
		self._route_parsed = 0L
		self._now = time.time()
		self._next_info = self._now + self.update_time

	def _reset_skip (self):
		# We are currently not skipping connection attempts
		self._skip_time = 0
		# when we can not connect to a peer how many time (in loop) should we back-off
		self._next_skip = 0

	def _more_skip (self):
		self._skip_time = time.time() + self._next_skip
		self._next_skip = int(1+ self._next_skip*1.2)
		if self._next_skip > 60:
			self._next_skip = 60

	def me (self,message):
		return "Peer %15s ASN %-7s %s" % (self.neighbor.peer_address,self.neighbor.peer_as,message)

	def stop (self):
		self._running = False
		self._restart = False
		self._restarted = False
		self._reset_skip()

	def reload (self,routes):
		self._updates = True
		self.neighbor.set_routes(routes)
		self._reset_skip()

	def restart (self,restart_neighbor=None):
		# we want to tear down the session and re-establish it
		self._running = False
		self._restart = True
		self._restarted = True
		self._neighbor = restart_neighbor
		self._reset_skip()

	def run (self):
		if self._loop:
			try:
				if self._skip_time > time.time():
					return None
				else:
					return self._loop.next()
			except StopIteration:
				self._loop = None
		elif self._restart:
			# If we are restarting, and the neighbor definition is different, update the neighbor
			if self._neighbor:
				self.neighbor = self._neighbor
				self._neighbor = None
			self._running = True
			self._loop = self._run()
		else:
			self.bgp.close()
			self.supervisor.unschedule(self)

	def _run (self,max_wait_open=10.0):
		try:
			self.bgp = Protocol(self)
			self.bgp.connect()

			self._reset_skip()

			_open = self.bgp.new_open(self._restarted)
			logger.message(self.me('>> %s' % _open))
			yield None

			start = time.time()
			while True:
				self.open = self.bgp.read_open(_open,self.neighbor.peer_address.ip)
				# OPEN or NOP
				if self.open.TYPE == Open.TYPE:
					logger.message(self.me('<< %s' % self.open))
					yield None
					break
				if time.time() - start > max_wait_open: 
					logger.message(self.me('Waited for an OPEN for too long - killing the session'))
					raise Notify(1,1,'The client took over %s seconds to send the OPEN, closing' % str(max_wait_open))
				yield None

			message = self.bgp.new_keepalive(force=True)
			logger.message(self.me('>> KEEPALIVE (OPENCONFIRM)'))
			yield True

			while True:
				message = self.bgp.read_keepalive()
				# KEEPALIVE or NOP
				if message.TYPE == KeepAlive.TYPE:
					logger.message(self.me('<< KEEPALIVE (ESTABLISHED)'))
					break
				yield None

			count = 0
			for count in self.bgp.new_announce():
				yield True
			if count:
				logger.message(self.me('>> %d UPDATE(s)' % count))

			eor = False
			if	self.neighbor.graceful_restart and \
				self.open.capabilities.announced(Capabilities.MULTIPROTOCOL_EXTENSIONS) and \
				self.open.capabilities.announced(Capabilities.GRACEFUL_RESTART):

				families = []
				for family in self.open.capabilities[Capabilities.GRACEFUL_RESTART].families():
					if family in self.neighbor.families:
						families.append(family)
				self.bgp.new_eors(families)
				if families:
					eor = True
					logger.message(self.me('>> EOR %s' % ', '.join(['%s %s' % (str(afi),str(safi)) for (afi,safi) in families])))
			
			if not eor:
				# If we are not sending an EOR, send a keepalive as soon as when finished
				# So the other routers knows that we have no (more) routes to send ...
				# (is that behaviour documented somewhere ??)
				c,k = self.bgp.new_keepalive(True)
				if k: logger.message(self.me('>> KEEPALIVE (no more UPDATE and no EOR)'))

			while self._running:
				self._now = time.time()
				if self._now > self._next_info:
					self._next_info = self._now + self.update_time
					display_update = True
				else:
					display_update = False

				c,k = self.bgp.new_keepalive()
				if k: logger.message(self.me('>> KEEPALIVE'))

				if display_update:
					logger.timers(self.me('Sending Timer %d second(s) left' % c))

				message = self.bgp.read_message()
				# let's read if we have keepalive before doing the timer check
				c = self.bgp.check_keepalive()

				if display_update:
					logger.timers(self.me('Receive Timer %d second(s) left' % c))

				if message.TYPE == KeepAlive.TYPE:
					logger.message(self.me('<< KEEPALIVE'))
				if message.TYPE == Update.TYPE:
					self._received_routes.extend(message)
					if message.routes:
						logger.message(self.me('<< UPDATE'))
						self._route_parsed += len(message.routes)
						if self._route_parsed:
							for route in message.routes:
								logger.routes(LazyFormat(self.me(''),str,route))
							if display_update:
								logger.supervisor(self.me('processed %d routes' % self._route_parsed))
					else:
						logger.message(self.me('<< UPDATE (not parsed)'))
				if message.TYPE not in (KeepAlive.TYPE,Update.TYPE,NOP.TYPE):
					 logger.message(self.me('<< %d' % ord(message.TYPE)))

				if self._updates:
					self._updates = False

					count = 0
					for count in self.bgp.new_update():
						yield True
					logger.message(self.me('>> UPDATE (%d)' % count))

				yield None
			
			if self.neighbor.graceful_restart and self.open.capabilities.announced(Capabilities.GRACEFUL_RESTART):
				logger.warning('Closing the connection without notification')
				self.bgp.close()
				return

			# User closing the connection
			raise Notify(6,3)
		except NotConnected, e:
			logger.warning('we can not connect to the peer %s' % str(e))
			self._more_skip()
			try:
				self.bgp.close()
			except Failure:
				pass
			return
		except Notify,e:
			logger.message(self.me('Sending Notification (%d,%d) [%s]  %s' % (e.code,e.subcode,str(e),e.data)))
			try:
				self.bgp.new_notification(e)
				self.bgp.close()
			except Failure:
				pass
			return
		except Notification, e:
			logger.message(self.me('Received Notification (%d,%d) from peer %s' % (e.code,e.subcode,str(e))))
			self.bgp.close()
			return
		except Failure, e:
			logger.warning(self.me(str(e)),'connection')
			self._more_skip()
			self.bgp.close()
			return
		except Exception, e:
			logger.warning(self.me('UNHANDLED EXCEPTION'))
			self._more_skip()
			if self.debug_trace:
				# should really go to syslog
				traceback.print_exc(file=sys.stdout)
				raise
			else:
				logger.warning(self.me(str(e)))
			if self.bgp: self.bgp.close()
			return

	def received_routes (self):
		if self._received_routes:
			message = self._received_routes.pop(0)
			yield str(message.route)
			del(message)
		else:
			raise StopIteration()
			
{'fraction_non_alphanumeric': 0.11115789473684211, 'fraction_numerical': 0.004932330827067669, 'mean_word_length': 3.2554389557205017, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 41, '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': '4084945', 'n_tokens_mistral': 6744, 'n_tokens_neox': 5519, 'n_words': 1518}