Как сделать, чтобы процесс в multiprocessing не завершался автоматически?

Question

[Вячеслав Грачунов]

Такой код:

def outgoing():
	BASEDIR = os.path.dirname(os.path.realpath(__file__))

	pool = PooledDB(creator=pymysql,
	 
	               host='127.0.0.1',

	               user='root',

	               password='password',

	               database='tgadmin_test',

	               autocommit=True,

	               charset='utf8mb4',

	               cursorclass=pymysql.cursors.DictCursor,

	               blocking=False,

	               maxconnections=20)


	db = pool.connection()
	cursor = db.cursor()

	sql = "SELECT * FROM tgmanager_applicationusers WHERE banned_at IS NULL ORDER BY RAND() LIMIT 1"
	cursor.execute(sql)

	user1 = cursor.fetchone()
	proxy_lst = user1['proxy'].split(':')


	# VoIPServerConfig.set_bitrate_config(80000, 100000, 60000, 5000, 5000)
	client = Client(session_name=user1['phone'], api_id=user1['api_id'], api_hash=user1['api_hash'], ipv6=False, proxy=dict(hostname=str(proxy_lst[0]), port=int(proxy_lst[1]), username=str(proxy_lst[2]), password=str(proxy_lst[3])), app_version="0.22.0.1205-arm64-v8a", device_model=user1['device'], system_version=user1['os'], workdir="/home/qwentor/share/autoreg/sessions")
	client.start()


	service = VoIPNativeIOService(client, receive_calls=False)
	start_call = service.start_call('@messengerN1')
	mp3 = os.listdir(os.path.join(BASEDIR, 'speech'))
	item = random.choice(mp3)
	print('MP3##############################', item)
	speech = int(item.split('/')[-1].split(".")[0])
	print('SPEECH', speech)
	sql = """SELECT speech FROM tgmanager_dialogues WHERE id = {id}""".format(id=speech)
	cursor.execute(sql)
	res = cursor.fetchone()['speech']
	print(res)

	stream = ffmpeg.input(os.path.join(BASEDIR, 'speech', item))
	print("S1")
	stream = ffmpeg.output(stream, os.path.join(BASEDIR, 'PCM', item.split('.')[0]+'.raw'), f='s16le', ac=1, ar=48000, acodec='pcm_s16le')
	ffmpeg.run(stream)
	print("S2")
	sleep(5)
	print("S3")
	start_call.play(os.path.join(BASEDIR, 'PCM', item.split('.')[0]+'.raw'))
	print("S4")
	start_call.set_output_file(os.path.join(BASEDIR, 'outgoing.raw'))
	print("S5")

Если его запустить как
outgoing()
то все номально, а если как

proc2 = multiprocessing.Process(target=outgoing, name='outgoing')
proc2.start()

то программа завершается по достижению выполнения print("S5"), хотя там должен идти звонок через tgvoip, т.е. она должна ждать снятия трубки и проболтать текст, а не ждет
В чем дело и можно ли сделать, чтобы программа не завершалась пока не нужно

Answer 1

[Сергей Тихонов]

Скорее всего какие-то из вызовов внутри outgoing() создают новые потоки, соответственно, пока они не завершатся, интерпретатор питона не останавливается, несмотря на то, что основной поток закончил свою работу.
Как именно это интерферирует с multiprocessing - это вопрос глубокой отладки, но я бы сначала попробовал сделать так, чтобы выход из outgoing() завершал процесс питона без мультипроцессинга.

Comments

[Вячеслав Грачунов]

Там звонок идет голосовой через телеграм (голос передается в трубку в raw формате аудио).
И цепочка используемых либ следующая:
pytgvoip-pyrogram --> pytgvoip --> libtgvoip

Подозреваю, что где-то нет обратной связи и данные для звонка передаются, а дальше все, уже работает только c++ либа. При обычном запуске она просто запускается и работает, а при мультипроцессинге питон считает, что процесс полностью выполнен после передачи данных в либу, а не после завершения звонка, и завершает его. Пока решил эту проблему тупым бесконечным циклом в конце функции. Но если есть более элегантное решение - буду рад услышать

Answer 2

[Ivan Yakushenko]

proc2.join()
На этом моменте скрипт будет висеть, пока не завершится процесс proc2

Comments

[Вячеслав Грачунов]

неа, джойн пробовал - однофигственно завершается раньше времени

