FFmpeg Custom Video Filter 만들기 #3

fromDK 23.05.31

2,613 8 0

facebook twitter kakao link

8 0

facebook twitter kakao link

이번엔 IPC로 영상을 수신하여 Python 엔진을 수행하고 결과를 다시 IPC로 송신하는 코드를 구현하였습니다.

Signal Handler 등록

이전 글에서 Video Filter에서 fork로 프로세스 생성 후, python코드를 실행하였습니다.

항상 모든 문제는 종료 시점에서 많이 발생하므로, FFmpeg의 Video Filter가 종료될 때, 자식 프로세스에 SIGQUIT 시그널을 전달하여 종료 될 수 있도록 알려줍니다.

만약 시그널 전달이 실패한다면, SIGKILL을 전달하여 부모잃은 좀비 프로세스가 되지 않도록 합니다.

static void uninit_pipe(AVFilterContext *ctx, SupernovaContext *supernova)
{
    ....
    
    int ret = 0;
    ret = kill(supernova->engine_pid, 3); // 3 : SIGQUIT
    if (ret ) {
        kill(supernova->engine_pid, 9);
    }
    
   ....
}

또한, 부모 프로세스에서 종료시 자식 프로세스가 종료되길 기다립니다.

    int status;
    while (waitpid(supernova->engine_pid, &status, WNOHANG) > 0)
        usleep(10*1000);

다시 python 코드로 돌아와서 시그널 핸들러를 등록 하고, 시그널 발생시 Loop를 종료하여 정상 종료 될 수 있도록 합니다.

def handler(signum, frame):
    global loop
    logger.info(f'Signal handler called with signal {signal.Signals(signum).name}:{signum}')
    logger.info(frame)
    loop = False


signal.signal(signal.SIGINT, handler)
signal.signal(signal.SIGQUIT, handler)

Named Pipe(FIFO) 오픈

이후, ML python 엔진의 입력 pipe와 출력 pipe가 생성 될때 까지 대기 하다가, open을 수행합니다.

    logger.debug('wait until fifo is open')
    while not args.in_fifo.exists() or not args.out_fifo.exists():
        time.sleep(0.1)

    try:
        in_fifo = open(args.in_fifo, mode='rb')
        out_fifo = open(args.out_fifo, mode='wb')
    except Exception as e:
        logger.error(e)
        exit(1)

Filtering Loop

테스트로 입력 받은 데이터를 다시 loopback으로 전달 하도록 구현하였습니다.

입력 영상의 Pixel Foramt을 확인 후, 읽어야 할 메모리의 크기를 계산하고, read() 함수를 통해 데이터를 읽어 드립니다.

이후 ML Engine에서 필터링을 수행합니다. 이번 예제에서는 단간하게 Loopback으로 구현하기 위해 입력 데이터를 다시 출력에 사용하였습니다.

이후 write()함수로 출력합니다. 위에서 signal을 수신한 경우, loop가 종료되도록 구현 하였습니다.

그리고 아래에서는 구현하지 않았으나, size가 0인 데이터를 write()하여 종료의 의미로 전달하는 것도 가능합니다.

    in_size = args.in_width * args.in_height
    if args.pix_format == 'AV_PIX_FMT_YUV420P':
        in_size = int(in_size * 3 / 2)
    else:
        assert False, 'unsupperted pixel format'

    in_data = []
    while loop:
        offset = 0
        in_data = []
        while loop and offset < in_size:
            try:
                buff = in_fifo.read(in_size)
                if not loop:
                    break

                if len(buff) > 0:
                    in_data.extend(buff)

                offset += len(buff)
            except Exception as e:
                logger.error(e)
                loop = False

        logger.debug(f'received size {len(in_data)}')


        # TODO : do filtering data in ML engine
        out_data = in_data
        out_size = in_size
   

        offset = 0
        while loop and offset < out_size:
            try:
                written = out_fifo.write(bytes(out_data[offset:]))
                if not loop:
                    break

                if written > 0:
                    offset += written
            except Exception as e:
                logger.error(e)
                loop = False

        logger.debug(f'written size {offset}')

Exceptional Case

위와 같이 구현하면 정상적인 케이스에서는 잘 동작합니다.

그런데 만약, Python 코드에서 동작 중, 오류가 발생하는 경우를 생각해 봅시다.

우선 FFmpeg에서는 write()를 수행하여 데이터를 전달하고, python엔진에서 필터링 된 데이터를 전송할때까지 read()함수에서 Block으로 기다리게 됩니다.

이때, python 코드에서 예외 발생하여 종료되는 경우, FFmpeg에서는 block걸린 채로 깨어나지 않고 계속 기다리게 됩니다.

이러한 문제를 해결하기 위해 보통 Named Pipe를 non-block모드로 오픈하거나, select(), epoll() 등을 이용한 I/O Multiplexing을 이용합니다.

아래 코드는 O_NONBLOCK 모드로 설정 한 예시 입니다. 이 경우 수신된 버퍼가 없는 경우, read()에서 -1을 리턴하며, errno가 EAGAIN으로 설정 됩니다.

이후 usleep()등으로 cpu점유율이 줄어들 수 있도록 한다음 다시 read()를 시도합니다.

이와 같이 non block 모드로 동작을 시키면, 위와 같이 python 코드에서 비정상 종료 되더라도, read()에서 block이 되지 않아 잘 돌아게가 됩니다.

        if ((supernova->out_pipe = open(supernova->in_pipe_path, O_RDONLY | O_NOCTTY | O_NONBLOCK)) < 0) {
            av_log(ctx, AV_LOG_ERROR, "failed to open supernova output pipe : %s", supernova->out_pipe_path);
        }

또는 Block모드로 설정하고, 아래와 같이 select()함수를 이용하여 file descriptor를 모니터링 할 수 있습니다.

read 할 fd만 Set된 경우 read()를 수행 하여, 무한 block 되는 것을 방지 할 수 있습니다.

int read_with_timeout(int fd, char* buf, int buf_size, int timeout_ms)
{
    int rx_len = 0;
    struct timeval timeout;
    fd_set readFds;
    int state;

    timeout.tv_sec = 0;
    timeout.tv_usec = timeout_ms * 1000;

    FD_ZERO(&readFds);
    FD_SET(fd, &readFds);
    state = select(fd + 1, &readFds, NULL, NULL, &timeout);
    switch (state) {
    case -1:
        perror("select error : ");
        exit(1);
        break;

    default:
        if (FD_ISSET(fd, &readFds)) {
            rx_len = read(fd, buf, buf_size);
        }
        break;
    }
    return rx_len;
}