import math
import os
import time
import unittest
import subprocess as sp
import asyncio
import requests_async as requests
from h11 import RemoteProtocolError
from threading import Lock
SERVER_PATH = os.path.realpath(os.path.join(os.path.curdir, '..', 'server'))
DYNAMIC_REQ_TIME = 0.2
DEFAULT_PORT = 8080
DEFAULT_THREAD_COUNT = 3
DEFAULT_QUEUE_SIZE = 7
DEFAULT_STATIC_PAGE = 'home.html'
DEFAULT_DYNAMIC_PAGE = 'output.cgi'
SERVER_ADDRESS = 'localhost' # 'localhost'
class RequestResult:
def __init__(self, req_ind, res=None, e=None):
self.res = res
self.e = e
self.req_ind = req_ind
def has_exception(self):
return self.e is not None
def is_exception_of_type(self, exc_type):
return isinstance(self.e, exc_type)
class RequestsTest(unittest.TestCase):
# noinspection HttpUrlsUsage
def __init__(self, *args, queue_size=DEFAULT_QUEUE_SIZE, thread_count=DEFAULT_THREAD_COUNT, policy='dt', **kwargs):
super().__init__(*args, **kwargs)
self.req_ind_mutex = Lock()
self.dyn_url = f'http://{SERVER_ADDRESS}:{DEFAULT_PORT}/{DEFAULT_DYNAMIC_PAGE}'
self.static_url = f'http://{SERVER_ADDRESS}:{DEFAULT_PORT}/{DEFAULT_STATIC_PAGE}'
self.not_found_url = f'http://{SERVER_ADDRESS}:{DEFAULT_PORT}/not_found'
self.forbidden_url = f'http://{SERVER_ADDRESS}:{DEFAULT_PORT}/forbidden.cgi'
self.queue_size = queue_size
self.max_reqs = self.queue_size
self.server_path = SERVER_PATH
self.thread_count = thread_count
self.policy = policy
if policy == 'random':
self.per_drop_size = math.ceil(0.25 * (self.queue_size-self.thread_count))
elif policy in ['dt', 'dh']:
self.per_drop_size = 1
elif policy == 'block':
self.per_drop_size = 0
self.last_req_index = 0
def setUp(self):
print('Setting up test:')
print(f'\tthread: {self.thread_count}')
print(f'\tqueue size: {self.queue_size}')
print(f'\tpolicy: {self.policy}')
os.chdir(os.path.dirname(self.server_path))
self.server = sp.Popen([self.server_path, f'{DEFAULT_PORT}', f'{self.thread_count}', f'{self.queue_size}', self.policy])
print(f'Server status is: {self.server.poll()}')
# input('Confirm open port and hit RETURN')
# print('')
def tearDown(self) -> None:
self.server.terminate()
async def make_req(self, url, method='get'):
try:
self.req_ind_mutex.acquire()
req_ind = self.last_req_index + 1
self.last_req_index += 1
self.req_ind_mutex.release()
req_headers = {'req_index': f'{req_ind}'}
# arrival_time = time.time() * 1000 # in milliseconds
req_ind = self.last_req_index
if method == 'get':
req = requests.get(url, headers=req_headers)
elif method == 'post':
req = requests.post(url, headers=req_headers)
elif method == 'delete':
req = requests.delete(url, headers=req_headers)
else:
self.fail('Unknown request method')
# print(f'Awaiting request {req_ind}')
response = await req
headers = response.headers
for header in headers:
if header.lower().startswith('stat'):
headers[header] = headers[header][1:]
# response_time = time.time() * 1000
# self.assertAlmostEqual(arrival_time, float(response.headers['stat-req-arrival']), delta=min(500 * DYNAMIC_REQ_TIME, (response_time - arrival_time) * 0.2))
except Exception as e:
r = RequestResult(req_ind=req_ind, e=e)
return r
else:
r = RequestResult(req_ind=req_ind, res=response)
return r
async def make_requests(self, url, total_reqs):
print(f'Requesting url: {self.dyn_url}')
self.last_req_index = 0
tasks = []
fail_expected_tasks = []
thread_stats = [{'count': 0, 'dyn': 0, 'static': 0} for _ in range(self.thread_count)]
expected_error_count = total_reqs - self.max_reqs + ((-(total_reqs - self.max_reqs)) % self.per_drop_size)
expected_average_dispatch = DYNAMIC_REQ_TIME * float(total_reqs - expected_error_count - min(self.thread_count, self.max_reqs)) / self.thread_count / 2
if self.policy == 'random':
for _ in range(total_reqs):
task = asyncio.ensure_future(self.make_req(url))
tasks.append(task)
elif self.policy == 'dt':
for _ in range(total_reqs - expected_error_count):
task = asyncio.ensure_future(self.make_req(url))
tasks.append(task)
for _ in range(expected_error_count):
task = asyncio.ensure_future(self.make_req(url))
fail_expected_tasks.append(task)
elif self.policy == 'dh':
for _ in range(expected_error_count):
task = asyncio.ensure_future(self.make_req(url))
fail_expected_tasks.append(task)
for _ in range(total_reqs - expected_error_count):
task = asyncio.ensure_future(self.make_req(url))
tasks.append(task)
responses = await asyncio.gather(*tasks, *fail_expected_tasks, return_exceptions=True)
responses = sorted(responses, key=lambda x: x.req_ind)
error_count = 0
total_dispatch = 0
for res in responses:
if res.is_exception_of_type(RemoteProtocolError):
error_count += 1
continue
elif res.has_exception():
raise res.e
res = res.res
total_dispatch += float(res.headers['stat-req-dispatch'])
count, dyn, static = int(res.headers['stat-thread-count']), int(res.headers['stat-thread-dynamic']), int(res.headers['stat-thread-static'])
tid = int(res.headers['stat-thread-id'])
thread_stats[tid]['count'] = max(count, thread_stats[tid]['count'])
thread_stats[tid]['dyn'] = max(dyn, thread_stats[tid]['dyn'])
thread_stats[tid]['static'] = max(static, thread_stats[tid]['static'])
self.assertEqual(count, dyn, f'Mismatch total request and dynamic request. Total: {count}. Dynamic: {dyn}')
self.assertEqual(static, 0, f'Unexpected static requests count. Expected {0}. Actual {static}')
# self.assertAlmostEqual(float(res.headers['stat-req-arrival']), arrival_time)
total_count = total_dyn = 0
for stat in thread_stats:
total_count += stat['count']
total_dyn += stat['dyn']
print(f'Requests succeeded: {total_count}')
print(f'Requests failed: {error_count}')
self.assertEqual(total_count, total_reqs - error_count)
self.assertEqual(total_count, total_dyn)
self.assertEqual(expected_error_count, error_count, f'Unexpected error count. Expected {expected_error_count}. Actual: {error_count}')
self.assertNotEqual(0, total_reqs - error_count, 'No request succeeded')
average_dispatch = total_dispatch / float(total_reqs - error_count)
self.assertAlmostEqual(expected_average_dispatch, average_dispatch, delta=max(1, expected_average_dispatch * 0.3),
msg=f'Unexpected average dispatch time. Expected: {expected_average_dispatch}. Actual: {average_dispatch}')
# TODO: Fix check. Should check that the correct requests failed for each policy type
# if self.policy == 'dh':
# for res in responses[self.thread_count:self.thread_count + error_count]:
# self.assertIsInstance(res.e, RemoteProtocolError)
# elif self.policy == 'dt':
# for res in responses[-error_count:]:
# self.assertIsInstance(res.e, RemoteProtocolError)
class TestDropTailRequests(RequestsTest):
def __init__(self, *args, **kwargs):
super().__init__(*args, policy='dt', **kwargs)
def test_drop_single(self):
asyncio.run(self.make_requests(self.dyn_url, self.max_reqs + 1))
def test_drop_double_queue_size(self):
asyncio.run(self.make_requests(self.dyn_url, self.max_reqs + self.queue_size * 2))
class TestDropHeadRequests(RequestsTest):
def __init__(self, *args, **kwargs):
super().__init__(*args, policy='dh', **kwargs)
def test_drop_single(self):
asyncio.run(self.make_requests(self.dyn_url, self.max_reqs + 1))
def test_drop_double_queue_size(self):
asyncio.run(self.make_requests(self.dyn_url, self.max_reqs + self.queue_size * 2))
class TestDropRandomRequests(RequestsTest):
def __init__(self, *args, **kwargs):
super().__init__(*args, queue_size=16, policy='random', **kwargs)
def test_single_drop_random(self):
asyncio.run(self.make_requests(self.dyn_url, self.max_reqs + 1))
def test_double_drop_random(self):
asyncio.run(self.make_requests(self.dyn_url, self.max_reqs + 2 * int(0.25 * self.queue_size)))
def test_no_drop(self):
asyncio.run(self.make_requests(self.dyn_url, self.max_reqs))
class TestMultiThreaded(RequestsTest):
def __init__(self, *args, **kwargs):
super().__init__(*args, queue_size=80, **kwargs)
def test_time_full_queue(self):
start_time = time.time()
req_count = self.max_reqs
asyncio.run(self.make_requests(self.dyn_url, req_count))
run_time = time.time() - start_time
expected_runtime = math.ceil(req_count / float(self.thread_count)) * DYNAMIC_REQ_TIME
# This is optimal so it must be greater
self.assertGreater(run_time, expected_runtime)
self.assertLess(run_time, expected_runtime * 2)
def test_better_with_more_threads(self):
start_time = time.time()
req_count = self.max_reqs
asyncio.run(self.make_requests(self.dyn_url, req_count))
few_threads_run_time = time.time() - start_time
self.server.terminate()
self.thread_count *= 3
self.setUp()
start_time = time.time()
asyncio.run(self.make_requests(self.dyn_url, req_count))
more_threads_run_time = time.time() - start_time
self.assertTrue(2 * more_threads_run_time < few_threads_run_time < 3 * more_threads_run_time, "Performance doesn't scale as expected with amount of threads")
class TestStatusCodes(RequestsTest):
def __init__(self, *args, **kwargs):
super().__init__(*args, thread_count=1, queue_size=1, **kwargs)
async def _make_req(self, url, expected_status, stat_map, method='get'):
task = asyncio.ensure_future(self.make_req(url, method=method))
res = await asyncio.ensure_future(task)
headers = res.res.headers
for k in stat_map:
self.assertIn(k, headers)
if stat_map[k] is not None:
self.assertAlmostEqual(float(headers[k]), stat_map[k], delta=0.03, msg=f'Unexpected value for {k}. Expected: {stat_map[k]}. Actual: {headers[k]}')
self.assertEqual(expected_status, res.res.status_code, f'Unexpected status code. Expected: {expected_status}. Actual: {res.res.status_code}')
def test_404(self):
stat_map = {
'stat-req-arrival': None,
'stat-req-dispatch': 0.00004,
'stat-thread-id': 0,
'stat-thread-count': 1,
'stat-thread-static': 0,
'stat-thread-dynamic': 0
}
asyncio.run(self._make_req(self.not_found_url, 404, stat_map))
def test_dynamic(self):
stat_map = {
'stat-req-arrival': None,
'stat-req-dispatch': 0.00004,
'stat-thread-id': 0,
'stat-thread-count': 1,
'stat-thread-static': 0,
'stat-thread-dynamic': 1
}
asyncio.run(self._make_req(self.dyn_url, 200, stat_map))
def test_static(self):
stat_map = {
'stat-req-arrival': None,
'stat-req-dispatch': 0.00004,
'stat-thread-id': 0,
'stat-thread-count': 1,
'stat-thread-static': 1,
'stat-thread-dynamic': 0
}
asyncio.run(self._make_req(self.static_url, 200, stat_map))
def test_forbidden(self):
stat_map = {
'stat-req-arrival': None,
'stat-req-dispatch': 0.00004,
'stat-thread-id': 0,
'stat-thread-count': 1,
'stat-thread-static': 0,
'stat-thread-dynamic': 0
}
asyncio.run(self._make_req(self.forbidden_url, 403, stat_map))
def test_post(self):
stat_map = {
'stat-req-arrival': None,
'stat-req-dispatch': 0.00004,
'stat-thread-id': 0,
'stat-thread-count': 1,
'stat-thread-static': 0,
'stat-thread-dynamic': 0
}
asyncio.run(self._make_req(self.static_url, 501, stat_map, method='post'))
class TestQueueSmallerThan(RequestsTest):
def __init__(self, *args, **kwargs):
super().__init__(*args, thread_count=5, queue_size=2, **kwargs)
def test_small_queue_enough_threads(self):
asyncio.run(self.make_requests(url=self.dyn_url, total_reqs=self.thread_count))
if __name__ == '__main__':
unittest.main()
Comments