<commit_before>#!/usr/bin/env python

import sys
import asyncore, asynchat
import re, socket, time, threading, os
from . import http_request, counter
from aquests.protocols.http import http_util, http_date
from aquests.lib.athreads import threadlib
from skitai import lifetime
from aquests.lib import producers, compressors
from collections import deque
import signal
import ssl
import skitai
from hashlib import md5

PID = []
ACTIVE_WORKERS = 0
SURVAIL = True
EXITCODE = 0
DEBUG = False

	
#-------------------------------------------------------------------
# server channel
#-------------------------------------------------------------------
class http_channel (asynchat.async_chat):	
	current_request = None
	channel_count = counter.counter ()
	closed = False
	is_rejected = False
	
	zombie_timeout = 2
	fifo_class = deque
	
	def __init__ (self, server, conn, addr):
		self.channel_number = http_channel.channel_count.inc ()
		self.request_counter = counter.counter()
		self.bytes_out = counter.counter()
		self.bytes_in = counter.counter()
		
		#asynchat.async_chat.__init__ (self, conn)
		self.ac_in_buffer = b''
		self.incoming = []
		self.producer_fifo = self.fifo_class ()
		asyncore.dispatcher.__init__(self, conn)
		
		self.server = server
		self.addr = addr		
		self.set_terminator (b'\r\n\r\n')
		self.in_buffer = b''
		self.creation_time = int (time.time())
		self.event_time = int (time.time())
		self.__history = []
		self.__sendlock = None
		self.producers_attend_to = []
		self.things_die_with = []
		
	def use_sendlock (self):
		self.__sendlock  = threading.Lock ()
		self.initiate_send = self._initiate_send_ts
			
	def get_history (self):
		return self.__history
			
	def reject (self):
		self.is_rejected = True		
	
	def _initiate_send_ts (self):
		lock = self.__sendlock
		lock.acquire ()
		try:
			asynchat.async_chat.initiate_send (self)		
			try: is_working = self.producer_fifo.working ()
			except AttributeError: 	is_working = len (self.producer_fifo)
		finally:	
			lock.release ()			
		if not is_working:
			self.done_request ()		
		
	def initiate_send (self):
		asynchat.async_chat.initiate_send (self)		
		try: is_working = self.producer_fifo.working ()
		except AttributeError: 	is_working = len (self.producer_fifo)
		if not is_working:
			self.done_request ()		
			
	def readable (self):		
		return not self.is_rejected and asynchat.async_chat.readable (self)
		
	def issent (self):
		return self.bytes_out.as_long ()
		
	def __repr__ (self):
		ar = asynchat.async_chat.__repr__(self) [1:-1]
		return '<%s channel#: %s requests:%s>' % (
				ar,
				self.channel_number,
				self.request_counter
				)
		
	def clean_shutdown_control (self, phase, time_in_this_phase):
		if phase == 3:
			self.reject ()
			if self.writable ():
				return 1
			self.close ()
			return 0
	      		
	def isconnected (self):
		return self.connected
	
	def handle_timeout (self):
		self.log ("killing zombie channel %s" % ":".join (map (str, self.addr)))
		self.close ()
	
	def set_timeout (self, timeout):
		self.zombie_timeout = timeout		
			
	def attend_to (self, thing):
		if not thing: return
		self.producers_attend_to.append (thing)
	
	def die_with (self, thing, tag):
		if not thing: return
		self.things_die_with.append ((thing, tag))
		
	def done_request (self):	
		self.producers_attend_to = [] # all producers are finished
							
	def send (self, data):
		print	("SEND", repr (data), self.get_terminator ())
		self.event_time = int (time.time())
		result = asynchat.async_chat.send (self, data)		
		self.server.bytes_out.inc (result)
		self.bytes_out.inc (result)
		return result
	
	def recv (self, buffer_size):
		self.event_time = int (time.time())		
		try:
			result = asynchat.async_chat.recv (self, buffer_size)
			lr = len (result)
			self.server.bytes_in.inc (lr)
			self.bytes_in.inc (lr)
			if not result:
				self.handle_close ()
				return b""		
			print	("RECV", repr(result), self.get_terminator ())
			return result
			
		except MemoryError:
			lifetime.shutdown (1, 1.0)
				
	def collect_incoming_data (self, data):
		#print ("collect_incoming_data", repr (data [:180]), self.current_request)
		if self.current_request:
			self.current_request.collect_incoming_data (data)
		else:
			self.in_buffer = self.in_buffer + data
				
	def found_terminator (self):
		if self.is_rejected:
			return
			
		if self.current_request:			
			self.current_request.found_terminator()
			if DEBUG:
				self.__history.append ("REQUEST DONE")
				self.__history = self.__history [-30:]
			
		else:
			header = self.in_buffer<commit_msg>fix (remove) chunked response when 1.0<commit_after>#!/usr/bin/env python

