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

Created by <NAME> on 2011-11-30.
Copyright (c) 2011 Exa Networks. All rights reserved.
"""

# http://code.google.com/speed/articles/web-metrics.html

from .util.logger import logger

class Reactor(object):
	def __init__(self, web, proxy, decider, content, client, resolver, poller):
		self.web = web            # Manage listening web sockets
		self.proxy = proxy        # Manage listening proxy sockets
		self.decider = decider    # Task manager for handling child decider processes
		self.content = content    # The Content Download manager
		self.client = client      # Currently open client connections
		self.resolver = resolver  # The DNS query manager
		self.poller = poller      # Interface to the poller
		self.running = True
		self._loop = None

	def run(self):
		poller = self.poller

#		count = 0
#		s_times = []
#		w_times = []

		decisions = []

		# Manually timeout queries
		timedout = self.resolver.cleanup()
		for client_id, command, decision in timedout:
			decisions.append((client_id, command, decision))

		self.resolver.expireCache()

		while self.running:
			# wait until we have something to do
			events = poller.poll()

			logger.debug('supervisor','events : ' + ', '.join(events.keys()))

			# handle new connections before anything else
			for sock in events.get('read_proxy',[]):
				for s, peer in self.proxy.accept(sock):
					self.client.newConnection(s, peer, 'proxy')


			# handle new connections before anything else
			for sock in events.get('read_web',[]):
				for s, peer in self.web.accept(sock):
					self.client.newConnection(s, peer, 'web')


			# incoming new requests from clients
			for client in events.get('opening_client',[]):
				client_id, peer, request, data, source = self.client.readRequest(client)
				if request:
					# we have a new request - decide what to do with it
					self.decider.request(client_id, peer, request, source)



			# incoming data from clients
			for client in events.get('read_client',[]):
				client_id, peer, request, data = self.client.readDataBySocket(client)
				if request:
					# tell the client to hang up
					self.client.sendDataByName(client_id, None)

				elif data:
					# we read something from the client so pass it on to the remote server
					status, flipflop = self.content.sendClientData(client_id, data)

					if flipflop:
						if status:
							self.client.corkUploadByName(client_id)
						else:
							self.client.uncorkUploadByName(client_id)

				elif data is None:
					self.content.endClientDownload(client_id)


			# incoming data - web pages
			for fetcher in events.get('read_download',[]):
				client_id, page_data = self.content.readData(fetcher)

				# send received data to the client that requested it
				status, flipflop = self.client.sendDataByName(client_id, page_data)

				# check to see if the client went away
				if status is None:
					if page_data is not None:
						# should we use the fetcher (socket) as an index here rather than the client id?
						# should we just wait for the next loop when we'll be notified of the client disconnect?
						self.content.endClientDownload(client_id)

				elif flipflop:
					# status should be true here - we don't read from the server when buffering
					if status:      # Buffering
						self.content.corkClientDownload(client_id)

					else:            # No buffer
						self.content.uncorkClientDownload(client_id)

			# decisions made by the child processes
			for worker in events.get('read_workers',[]):
				client_id, command, decision = self.decider.getDecision(worker)

				# check that the client didn't get bored and go away
				if client_id in self.client:
					if self.resolver.resolves(command, decision):
						identifier, response = self.resolver.startResolving(client_id, command, decision)

						if response:
							cid, command, decision = response
							decisions.append((client_id, command, decision))

						# something went wrong
						elif not identifier:
							commmand, decision = self.decider.showInternalError()
					else:
						decisions.append((client_id, command, decision))

			# decisions with a resolved hostname
			for resolver in events.get('read_resolver', []):
				response = self.resolver.getResponse(resolver)
				if response:
					client_id, command, decision = response
					decisions.append((client_id, command, decision))

			# all decisions we are currently able to process
			for client_id, command, decision in decisions:
				response, restricted = self.content.getContent(client_id, command, decision)

				# Signal to the client that we'll be streaming data to it or
				# give it the location of the local content to return.
				data = self.client.startData(client_id, response, restricted)<commit_msg>An identifier of 0 is allowed<commit_after>#!/usr/bin/env python
# encoding: utf-8
"""
server.py

