Как параллельно сконвертировать rstp потоки ip камер в изображения на golang?

Question

[Maxim_SH]

Всем привет!
Ситуация следующая есть n-ое количество ip камер, с которых необходимо конвертировать rtsp поток в изображение и желательно делать это параллельно.
Сделал тестовый пример с 3 камерами, выглядит следующим образом:

func convert1() {
	cmd := "/usr/bin/ffmpeg"
	args := []string{"-i", "rtsp://ip:port", "-f", "image2", "filename.ext"}

	exec.Command(cmd, args...).Run()
}

func convert2() {
	cmd := "/usr/bin/ffmpeg"
	args := []string{"-i", "rtsp://ip:port", "-f", "image2", "filename.ext"}

	exec.Command(cmd, args...).Run()
}

func convert3() {
	cmd := "/usr/bin/ffmpeg"
	args := []string{"-i", "rtsp://ip:port", "-f", "image2", "filename.ext"}

	exec.Command(cmd, args...).Run()
}

func main() {
	go convert1()
	go convert2()
	convert3()
	fmt.Println("RTSP streams of videocameras successfully converted!")
}

В этом случае все отрабатывает отлично. Но как можно реализовать подобное, если мне допустим приходит информация о камерах в json и количество камер может быть все время разное. Как вариант преобразовывать json допустим в map, где ключи это наименования изображений, а значения - ссылки на rtsp потоки камер, либо наоборот. Затем проходится по этому map'у и производить конвертацию на основании тех данных, что содержатся в map'е. Тогда каким образом реализовать параллельную конвертацию, аналогично вышеприведенному примеру? Прошу помощи в решении данного вопроса.

Comments

[abcyu]

А что именно вызывает затруднение?

[Maxim_SH]

Каким образом параллельно выполнять функцию допустим такого вида
func convert(stream, img_name string) {
cmd := "/usr/bin/ffmpeg"
args := []string{"-i", stream, "-f", "image2", img_name}

exec.Command(cmd, args...).Run()
}
в цикле для каждого элемента map'а?

Answer 1

[Александр Павлюк]

Сделайте одну функцию конвертации, которая бы принимала параметры: ип, порт, выходной файл.
В цикле запускайте эту функцию с нужными параметрами через go
Для ожидания завершения горутин можно использовать sync.WaitGroup
nathanleclaire.com/blog/2014/02/15/how-to-wait-for...

Comments

[Maxim_SH]

Спасибо, этот способ помог мне. Но есть еще вопрос, будет ли этот вариант работать, если данные по которым я прохожусь в цикле и использую их для вызова конвертации, будут динамическими (будут парситься из json ответа в моем случае), а не задаваться в виде фиксированного количества в скрипте?

[Александр Павлюк]

Maxim_SH: будет, естественно. Количество просто будете брать из размера массива или map-а

Answer 2

[Maxim_SH]

Приведу полный код решения вопроса на основе ответа Александра Павлюка, возможно другим пригодится:

package main

import (
	"os/exec"
	"sync"
)

func convert(stream, fileName string) {
	cmd := "/usr/bin/ffmpeg"
	args := []string{"-i", stream, "-f", "image2", fileName}

	exec.Command(cmd, args...).Run()
}

func main() {
	rtspStreams := map[string]string{
		"filename.ext": "rtsp://ip:port",
		"filename.ext": "rtsp://ip:port",
		"filename.ext": "rtsp://ip:port",
	}

	var wg sync.WaitGroup

	wg.Add(len(rtspStreams))

	for fileName, stream := range rtspStreams {
		go func(fileName, stream string) {
			defer wg.Done()
			convert(stream, fileName)
		}(fileName, stream)
	}

	wg.Wait()
}

Comments

[Александр Павлюк]

Только вот у map имена всех ключей уникальны, не забывайте. Поэтому, чтобы иметь возможность использовать одинаковые имена файлов, лучше использовать массив структур.

[Maxim_SH]

ок, спасибо, маловероятно что имена файлов будут совпадать, но тем не менее приму это к сведению

[Oleg Shevelev]

Почему не используете каналы?:)
Не думали что в json может прилететь через несколько лет 100500 записей сразу?

Сделайте управляемое количество одинаковых горутин которые слушают один поток. Кто получит сообщение из потока первый тот его и обработает. Количество горутин варьируйте в зависимости от количества ядер процессора.

rtspStreams := map[string]string{

Вряд ли вы собирались прописывать этот map в программе вручную - а значит нужен какой-то интерфейс получения заданий. А значит сервис можно сделать постоянно живущим (демон/сервис, называйте как хотите, например получающий задания по http), а не время от времени запускаемый код. А значит синхронизация по sync.WaitGroup не требуется вовсе (если использовать каналы).

Что бы в вашем коде сделать ограничение на количество одновременных выполнений (не превышение лимита) придётся написать не элегантный и никому не нужный костыль в коде. И если сами горутины очень лёгкие и 100500 их не страшно, то вот ffmpeg будет не таким быстрым и больше 2-4 потоков на процессорное ядро будет бессмысленно.

[Oleg Shevelev]

Так же не понятно что будете делать в такой ситуации:
1. есть пачка данных в map которые вы начали обрабатывать и обрабатывать долго (ffmpeg же не быстрый)
2. пока программа работала с прошлой пачкой данных пришла новая пачка данных (ну а вдруг? Вдруг эти задания софт генерирует например раз в 10 минут или клиентами)
Что будете делать? Запускать вторый экземпляр программы и идти по новой пачке данных? В результате +100500 горутин и запусков ffmpeg. Не равномерное распределение нагрузки (то густо то пусто), не контролируемое время выполнения каждого задания (чем больше заданий одновременно тем медленнее они делаются).

У меня сервис распознавания документов (картинка -> текст) и используется смешанный режим.
1. есть группа заданий, каждая группа заданий работает независимо от другой
2. таких групп заданий может быть много (в самом начале было определено что не больше некоторого разумного числа + есть возможность приостановить или возобновить любые группы через удобную админку обычным менеджером)
3. в каждой группе пачка из примерно 10 000 - 150 000 заданий на распознавание
4. каждое задание подразумевает от 1 до 6 страниц текста которые распознаются по отдельности.

Сложность (фактически максимальное время) алгоритма для распознавания одного задания (1-6 страниц) можете посчитать сами (исходя из количество групп заданий и количества ядер)

[Maxim_SH]

Oleg Shevelev: Ок, спасибо за советы, Олег! С Вами можно по скайпу будет на связь выйти, чтобы пообщаться по вышеизложенными вами предложениями более подробно?

[Oleg Shevelev]

Конечно. Напишите мне завтра во второй половине дня по аккаунту: mantyr