import sys
import asyncore, asynchat
import re, socket, time, threading, os
from . import http_request, counter
from aquests.protocols.http import http_util, http_date
from aquests.lib.athreads import threadlib
from skitai import lifetime
from aquests.lib import producers, compressors
from collections import deque
import signal
import ssl
import skitai
from hashlib import md5

PID = []
ACTIVE_WORKERS = 0
SURVAIL = True
EXITCODE = 0
DEBUG = False

	
#-------------------------------------------------------------------
# server channel
#-------------------------------------------------------------------
class http_channel (asynchat.async_chat):	
	current_request = None
	channel_count = counter.counter ()
	closed = False
	is_rejected = False
	
	zombie_timeout = 2
	fifo_class = deque
	
	def __init__ (self, server, conn, addr):
		self.channel_number = http_channel.channel_count.inc ()
		self.request_counter = counter.counter()
		self.bytes_out = counter.counter()
		self.bytes_in = counter.counter()
		
		#asynchat.async_chat.__init__ (self, conn)
		self.ac_in_buffer = b''
		self.incoming = []
		self.producer_fifo = self.fifo_class ()
		asyncore.dispatcher.__init__(self, conn)
		
		self.server = server
		self.addr = addr		
		self.set_terminator (b'\r\n\r\n')
		self.in_buffer = b''
		self.creation_time = int (time.time())
		self.event_time = int (time.time())
		self.__history = []
		self.__sendlock = None
		self.producers_attend_to = []
		self.things_die_with = []
		
	def use_sendlock (self):
		self.__sendlock  = threading.Lock ()
		self.initiate_send = self._initiate_send_ts
			
	def get_history (self):
		return self.__history
			
	def reject (self):
		self.is_rejected = True		
	
	def _initiate_send_ts (self):
		lock = self.__sendlock
		lock.acquire ()
		try:
			asynchat.async_chat.initiate_send (self)		
			try: is_working = self.producer_fifo.working ()
			except AttributeError: 	is_working = len (self.producer_fifo)
		finally:	
			lock.release ()			
		if not is_working:
			self.done_request ()		
		
	def initiate_send (self):
		asynchat.async_chat.initiate_send (self)		
		try: is_working = self.producer_fifo.working ()
		except AttributeError: 	is_working = len (self.producer_fifo)
		if not is_working:
			self.done_request ()		
			
	def readable (self):		
		return not self.is_rejected and asynchat.async_chat.readable (self)
		
	def issent (self):
		return self.bytes_out.as_long ()
		
	def __repr__ (self):
		ar = asynchat.async_chat.__repr__(self) [1:-1]
		return '<%s channel#: %s requests:%s>' % (
				ar,
				self.channel_number,
				self.request_counter
				)
		
	def clean_shutdown_control (self, phase, time_in_this_phase):
		if phase == 3:
			self.reject ()
			if self.writable ():
				return 1
			self.close ()
			return 0
	      		
	def isconnected (self):
		return self.connected
	
	def handle_timeout (self):
		self.log ("killing zombie channel %s" % ":".join (map (str, self.addr)))
		self.close ()
	
	def set_timeout (self, timeout):
		self.zombie_timeout = timeout		
			
	def attend_to (self, thing):
		if not thing: return
		self.producers_attend_to.append (thing)
	
	def die_with (self, thing, tag):
		if not thing: return
		self.things_die_with.append ((thing, tag))
		
	def done_request (self):	
		self.producers_attend_to = [] # all producers are finished
							
	def send (self, data):
		#print	("SEND", repr (data), self.get_terminator ())
		self.event_time = int (time.time())
		result = asynchat.async_chat.send (self, data)		
		self.server.bytes_out.inc (result)
		self.bytes_out.inc (result)
		return result
	
	def recv (self, buffer_size):
		self.event_time = int (time.time())		
		try:
			result = asynchat.async_chat.recv (self, buffer_size)
			lr = len (result)
			self.server.bytes_in.inc (lr)
			self.bytes_in.inc (lr)
			if not result:
				self.handle_close ()
				return b""		
			#print	("RECV", repr(result), self.get_terminator ())
			return result
			
		except MemoryError:
			lifetime.shutdown (1, 1.0)
				
	def collect_incoming_data (self, data):
		#print ("collect_incoming_data", repr (data [:180]), self.current_request)
		if self.current_request:
			self.current_request.collect_incoming_data (data)
		else:
			self.in_buffer = self.in_buffer + data
				
	def found_terminator (self):
		if self.is_rejected:
			return
			
		if self.current_request:			
			self.current_request.found_terminator()
			if DEBUG:
				self.__history.append ("REQUEST DONE")
				self.__history = self.__history [-30:]
			
		else:
			header = self.in_buffer
{'fraction_non_alphanumeric': 0.13611020006559527, 'fraction_numerical': 0.003935716628402755, 'mean_word_length': 3.1449932034435886, 'pattern_counts': {'":': 2, '<': 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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2201970', 'n_tokens_mistral': 3590, 'n_tokens_neox': 3100, 'n_words': 742}