Created by <NAME> on 2011-11-30.
Copyright (c) 2011 Exa Networks. All rights reserved.
"""

# http://code.google.com/speed/articles/web-metrics.html

from .util.logger import logger

class Reactor(object):
	def __init__(self, web, proxy, decider, content, client, resolver, poller):
		self.web = web            # Manage listening web sockets
		self.proxy = proxy        # Manage listening proxy sockets
		self.decider = decider    # Task manager for handling child decider processes
		self.content = content    # The Content Download manager
		self.client = client      # Currently open client connections
		self.resolver = resolver  # The DNS query manager
		self.poller = poller      # Interface to the poller
		self.running = True
		self._loop = None

	def run(self):
		poller = self.poller

#		count = 0
#		s_times = []
#		w_times = []

		decisions = []

		# Manually timeout queries
		timedout = self.resolver.cleanup()
		for client_id, command, decision in timedout:
			decisions.append((client_id, command, decision))

		self.resolver.expireCache()

		while self.running:
			# wait until we have something to do
			events = poller.poll()

			logger.debug('supervisor','events : ' + ', '.join(events.keys()))

			# handle new connections before anything else
			for sock in events.get('read_proxy',[]):
				for s, peer in self.proxy.accept(sock):
					self.client.newConnection(s, peer, 'proxy')


			# handle new connections before anything else
			for sock in events.get('read_web',[]):
				for s, peer in self.web.accept(sock):
					self.client.newConnection(s, peer, 'web')


			# incoming new requests from clients
			for client in events.get('opening_client',[]):
				client_id, peer, request, data, source = self.client.readRequest(client)
				if request:
					# we have a new request - decide what to do with it
					self.decider.request(client_id, peer, request, source)



			# incoming data from clients
			for client in events.get('read_client',[]):
				client_id, peer, request, data = self.client.readDataBySocket(client)
				if request:
					# tell the client to hang up
					self.client.sendDataByName(client_id, None)

				elif data:
					# we read something from the client so pass it on to the remote server
					status, flipflop = self.content.sendClientData(client_id, data)

					if flipflop:
						if status:
							self.client.corkUploadByName(client_id)
						else:
							self.client.uncorkUploadByName(client_id)

				elif data is None:
					self.content.endClientDownload(client_id)


			# incoming data - web pages
			for fetcher in events.get('read_download',[]):
				client_id, page_data = self.content.readData(fetcher)

				# send received data to the client that requested it
				status, flipflop = self.client.sendDataByName(client_id, page_data)

				# check to see if the client went away
				if status is None:
					if page_data is not None:
						# should we use the fetcher (socket) as an index here rather than the client id?
						# should we just wait for the next loop when we'll be notified of the client disconnect?
						self.content.endClientDownload(client_id)

				elif flipflop:
					# status should be true here - we don't read from the server when buffering
					if status:      # Buffering
						self.content.corkClientDownload(client_id)

					else:            # No buffer
						self.content.uncorkClientDownload(client_id)

			# decisions made by the child processes
			for worker in events.get('read_workers',[]):
				client_id, command, decision = self.decider.getDecision(worker)

				# check that the client didn't get bored and go away
				if client_id in self.client:
					if self.resolver.resolves(command, decision):
						identifier, response = self.resolver.startResolving(client_id, command, decision)

						if response:
							cid, command, decision = response
							decisions.append((client_id, command, decision))

						# something went wrong
						elif identifier is None:
							commmand, decision = self.decider.showInternalError()
					else:
						decisions.append((client_id, command, decision))

			# decisions with a resolved hostname
			for resolver in events.get('read_resolver', []):
				response = self.resolver.getResponse(resolver)
				if response:
					client_id, command, decision = response
					decisions.append((client_id, command, decision))

			# all decisions we are currently able to process
			for client_id, command, decision in decisions:
				response, restricted = self.content.getContent(client_id, command, decision)

				# Signal to the client that we'll be streaming data to it or
				# give it the location of the local content to return.
				data = self.client.startData(client_id, response, restricted)
{'fraction_non_alphanumeric': 0.09058725531028738, 'fraction_numerical': 0.0030195751770095793, 'mean_word_length': 3.293696915511846, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6851722', 'n_tokens_mistral': 3422, 'n_tokens_neox': 2755, 'n_words': 929}