вот тест того, как программа должна работать и как возникает утечка памяти:
Код: Выделить всё
for i in $(seq 1 5000); do curl http://localhost:8023/nonexisting-path$1 --proxy http://localhost:8081 --insecure --location; done > /dev/null- инструмент работает на порту 8082 на локальном хосте, прокси-сервер работает на порту 8081, а другой тестовый сервер работает на порту 8023.
- прокси-сервер, когда он отправляет и получает ответ, он разбивает тело запроса и тело ответа на части, хеширует их, а затем отправляет на сервер с помощью POST http://localhost:8082/store / конечная точка.
- сервер сохраняет поток в приведенном ниже коде, используя функцию store_flow и метод __setitem__.
- когда я хочу получить потоки, я использую метод __getitem__ для восстановления фрагментированного потока.
- когда я запускаю приведенную выше команду Curl для сохранения 5000 на сервере (размер ответа 404 для конечной точки nonexisting-path{i} составляет около 250 КБ), сервер использует не более 150 МБ памяти, и потоки перестраиваются, как и ожидалось.
- но когда я пытаюсь получить все 5000 потоков с сервера с помощью GET http://localhost:8082/getContent/, объем памяти достигает 1,2 ГБ, просто запустив метод __getitem__ для каждого поток, возврат всех потоков усугубляет проблему, но при выходе из функции он возвращается обратно к 1,2 ГБ.
Я также пробовал вручную собирать мусор после каждого вызова на __getitem__, а память остается прежней.
вот код MVP, содержащий ту же проблему:
Код: Выделить всё
import datetime
import base64
import operator
import xxhash
import base64
from functools import reduce
from fastapi import FastAPI
from fastapi.requests import Request
from fastapi.responses import JSONResponse
def reduce_list_to_int(int_list:list[int]):
""" reduce a list of ints to a single integer """
return reduce(operator.xor, int_list, 0)
class HTTPFlowStorage:
""" efficiently stores HTTP flows that are mostly identical """
def __init__(self):
self._flows = dict()
self._hashed_content = dict()
self._hashed_headers = dict()
self._content_hashes_hashes = dict()
self._headers_hashes_hashes = dict()
def __getitem__(self, ts:str):
if not self._flows.get(ts):
return None
flow = dict(self._flows[ts])
hashed_bodies = self._hashed_content
collected_req = []
collected_res = []
# collect request and response bodies from their hashes
req = flow['request-body']
res = flow['response-body']
for i in req:
collected_req.append(hashed_bodies[i])
for i in res:
collected_res.append(hashed_bodies[i])
# collect body and encode as base64 to help transfer bytes
flow['request-body'] = base64.b64encode(b''.join(collected_req)).decode('ascii')
flow['response-body'] = base64.b64encode(b''.join(collected_res)).decode('ascii')
# collect hashed headers
new_headers = {}
for h in flow['request-headers']:
v = flow['request-headers'][h]
new_headers[h] = self._hashed_headers[v]
flow['request-headers'] = new_headers
new_headers = {}
for h in flow['response-headers']:
v = flow['response-headers'][h]
new_headers[h] = self._hashed_headers[v]
flow['response-headers'] = new_headers
return flow
def __setitem__(self, ts:str, flow:dict):
def _add_new_content_hashes(hashes, content):
"""Stores newly found hashes and their content in the hashes object """
not_matched_res = 0
for h in content:
hi = int(h)
if not hashes.get(hi):
hashes[hi] = base64.urlsafe_b64decode(content[h])
not_matched_res += 1
def _add_new_headers_hashes(hashes, hdrs):
"""Stores newly found headers in the hashes object and returns
the new headers """
new_headers = dict()
for hdr in hdrs:
hv = hdrs[hdr]
hs = int(xxhash.xxh3_64_digest(hv.encode('utf-8')).hex(), 16)
if hs not in hashes:
hashes[hs] = hv
new_headers[hdr] = hs
return new_headers
# include new hashes in the hashed_content
req = flow['request-body']
res = flow['response-body']
_add_new_content_hashes(self._hashed_content, req)
_add_new_content_hashes(self._hashed_content, res)
flow['request-body'] = list(int(i) for i in req.keys())
flow['response-body'] = list(int(i) for i in res.keys())
# hash and store headers
req_headers = flow['request-headers']
res_headers = flow['response-headers']
nrh = _add_new_headers_hashes(self._hashed_headers, req_headers)
nrsh =_add_new_headers_hashes(self._hashed_headers, res_headers)
flow['request-headers'] = nrh
flow['response-headers'] = nrsh
# further hash the bodies content and check if there is a
# body extactly like it in previous bodies and use it instead of
# new body
hb_res = reduce_list_to_int(flow['response-body'])
hb_req = reduce_list_to_int(flow['request-body'])
if hb_res in self._content_hashes_hashes:
flow['response-body'] = self._content_hashes_hashes[hb_res]
else :
self._content_hashes_hashes[hb_res] = flow['response-body']
if hb_req in self._headers_hashes_hashes:
flow['request-body'] = self._headers_hashes_hashes[hb_req]
else :
self._headers_hashes_hashes[hb_req] = flow['request-body']
self._flows[ts] = flow
def values(self):
return list( self[i] for i in self._flows)
def __len__(self):
return len(self._flows)
CACHE = HTTPFlowStorage()
app = FastAPI()
@app.post("/store/")
async def store_flow(flow:dict):
"""Stores a flow from the proxy in the `CACHE` object."""
timestamp = datetime.datetime.now().timestamp()
flow['timestamp'] = timestamp
flow['index'] = len(CACHE) +1
CACHE[str(timestamp)] = flow
@app.get("/getContent/")
async def get_history(req:Request):
flows = CACHE.values()
# returning None here would create the same problem.
# not to return the all flows increases the memory leak even more ??
return JSONResponse(flows[0])
версия Python: python3.10.12
версия fastapi: fastapi==0.111.0
версия uvicorn: 0.29.0
Подробнее здесь: https://stackoverflow.com/questions/785 ... understand
