Could not get rid of a blocking ReadFrame.

[Mr3h1v]

Even i have set inputFormatCtx.SetInterruptCallback() and called Interrupt method of the result.

[asticode]

Can you share some code? It's hard to help you without having more information

[Mr3h1v]

Here is example code. udp ts stream will end after about 30s, and call Interrupt at 35th second, but app still blocking.

package main

import (
    "errors"
    "fmt"
    "log"
    "net"
    "time"

    "github.com/asticode/go-astiav"
)

var (
    cb     astiav.IOInterrupter
    input  = "0.0.0.0:10001"
    output = "output.flv"
)

func main() {
    time.AfterFunc(35*time.Second, func() {
        log.Println("try to interrupt")
        cb.Interrupt()
    })

    addr, _ := net.ResolveUDPAddr("udp", input)
    conn, _ := net.ListenUDP("udp", addr)
    defer conn.Close()

    // Alloc packet
    pkt := astiav.AllocPacket()
    defer pkt.Free()

    // Alloc input format context
    inputFormatContext := astiav.AllocFormatContext()
    if inputFormatContext == nil {
        log.Fatal(errors.New("main: input format context is nil"))
    }
    defer inputFormatContext.Free()

    // Set interrupt callback
    cb = inputFormatContext.SetInterruptCallback()

    // Alloc io context
    ioContext, err := astiav.AllocIOContext(4096, false, conn.Read, nil, nil)
    if err != nil {
        log.Fatal(fmt.Errorf("main: allocating io context failed: %w", err))
    }
    defer ioContext.Free()

    // Store io context
    inputFormatContext.SetPb(ioContext)

    // Open input
    if err := inputFormatContext.OpenInput("", astiav.FindInputFormat("mpegts"), nil); err != nil {
        log.Fatal(fmt.Errorf("main: opening input failed: %w", err))
    }
    defer inputFormatContext.CloseInput()

    // Find stream info
    if err := inputFormatContext.FindStreamInfo(nil); err != nil {
        log.Fatal(fmt.Errorf("main: finding stream info failed: %w", err))
    }

    // Alloc output format context
    outputFormatContext, err := astiav.AllocOutputFormatContext(nil, "flv", output)
    if err != nil {
        log.Fatal(fmt.Errorf("main: allocating output format context failed: %w", err))
    }
    if outputFormatContext == nil {
        log.Fatal(errors.New("main: output format context is nil"))
    }
    defer outputFormatContext.Free()

    // Loop through streams
    inputStreams := make(map[int]*astiav.Stream)  // Indexed by input stream index
    outputStreams := make(map[int]*astiav.Stream) // Indexed by input stream index
    for _, is := range inputFormatContext.Streams() {
        // Only process audio or video
        if is.CodecParameters().MediaType() != astiav.MediaTypeAudio &&
            is.CodecParameters().MediaType() != astiav.MediaTypeVideo {
            continue
        }

        // Add input stream
        inputStreams[is.Index()] = is

        // Add stream to output format context
        os := outputFormatContext.NewStream(nil)
        if os == nil {
            log.Fatal(errors.New("main: output stream is nil"))
        }

        // Copy codec parameters
        if err = is.CodecParameters().Copy(os.CodecParameters()); err != nil {
            log.Fatal(fmt.Errorf("main: copying codec parameters failed: %w", err))
        }

        // Reset codec tag
        os.CodecParameters().SetCodecTag(0)

        // Add output stream
        outputStreams[is.Index()] = os
    }

    // If this is a file, we need to use an io context
    if !outputFormatContext.OutputFormat().Flags().Has(astiav.IOFormatFlagNofile) {
        // Open io context
        ioContext, err := astiav.OpenIOContext(output, astiav.NewIOContextFlags(astiav.IOContextFlagWrite))
        if err != nil {
            log.Fatal(fmt.Errorf("main: opening io context failed: %w", err))
        }
        defer ioContext.Close() //nolint:errcheck

        // Update output format context
        outputFormatContext.SetPb(ioContext)
    }

    // Write header
    if err = outputFormatContext.WriteHeader(nil); err != nil {
        log.Fatal(fmt.Errorf("main: writing header failed: %w", err))
    }

    // Loop through packets
    for {
        // Read frame
        if err = inputFormatContext.ReadFrame(pkt); err != nil {
            if errors.Is(err, astiav.ErrEof) {
                break
            }
            log.Fatal(fmt.Errorf("main: reading frame failed: %w", err))
        }

        // Get input stream
        inputStream, ok := inputStreams[pkt.StreamIndex()]
        if !ok {
            pkt.Unref()
            continue
        }

        // Get output stream
        outputStream, ok := outputStreams[pkt.StreamIndex()]
        if !ok {
            pkt.Unref()
            continue
        }

        // Update packet
        pkt.SetStreamIndex(outputStream.Index())
        pkt.RescaleTs(inputStream.TimeBase(), outputStream.TimeBase())
        pkt.SetPos(-1)

        // Write frame
        if err = outputFormatContext.WriteInterleavedFrame(pkt); err != nil {
            log.Fatal(fmt.Errorf("main: writing interleaved frame failed: %w", err))
        }
    }

    // Write trailer
    if err = outputFormatContext.WriteTrailer(); err != nil {
        log.Fatal(fmt.Errorf("main: writing trailer failed: %w", err))
    }

    // Success
    log.Println("success")
}

[asticode]

FYI the very same code (minus the udp binding) works properly if you replace inputFormatContext.OpenInput("", astiav.FindInputFormat("mpegts"), nil) with inputFormatContext.OpenInput("udp://"+input, astiav.FindInputFormat("mpegts"), nil) and remove the custom io context. Therefore if that fits your original use case, I'd recommend using this instead.

However, if you can't, the reason why this is blocking is because the conn.Read method is blocking and therefore the interrupt callback is never called (the interrupt callback is not called in parallel of the read function). To fix this problem, you need to find a way to provide a read function to AllocIOContext that doesn't block when the stream has stopped.

[asticode]

For instance replacing cb.Interrupt() with conn.Close() in your example actually unblocks everything 👍 Needless to say that you can remove the IO interrupter logic in that case

[Mr3h1v]

Thanks a lot. That is just a simplified example above. I must to use AllocIOContext because i only got an io.Reader In my real problem. I will find a way to close it.