Я пытаюсь использовать python многопроцессорную обработку в существующей службе - существующий код запускает http-сервер, а затем начинает выполнять код в многопоточном окружении. Перед началом первого потока (daemoni c) определяются некоторые конечные точки API. Принимая во внимание, что после того, как это началось, 2 потока порождены, чтобы бежать в фоновом режиме, толкая данные к месту назначения, как и когда необходимо
ПРИМЕЧАНИЕ , что многопроцессный python запускается "в первом потоке" Проблема появляется через несколько часов, когда запросы к curl к этим конечным точкам начинают зависать, хотя программа, как многопроцессный, так и другие рабочие потоки работают без ошибок
пара моментов для кода ниже:
- Требуется, чтобы последовательность запуска не изменялась, поэтому я должен определить многопроцессорность в первом потоке вместо основного процесса
- используемая номенклатура указывает на назначение
- целевой метод многопроцессорной обработки также потребует контекстного объекта, а также объектов из других модулей в проекте, например: self.anotherprojectmodule_ python
- целевой метод многопроцессорной обработки имеет время True l oop, и "прерывается", если возникает исключение, после чего завершается многопроцессорная обработка в м ain code
- ToCreateObjectof_AnotherModule включает операцию ввода-вывода
- Я уже проверял процессы zomb ie, используя "ps aux", не указывает, что cpu_time перегружен, ошибок засоления не обнаружено
- Если многопроцессорная обработка инициализирована где-либо вне первого потока, конечные точки возвращают ошибку: соединение отказано
- Операции ввода / вывода являются вызовами API, и данные не записываются на диск; журналы создаются с помощью регистратора
Пожалуйста, помогите !! Примите во внимание некоторые объяснения, я довольно новичок в python
class MyService(ServerService):
_th_event = threading.Event()
def __init__(self, context):
self.anotherprojectmodule_python = ToCreateObjectof_AnotherModule(context)
def second_method(self):
self._first_thread = threading.Thread(target=self._first_target, name="first", daemon=True)
define_api_endpoint_on_http_server("/endpoint/url1", self._target_f1)
define_api_endpoint_on_http_server("/endpoint/url2", self._target_f2)
define_api_endpoint_on_http_server("/endpoint/url3", self._target_f3)
def lets_begin_thread(self):
self._first_thread.start()
def then_lets_begin_action(self):
self._worker1_thread = threading.Thread(target=self._some_func, name="worker1", daemon=True)
self._worker2_thread = threading.Thread(target=self._func_with_i/o, name="worker2", daemon=True)
self._separate_worker = multiprocessing.Process(target=self._another_func_with_i/o, name="SeparateMemory", daemon=True)
self._worker1_thread.start()
self._worker2_thread.start()
while not MyService.th_event.is_set():
time.sleep(5)
if not self._worker1_thread.is_alive():
self._worker1_thread = threading.Thread(target=self._some_func, name="worker1", daemon=True)
self._worker1_thread.start()
if not self._worker2_thread.is_alive():
self._worker2_thread = threading.Thread(target=self._func_with_i/o, name="worker2", daemon=True)
self._worker2_thread.start()
if not self._separate_worker.is_alive():
if self.anotherprojectmodule_python.CheckGet_value_from_worker2_target():
self._separate_worker = multiprocessing.Process(target=self._another_func_with_i/o, name="SeparateMemory", daemon=True)
self._separate_worker.start()
log.debug("Exit..")
Добавление 2 трассировок памяти с помощью tracemallo c
Top 10 lines |
'#1: /usr/lib/python3.5/linecache.py:137: 1711.0 KiB | lines = fp.readlines() |
# 2:: 476: 1234,3 КиБ | #3: /url/to/py-pkg/uvloop/__init__.py:19: 506.8 KiB | return Loop() |
# 4: /url/to/py-pkg/json/decoder.py:355: 227,1 КиБ | obj, end = self.scan_once (s, idx) | #5: /url/to/py-pkg/idna/uts46data.py:8204: 154.5 KiB | + _seg_78() |
# 6: /url/to/py-pkg/tracemalloc.py:65: 130,4 КиБ | return (self.size, self.count, self.traceback) | #7: /url/to/py-pkg/site-packages/pyformance/stats/samples.py:74: 64.4 KiB | self.values[priority] = value |
# 8: /url/to/py-pkg/urllib/parse.py:535: 57,1 КиБ | для a в _hexdig для b в _hexdig} | #9: /url/to/py-pkg/site-packages/pika/adapters/select_connection.py:590: 24.1 KiB | events = self._poll.poll(self.get_next_deadline()) |
# 10: /url/to/py-pkg/http/cookiejar.py:1224: 13,4 КиБ | доходность от глубоких значений (объект) | 1828 other: 1146.3 KiB |
Общий выделенный размер: 5269,4 КиБ | No changes for nodes or appstacks or packages. |
Время, затраченное на заполнение register_members: 0.1754981119884178 |
Top 10 lines |
# 1: /url/to/py-pkg/linecache.py:137: 1711,0 КиБ | lines = fp.readlines () | #2: <frozen importlib._bootstrap_external>:476: 1234.3 KiB |
# 3: /url/to/py-pkg/site-packages/uvloop/init.py:19: 506,8 КиБ | возврат L oop () | #4: /url/to/py-pkg/json/decoder.py:355: 228.5 KiB | obj, end = self.scan_once(s, idx) |
# 5: /url/to/py-pkg/idna/uts46data.py:8204: 154,5 КиБ | + _seg_78 () | #6: /url/to/py-pkg/tracemalloc.py:65: 130.4 KiB | return (self.size, self.count, self.traceback) |
# 7: /url/to/py-pkg/site-packages/pyformance/stats/samples.py:74: 64,4 КиБ | self.values [приоритет] = значение | #8: /url/to/py-pkg/urllib/parse.py:535: 57.1 KiB | for a in _hexdig for b in _hexdig} |
# 9: /url/to/py-pkg/site-packages/pika/adapters/select_connection.py:590: 24,1 КиБ | events = self._poll.poll (self.get_next_deadline ()) | #10: /url/to/py-pkg/http/cookiejar.py:1224: 13.4 KiB |
доходность от глубоких значений (obj) | 1837 other: 1166.7 KiB |
Общий выделенный размер: 5291,3 КБ |
Отслеживание сопрограммы: OrderedDict ([('txt', 'cb = [run_until_complete .. ()]>'), ('type', ""), ('Done', False), ('отменено', False),
('stack', [OrderedDict ([('f_code', 'кодовый объект _ bootstrap в 0x7f8246656420, file "/ usr) /lib/python3.5/threading.py ", строка 868 '), (' f_lineno ',' 882 ')]),
OrderedDict ([(' f_code ',' объект кода _bootstrap_inner в 0x7f82466565d0, файл "/usr/lib/python3.5/threading.py", строка 899 '), (' f_lineno ',' 914 ')]),
OrderedDict ([(' f_code ' , 'объект кода выполняется по адресу 0x7f8246656390, файл "/usr/lib/python3.5/threading.py", строка 851'), ('f_lineno', '862')]),
OrderedDict ([('f_code', 'объект кода _start_services в 0x7f8240449a50, файл "/url/to/parent_code_file.py.py", строка 139'), ('f_lineno', '222')]),
OrderedDict ([('f_code', 'объект кода начинается с 0x7f8246064ae0, файл "/ usr / lib / python3. 5 / multiprocessing / process.py ", строка 95 '), (' f_lineno ',' 105 ')]),
OrderedDict ([(' f_code ',' кодовый объект _Popen at 0x7f8246060780, file" / usr / lib / python3 .5 / multiprocessing / context.py ", строка 210 '), (' f_lineno ',' 212 ')]),
OrderedDict ([(' f_code ',' code объект _Popen at 0x7f8246060c90, файл "/usr/lib/python3.5/multiprocessing/context.py", строка 264 '), (' f_lineno ',' 267 ')]),
OrderedDict ( [('f_code', 'кодовый объект init в 0x7f823fab4810, файл "/usr/lib/python3.5/multiprocessing/popen_fork.py", строка 16'), ('f_lineno', ' 20 ')]),
OrderedDict ([(' f_code ',' код объекта _launch at 0x7f821c2799c0, файл "/usr/lib/python3.5/multiprocessing/popen_fork.py", строка 64 ') ), ('f_lineno', '74')]),
OrderedDict ([('f_code', 'кодовый объект _ bootstrap в 0x7f8245feb390, файл "/usr/lib/python3.5 /multiprocessing/process.py ", строка 228 '), (' f_lineno ',' 249 ')]),
OrderedDict ([(' f_code ',' объект кода выполняется по адресу 0x7f8246064a50, файл" / usr) /lib/python3.5/mult iprocessing / process.py ", строка 88 '), (' f_lineno ',' 93 ')]),
OrderedDict ([(' f_code ',' объект кода collect_api_telemetry в 0x7f8240461150, файл" / url / to / code_file.py ", строка 35 '), (' f_lineno ',' 108 ')]),
OrderedDict ([(' f_code ',' объект кода initiate_api_crawl в 0x7f8240330540, файл" / url / to / code_file.py ", строка 257 '), (' f_lineno ',' 278 ')])]), (' исключение ', нет)])