[Ivan Yakushenko]

Вячеслав Грачунов, значит процесс outgoing завершается. Если что, в отдельном процессe любые ошибки, как и принты не отображаются в основном процессе, так что вполне возможно, что где-то случается ошибка о которой ты не знаешь.

[Вячеслав Грачунов]

Ivan Yakushenko, принты точно отображаются, вижу их в консоли. Сейчас сделал внутри outgoing() намеренно деление на ноль - ошибка тоже в консоль попала

[Ivan Yakushenko]

Вячеслав Грачунов, покажи полный код функции outgoing и как ты её запускаешь.

[Вячеслав Грачунов]

import multiprocessing
import pyrogram
from pyrogram import Client, MessageHandler, Filters
from pyrogram.errors.exceptions.bad_request_400 import *
from pyrogram.errors.exceptions.unauthorized_401 import *
from tgvoip import VoIPServerConfig
from tgvoip_pyrogram import VoIPFileStreamService, VoIPIncomingFileStreamCall, VoIPNativeIOService, VoIPIncomingNativeIOCall
import ffmpeg
import os
import pymysql
from DBUtils.PooledDB import PooledDB
import random
from time import sleep
import soothingsounds as ss
from pathlib import Path
from pprint import pprint
from mutagen.mp3 import MP3

def incoming():
	flag = True
	BASEDIR = os.path.dirname(os.path.realpath(__file__))

	pool = PooledDB(creator=pymysql,
	 
	               host='127.0.0.1',

	               user='root',

	               password='password',

	               database='tgadmin_test',

	               autocommit=True,

	               charset='utf8mb4',

	               cursorclass=pymysql.cursors.DictCursor,

	               blocking=False,

	               maxconnections=20)


	db = pool.connection()
	cursor = db.cursor()

	sql = "SELECT * FROM tgmanager_applicationusers WHERE banned_at IS NULL ORDER BY RAND() LIMIT 1"
	cursor.execute(sql)

	user1 = cursor.fetchone()
	proxy_lst = user1['proxy'].split(':')


	# VoIPServerConfig.set_bitrate_config(80000, 100000, 60000, 5000, 5000)
	client = Client(session_name='[phone]', api_id=21724, api_hash='3e0cb5efcd52300aec5994fdfc5bdc16', ipv6=False, proxy=dict(hostname=str(proxy_lst[0]), port=int(proxy_lst[1]), username=str(proxy_lst[2]), password=str(proxy_lst[3])), app_version="0.22.0.1205-arm64-v8a", device_model='Xiaomi Mi 9', system_version='Android 10 Q (29)')
	client.start()

	service = VoIPNativeIOService(client, receive_calls=True)
	@service.on_incoming_call
	def process_call(call: VoIPIncomingNativeIOCall):  # use VoIPIncomingNativeIOCall for native I/O
		call.accept()
		call.set_output_file(os.path.join(BASEDIR, 'incoming.raw'))
		mp3 = os.listdir(os.path.join(BASEDIR, 'speech'))
		item = random.choice(mp3)
		print('MP3##############################', item)
		pprint(item)
		speech = int(item.split('/')[-1].split(".")[0])
		print('SPEECH', speech)
		sql = """SELECT speech FROM tgmanager_dialogues WHERE id = {id}""".format(id=speech)
		cursor.execute(sql)
		res = cursor.fetchone()['speech']
		print(res)

		stream = ffmpeg.input(os.path.join(BASEDIR, 'speech', item))
		stream = ffmpeg.output(stream, os.path.join(BASEDIR, 'PCM', item.split('.')[0]+'.raw'), f='s16le', ac=1, ar=48000, acodec='pcm_s16le')
		ffmpeg.run(stream)
		# sleep(15)
		call.play(os.path.join(BASEDIR, 'PCM', item.split('.')[0]+'.raw'))
		print("BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB")

		audio = MP3(os.path.join(BASEDIR, 'speech', item))
		print('DURATION', audio.info.length)



		@call.on_call_ended
		def call_ended(call):
			nonlocal client
			nonlocal flag
			client.stop()
			flag = False

	while flag:
		pass


