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

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

from threading import Thread
from Queue import Empty
import subprocess
import errno

import os
import time
import socket

#import fcntl

from exaproxy.http.header import Header

from exaproxy.util.logger import logger
from exaproxy.configuration import configuration

from exaproxy.network.resolver import DNSResolver

class Worker (Thread):
	# TODO : if the program is a function, fork and run :)
	
	def __init__ (self, name, request_box, program):
		# XXX: all this could raise things
		r, w = os.pipe()                                # pipe for communication with the main thread
		self.response_box_write = os.fdopen(w,'w',0)    # results are written here
		self.response_box_read = os.fdopen(r,'r',0)     # read from the main thread

		self.resolver = DNSResolver(configuration.RESOLV)

		self.wid = name                               # a unique name
		self.creation = time.time()                   # when the thread was created
	#	self.last_worked = self.creation              # when the thread last picked a task
		self.request_box = request_box                # queue with HTTP headers to process

		self.program = program                        # the squid redirector program to fork 
		self.running = True                           # the thread is active

		# Do not move, we need the forking AFTER the setup
		self.process = self._createProcess()          # the forked program to handle classification
		Thread.__init__(self)

	def _createProcess (self):
		try:
			process = subprocess.Popen([self.program,],
				stdin=subprocess.PIPE,
				stdout=subprocess.PIPE,
				universal_newlines=True,
			)
			logger.debug('worker %s' % self.wid,'spawn process %s' % self.program)
		except KeyboardInterrupt:
			process = None
		except (subprocess.CalledProcessError,OSError,ValueError):
			logger.error('worker %s' % self.wid,'could not spawn process %s' % self.program)
			process = None
		return process

	def destroyProcess (self):
		logger.error('worker %s' % self.wid,'destroying process %s' % self.program)
		if not self.process:
			return
		try:
			if self.process:
				self.process.terminate()
				self.process.wait()
				logger.info('worker %s' % self.wid,'terminated process PID %s' % pid)
		except OSError, e:
			# No such processs
			if e[0] != errno.ESRCH:
				logger.error('worker %s' % self.wid,'PID %s died' % pid)

	def stop (self):
		logger.debug('worker %s' % self.wid,'shutdown')
		# The worker thread may be blocked reading from the queue
		# so the shutdown will not be immediate
		self.running = False


	def _classify (self, client_ip, method, url):
		squid = '%s %s - %s -' % (url, client_ip, method)
		#logger.info('worker %s' % self.wid, 'sending to classifier: [%s]' % squid)
		try:
			self.process.stdin.write(squid + os.linesep)
			response = self.process.stdout.readline().strip()

			if not response:
				classification, data = 'permit', None

			elif response.startswith('http://'):
				response = response[7:]

				if response == url:
					classification, data = 'permit', None
				elif response.startswith(url.split('/', 1)[0]+'/'):
					classification, data = 'rewrite', ('/'+response.split('/', 1)[1]) if '/' in url else ''
				else:
					classification, data = 'redirect', 'http://' + response
<commit_msg>Fix bad reference to pid instead of self.process.pid<commit_after>#!/usr/bin/env python
# encoding: utf-8
"""
process.py

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

from threading import Thread
from Queue import Empty
import subprocess
import errno

import os
import time
import socket

#import fcntl

from exaproxy.http.header import Header

from exaproxy.util.logger import logger
from exaproxy.configuration import configuration

from exaproxy.network.resolver import DNSResolver

class Worker (Thread):
	# TODO : if the program is a function, fork and run :)
	
	def __init__ (self, name, request_box, program):
		# XXX: all this could raise things
		r, w = os.pipe()                                # pipe for communication with the main thread
		self.response_box_write = os.fdopen(w,'w',0)    # results are written here
		self.response_box_read = os.fdopen(r,'r',0)     # read from the main thread

		self.resolver = DNSResolver(configuration.RESOLV)

		self.wid = name                               # a unique name
		self.creation = time.time()                   # when the thread was created
	#	self.last_worked = self.creation              # when the thread last picked a task
		self.request_box = request_box                # queue with HTTP headers to process

		self.program = program                        # the squid redirector program to fork 
		self.running = True                           # the thread is active

		# Do not move, we need the forking AFTER the setup
		self.process = self._createProcess()          # the forked program to handle classification
		Thread.__init__(self)

	def _createProcess (self):
		try:
			process = subprocess.Popen([self.program,],
				stdin=subprocess.PIPE,
				stdout=subprocess.PIPE,
				universal_newlines=True,
			)
			logger.debug('worker %s' % self.wid,'spawn process %s' % self.program)
		except KeyboardInterrupt:
			process = None
		except (subprocess.CalledProcessError,OSError,ValueError):
			logger.error('worker %s' % self.wid,'could not spawn process %s' % self.program)
			process = None
		return process

	def destroyProcess (self):
		logger.error('worker %s' % self.wid,'destroying process %s' % self.program)
		if not self.process:
			return
		try:
			if self.process:
				self.process.terminate()
				self.process.wait()
				logger.info('worker %s' % self.wid,'terminated process PID %s' % self.process.pid)
		except OSError, e:
			# No such processs
			if e[0] != errno.ESRCH:
				logger.error('worker %s' % self.wid,'PID %s died' % self.process.pid)

	def stop (self):
		logger.debug('worker %s' % self.wid,'shutdown')
		# The worker thread may be blocked reading from the queue
		# so the shutdown will not be immediate
		self.running = False


	def _classify (self, client_ip, method, url):
		squid = '%s %s - %s -' % (url, client_ip, method)
		#logger.info('worker %s' % self.wid, 'sending to classifier: [%s]' % squid)
		try:
			self.process.stdin.write(squid + os.linesep)
			response = self.process.stdout.readline().strip()

			if not response:
				classification, data = 'permit', None

			elif response.startswith('http://'):
				response = response[7:]

				if response == url:
					classification, data = 'permit', None
				elif response.startswith(url.split('/', 1)[0]+'/'):
					classification, data = 'rewrite', ('/'+response.split('/', 1)[1]) if '/' in url else ''
				else:
					classification, data = 'redirect', 'http://' + response

{'fraction_non_alphanumeric': 0.11445170660856935, 'fraction_numerical': 0.006100217864923747, 'mean_word_length': 3.045828437132785, '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': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1983985', 'n_tokens_mistral': 2400, 'n_tokens_neox': 2077, 'n_words': 659}