exabgp.reactor.loop.Reactor.run()

Complexity

Conditions

54

Size

Total Lines	193
Code Lines	130

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

# encoding: utf-8
	def run (self, validate, root):
		self.daemon.daemonise()

		# Make sure we create processes once we have closed file descriptor
		# unfortunately, this must be done before reading the configuration file
		# so we can not do it with dropped privileges
		self.processes = Processes()

		# we have to read the configuration possibly with root privileges
		# as we need the MD5 information when we bind, and root is needed
		# to bind to a port < 1024

		# this is undesirable as :
		# - handling user generated data as root should be avoided
		# - we may not be able to reload the configuration once the privileges are dropped

		# but I can not see any way to avoid it
		for ip in self._ips:
			if not self.listener.listen_on(ip, None, self._port, None, False, None):
				return self.Exit.listening

		if not self.load():
			return self.Exit.configuration

		if validate:  # only validate configuration
			self.logger.warning('','configuration')
			self.logger.warning('parsed Neighbors, un-templated','configuration')
			self.logger.warning('------------------------------','configuration')
			self.logger.warning('','configuration')
			for key in self._peers:
				self.logger.warning(str(self._peers[key].neighbor),'configuration')
				self.logger.warning('','configuration')
			return self.Exit.validate

		for neighbor in self.configuration.neighbors.values():
			if neighbor.listen:
				if not self.listener.listen_on(neighbor.md5_ip, neighbor.peer_address, neighbor.listen, neighbor.md5_password, neighbor.md5_base64, neighbor.ttl_in):
					return self.Exit.listening

		if not self.early_drop:
			self.processes.start(self.configuration.processes)

		if not self.daemon.drop_privileges():
			self.logger.critical('could not drop privileges to \'%s\' refusing to run as root' % self.daemon.user,'reactor')
			self.logger.critical('set the environmemnt value exabgp.daemon.user to change the unprivileged user','reactor')
			return self.Exit.privileges

		if self.early_drop:
			self.processes.start(self.configuration.processes)

		# This is required to make sure we can write in the log location as we now have dropped root privileges
		if not self.logger.restart():
			self.logger.critical('could not setup the logger, aborting','reactor')
			return self.Exit.log

		if not self.daemon.savepid():
			return self.Exit.pid

		# did we complete the run of updates caused by the last SIGUSR1/SIGUSR2 ?
		reload_completed = False

		wait = environment.settings().tcp.delay
		if wait:
			sleeptime = (wait * 60) - int(time.time()) % (wait * 60)
			self.logger.debug('waiting for %d seconds before connecting' % sleeptime,'reactor')
			time.sleep(float(sleeptime))

		workers = {}
		peers = set()
		api_fds = []

		while True:
			try:
				if self.signal.received:
					for key in self._peers:
						if self._peers[key].neighbor.api['signal']:
							self._peers[key].reactor.processes.signal(self._peers[key].neighbor,self.signal.number)

					signaled = self.signal.received
					self.signal.rearm()

					if signaled == Signal.SHUTDOWN:
						self.shutdown()
						break

					if signaled == Signal.RESTART:
						self.restart()
						continue

					if not reload_completed:
						continue

					if signaled == Signal.FULL_RELOAD:
						self._reload_processes = True

					if signaled in (Signal.RELOAD, Signal.FULL_RELOAD):
						self.load()
						self.processes.start(self.configuration.processes,self._reload_processes)
						self._reload_processes = False
						continue

				if self.listener.incoming():
					# check all incoming connection
					self.asynchronous.schedule(str(uuid.uuid1()),'checking for new connection(s)',self.listener.new_connections())

				peers = self.active_peers()
				if self._completed(peers):
					reload_completed = True

				sleep = self._sleep_time

				# do not attempt to listen on closed sockets even if the peer is still here
				for io in list(workers.keys()):
					if io == -1:
						self._poller.unregister(io)
						del workers[io]

				# give a turn to all the peers
				for key in list(peers):
					peer = self._peers[key]

					# limit the number of message handling per second
					if self._rate_limited(key,peer.neighbor.rate_limit):
						peers.discard(key)
						continue

					# handle the peer
					action = peer.run()

					# .run() returns an ACTION enum:
					# * immediate if it wants to be called again
					# * later if it should be called again but has no work atm
					# * close if it is finished and is closing down, or restarting
					if action == ACTION.CLOSE:
						if key in self._peers:
							del self._peers[key]
						peers.discard(key)
					# we are loosing this peer, not point to schedule more process work
					elif action == ACTION.LATER:
						io = peer.socket()
						if io != -1:
							self._poller.register(io, self.READ_ONLY)
							workers[io] = key
						# no need to come back to it before a a full cycle
						peers.discard(key)
					elif action == ACTION.NOW:
						sleep = 0

					if not peers:
						break

				# read at least on message per process if there is some and parse it
				for service,command in self.processes.received():
					self.api.text(self,service,command)
					sleep = 0

				self.asynchronous.run()

				if api_fds != self.processes.fds:
					for fd in api_fds:
						if fd == -1:
							continue
						if fd not in self.processes.fds:
							self._poller.unregister(fd)
					for fd in self.processes.fds:
						if fd == -1:
							continue
						if fd not in api_fds:
							self._poller.register(fd, self.READ_ONLY)
					api_fds = self.processes.fds

				for io in self._wait_for_io(sleep):
					if io not in api_fds:
						peers.add(workers[io])

				if self._stopping and not self._peers.keys():
					self._termination('exiting on peer termination',self.Exit.normal)

			except KeyboardInterrupt:
				self._termination('^C received',self.Exit.normal)
			except SystemExit:
				self._termination('exiting', self.Exit.normal)
			# socket.error is a subclass of IOError (so catch it first)
			except socket.error:
				self._termination('socket error received',self.Exit.socket)
			except IOError:
				self._termination('I/O Error received, most likely ^C during IO',self.Exit.io_error)
			except ProcessError:
				self._termination('Problem when sending message(s) to helper program, stopping',self.Exit.process)
			except select.error:
				self._termination('problem using select, stopping',self.Exit.select)

		return self.exit_code