def outgoing():
	flag = True
	BASEDIR = os.path.dirname(os.path.realpath(__file__))

	pool = PooledDB(creator=pymysql,
	 
	               host='127.0.0.1',

	               user='root',

	               password='password',

	               database='tgadmin_test',

	               autocommit=True,

	               charset='utf8mb4',

	               cursorclass=pymysql.cursors.DictCursor,

	               blocking=False,

	               maxconnections=20)


	db = pool.connection()
	cursor = db.cursor()

	sql = "SELECT * FROM tgmanager_applicationusers WHERE banned_at IS NULL ORDER BY RAND() LIMIT 1"
	cursor.execute(sql)

	user1 = cursor.fetchone()
	proxy_lst = user1['proxy'].split(':')


	# VoIPServerConfig.set_bitrate_config(80000, 100000, 60000, 5000, 5000)
	client = Client(session_name=user1['phone'], api_id=user1['api_id'], api_hash=user1['api_hash'], ipv6=False, proxy=dict(hostname=str(proxy_lst[0]), port=int(proxy_lst[1]), username=str(proxy_lst[2]), password=str(proxy_lst[3])), app_version="0.22.0.1205-arm64-v8a", device_model=user1['device'], system_version=user1['os'], workdir="/home/qwentor/share/autoreg/sessions")
	client.start()


	service = VoIPNativeIOService(client, receive_calls=False)
	call = service.start_call('@messengerN1')

	@call.on_call_accepted 
	def on_call_accepted(call):
		nonlocal client
		nonlocal flag
		print("###################CALL!!!!!!!!###################")
		mp3 = os.listdir(os.path.join(BASEDIR, 'speech'))
		item = random.choice(mp3)
		print('MP3##############################', item)
		pprint(item)
		speech = int(item.split('/')[-1].split(".")[0])
		print('SPEECH', speech)
		sql = """SELECT speech FROM tgmanager_dialogues WHERE id = {id}""".format(id=speech)
		cursor.execute(sql)
		res = cursor.fetchone()['speech']
		print(res)

		stream = ffmpeg.input(os.path.join(BASEDIR, 'speech', item))
		print("S1")
		stream = ffmpeg.output(stream, os.path.join(BASEDIR, 'PCM', item.split('.')[0]+'.raw'), f='s16le', ac=1, ar=48000, acodec='pcm_s16le')
		ffmpeg.run(stream)
		print("S2")
		sleep(5)
		print("S3")
		call.play(os.path.join(BASEDIR, 'PCM', item.split('.')[0]+'.raw'))
		print("S4")
		call.set_output_file(os.path.join(BASEDIR, 'outgoing.raw'))
		print("S5")

		audio = MP3(os.path.join(BASEDIR, 'speech', item))
		print('DURATION', audio.info.length)

		
		# call.stop()
		# print("Call STOPPED!!!!!!!!!!!!!")
		# flag = False



		@call.on_call_ended
		def call_ended(call):
			nonlocal client
			nonlocal flag
			client.stop()
			print("Client STOPPED!!!!!!!!!!!!!")
			flag = False


	while flag:
		pass

proc1 = multiprocessing.Process(target=incoming, name='incoming')
proc1.start()

proc2 = multiprocessing.Process(target=outgoing, args=(), name='outgoing')
proc2.start()

[Вячеслав Грачунов]

Ivan Yakushenko, вот если в конце убрать while flag, то перестает работать

[Ivan Yakushenko]

proc1 = multiprocessing.Process(target=incoming, name='incoming')
proc2 = multiprocessing.Process(target=outgoing, args=(), name='outgoing')

proc1.start()
proc2.start()

proc1.join()
proc2.join()

join не может не работать. Попробуй запусти так:

proc1 = multiprocessing.Process(target=incoming, name='incoming')
proc1.start()

outgoing()

Что-бы убедиться, что сама по себе функция работает и проблема именно в процессе.

[Вячеслав Грачунов]

Ivan Yakushenko, именно так джойны и ставил - нет толку

А так запускать бесполезно - проблема одинаковая в обоих функциях

[Ivan Yakushenko]

Вячеслав Грачунов, ну так значит проблема в функции и она не выполняется до конца. Убирай процессы, запускай в debug-режиме код и построчно смотри как работает программа.

[Вячеслав Грачунов]

Ivan Yakushenko, проблема в том что дебажил, по коду отрабатывает до конца, "S5" выводит и завершается. А не должна, должна работать libtgvoip и в однопроцессном режиме работает, и при while flag работает в мультипроцессном. А вот если while flag убрать - приехали