Request for Help on Using Hardware Transcoding

[DarkiT]

As a Go beginner, this issue has troubled me for three days. I have tried using both SoftwareScaleContext and Filters to convert frames into images. Despite my efforts to understand the code in the demo, I still haven't achieved the result I want. Currently, I can successfully transcode using SoftwareScaleContext, but as soon as I do, the memory usage keeps increasing, and I can't figure out which part of the code is causing the problem. When trying the Filters approach, I've spent the whole day without making any progress. Can someone help me fix the code below so that both methods work properly?

I'm currently using the go-astiav version from the n6.1.1 branch. As for why I'm not using the latest version, it's because I'm working with ffmpeg-rockchip, and the latest 7.0 branch of ffmpeg-rockchip is not compatible with go-astiav.

package main

import (
	"errors"
	"fmt"
	"image"
	"strconv"

	"github.com/asticode/go-astiav"
	log "github.com/darkit/slog"
	"github.com/deepch/vdk/av"
	"github.com/deepch/vdk/codec/h264parser"
	"github.com/deepch/vdk/codec/h265parser"
)

type VideoDecoder struct {
	fps                   int
	width                 int
	height                int
	decCodec              *astiav.Codec
	decCodecContext       *astiav.CodecContext
	hardwareDeviceContext *astiav.HardwareDeviceContext
	hardwarePixelFormat   astiav.PixelFormat
	softwareFrame         *astiav.Frame
	hardwareFrame         *astiav.Frame
	rgbaFrame             *astiav.Frame
	avPkt                 *astiav.Packet
	swsContext            *astiav.SoftwareScaleContext
	filterGraph           *astiav.FilterGraph
	buffersrcContext      *astiav.FilterContext
	buffersinkContext     *astiav.FilterContext
}

func NewVideoDecoder(stream av.CodecData, useHardware bool) (*VideoDecoder, error) {
	var (
		codecID   astiav.CodecID
		extraData []byte
		fps       int
		width     int
		height    int
	)

	astiav.SetLogLevel(astiav.LogLevelDebug)

	switch stream.Type() {
	case av.H264:
		codecID = astiav.CodecIDH264
		h264 := stream.(h264parser.CodecData)
		fps = h264.FPS()
		extraData = h264.AVCDecoderConfRecordBytes()
		width, height = h264.Width(), h264.Height()
	case av.H265:
		codecID = astiav.CodecIDH265
		h265 := stream.(h265parser.CodecData)
		fps = h265.FPS()
		extraData = h265.AVCDecoderConfRecordBytes()
		width, height = h265.Width(), h265.Height()
	default:
		return nil, fmt.Errorf("unsupported codec type: %v", stream.Type())
	}

	v := &VideoDecoder{
		fps:    fps,
		width:  width,
		height: height,
	}

	// Initialize decoder
	if useHardware {
		v.decCodec = findHardwareDecoder(codecID)
	} else {
		v.decCodec = astiav.FindDecoder(codecID)
	}
	if v.decCodec == nil {
		return nil, fmt.Errorf("decoder not found: %v", codecID)
	}
	log.Infof("Found hardware-accelerated decoder: %v", codecID.String())

	v.decCodecContext = astiav.AllocCodecContext(v.decCodec)
	if v.decCodecContext == nil {
		return nil, fmt.Errorf("unable to allocate decoder context")
	}

	v.decCodecContext.SetWidth(width)
	v.decCodecContext.SetHeight(height)
	v.decCodecContext.SetExtraData(extraData)

	if useHardware {
		// Initialize hardware decoding
		hardwareDeviceType := astiav.FindHardwareDeviceTypeByName("rkmpp")
		//hardwareDeviceType := astiav.FindHardwareDeviceTypeByName(v.decCodec.String())
		for _, p := range v.decCodec.HardwareConfigs() {
			if p.MethodFlags().Has(astiav.CodecHardwareConfigMethodFlagHwDeviceCtx) && p.HardwareDeviceType() == hardwareDeviceType {
				v.hardwarePixelFormat = p.PixelFormat()
				break
			}
		}

		v.hardwareDeviceContext, _ = astiav.CreateHardwareDeviceContext(hardwareDeviceType, "0", nil)
		if v.hardwareDeviceContext != nil {
			v.decCodecContext.SetHardwareDeviceContext(v.hardwareDeviceContext)
			v.decCodecContext.SetPixelFormatCallback(func(pfs []astiav.PixelFormat) astiav.PixelFormat {
				log.Infof("list: %v", pfs)
				for _, pf := range pfs {
					if pf == v.hardwarePixelFormat {
						v.hardwarePixelFormat = pf
						return pf
					}
				}
				return astiav.PixelFormatNone
			})
		}

		log.Info("Video codec info", "stream", stream.Type().String(), "decoder", v.decCodec.String(), "pixel", v.hardwarePixelFormat.String())
	}

	if err := v.decCodecContext.Open(v.decCodec, nil); err != nil {
		return nil, fmt.Errorf("failed to open decoder: %v", err)
	}

	v.softwareFrame = astiav.AllocFrame()
	v.hardwareFrame = astiav.AllocFrame()
	v.rgbaFrame = astiav.AllocFrame()
	v.avPkt = astiav.AllocPacket()

	return v, nil
}

func (v *VideoDecoder) DecodePacketWithSoftwareScaleContext(pkt *av.Packet) (image.Image, error) {
	defer v.hardwareFrame.Unref()
	defer v.softwareFrame.Unref()
	defer v.rgbaFrame.Unref()
	defer v.avPkt.Unref()

	_ = v.avPkt.FromData(pkt.Data)

	if err := v.decCodecContext.SendPacket(v.avPkt); err != nil {
		return nil, fmt.Errorf("failed to send packet to decoder: %v", err)
	}

	for {
		err := v.decCodecContext.ReceiveFrame(v.hardwareFrame)
		if err != nil {
			if errors.Is(err, astiav.ErrEagain) || errors.Is(err, astiav.ErrEof) {
				return nil, nil
			}
			return nil, fmt.Errorf("failed to receive frame: %v", err)
		}

		var frame *astiav.Frame

		if v.hardwareFrame.PixelFormat() == v.hardwarePixelFormat {
			if err := v.hardwareFrame.TransferHardwareData(v.softwareFrame); err != nil {
				return nil, fmt.Errorf("failed to transfer hardware frame to software frame: %v", err)
			}

			v.softwareFrame.SetPts(v.hardwareFrame.Pts())
			frame = v.softwareFrame
		} else {
			frame = v.hardwareFrame
		}

		if v.swsContext == nil {
			v.swsContext, err = astiav.CreateSoftwareScaleContext(
				frame.Width(), frame.Height(), frame.PixelFormat(),
				frame.Width(), frame.Height(), astiav.PixelFormatRgba,
				astiav.NewSoftwareScaleContextFlags(astiav.SoftwareScaleContextFlagBilinear),
			)
			if err != nil {
				return nil, fmt.Errorf("failed to create software scaling context: %v", err)
			}
		}

		if err := v.swsContext.ScaleFrame(frame, v.rgbaFrame); err != nil {
			return nil, fmt.Errorf("failed to scale frame: %v", err)
		}

		img, err := v.rgbaFrame.Data().GuessImageFormat()
		if err != nil {
			return nil, fmt.Errorf("failed to guess image format: %v", err)
		}

		if err := v.rgbaFrame.Data().ToImage(img); err != nil {
			return nil, fmt.Errorf("failed to convert frame to image: %v", err)
		}

		return img, nil
	}
}

func (v *VideoDecoder) DecodePacketWithFilter(pkt *av.Packet) (image.Image, error) {
	defer v.hardwareFrame.Unref()
	defer v.softwareFrame.Unref()
	defer v.rgbaFrame.Unref()
	defer v.avPkt.Unref()

	_ = v.avPkt.FromData(pkt.Data)

	if err := v.decCodecContext.SendPacket(v.avPkt); err != nil {
		return nil, fmt.Errorf("failed to send packet to decoder: %v", err)
	}

	for {
		err := v.decCodecContext.ReceiveFrame(v.hardwareFrame)
		if err != nil {
			if errors.Is(err, astiav.ErrEagain) || errors.Is(err, astiav.ErrEof) {
				return nil, nil
			}
			return nil, fmt.Errorf("failed to receive frame: %v", err)
		}

		var frame *astiav.Frame

		if v.hardwareFrame.PixelFormat() == v.hardwarePixelFormat {
			if err = v.hardwareFrame.TransferHardwareData(v.softwareFrame); err != nil {
				return nil, fmt.Errorf("failed to transfer hardware frame to software frame: %v", err)
			}

			v.softwareFrame.SetPts(v.hardwareFrame.Pts())
			frame = v.softwareFrame
		} else {
			frame = v.hardwareFrame
		}

		if v.filterGraph == nil {
			v.filterGraph = astiav.AllocFilterGraph()
			buffersrc := astiav.FindFilterByName("buffer")
			buffersink := astiav.FindFilterByName("buffersink")

			args := astiav.FilterArgs{
				"video_size":   fmt.Sprintf("%dx%d", v.decCodecContext.Width(), v.decCodecContext.Height()),
				"pix_fmt":      strconv.Itoa(int(v.decCodecContext.PixelFormat())),
				"pixel_aspect": v.decCodecContext.SampleAspectRatio().String(),
				"time_base":    "1/1000",
			}

			v.buffersrcContext, err = v.filterGraph.NewFilterContext(buffersrc, "in", args)
			if err != nil {
				return nil, fmt.Errorf("failed to create buffersrc context: %v", err)
			}

			v.buffersinkContext, err = v.filterGraph.NewFilterContext(buffersink, "out", nil)
			if err != nil {
				return nil, fmt.Errorf("failed to create buffersink context: %v", err)
			}

			outputs := []*astiav.FilterInOut{
				astiav.NewFilterInOut().SetContext(v.buffersrcContext).SetPadIdx(0).SetNext(nil),
			}
			inputs := []*astiav.FilterInOut{
				astiav.NewFilterInOut().SetContext(v.buffersinkContext).SetPadIdx(0).SetNext(nil),
			}

			if err := v.filterGraph.Parse(inputs, outputs); err != nil {
				return nil, fmt.Errorf("failed to parse filter graph: %v", err)
			}
			if err := v.filterGraph.Configure(); err != nil {
				return nil, fmt.Errorf("failed to configure filter graph: %v", err)
			}
		}

		if err := v.buffersrcContext.AddFrame(frame); err != nil {
			return nil, fmt.Errorf("failed to add frame to filter graph: %v", err)
		}

		for {
			err := v.buffersinkContext.GetFrame(v.softwareFrame)
			if err != nil {
				if errors.Is(err, astiav.ErrEagain) || errors.Is(err, astiav.ErrEof) {
					break
				}
				return nil, fmt.Errorf("failed to get frame from filter graph: %v", err)
			}

			if v.swsContext == nil {
				v.swsContext, err = astiav.CreateSoftwareScaleContext(
					v.softwareFrame.Width(), v.softwareFrame.Height(), v.softwareFrame.PixelFormat(),
					v.softwareFrame.Width(), v.softwareFrame.Height(), astiav.PixelFormatRgba,
					astiav.NewSoftwareScaleContextFlags(astiav.SoftwareScaleContextFlagBilinear),
				)
				if err != nil {
					return nil, fmt.Errorf("failed to create software scaling context: %v", err)
				}
			}

			if err := v.swsContext.ScaleFrame(v.softwareFrame, v.rgbaFrame); err != nil {
				return nil, fmt.Errorf("failed to scale frame: %v", err)
			}

			img, err := v.rgbaFrame.Data().GuessImageFormat()
			if err != nil {
				return nil, fmt.Errorf("failed to guess image format: %v", err)
			}

			if err := v.rgbaFrame.Data().ToImage(img); err != nil {
				return nil, fmt.Errorf("failed to convert frame to image: %v", err)
			}

			return img, nil
		}
	}
}

func findHardwareDecoder(codecID astiav.CodecID) *astiav.Codec {
	decoder := astiav.FindDecoder(codecID)
	if decoder == nil {
		return nil
	}

	hwConfigs := decoder.HardwareConfigs()
	if len(hwConfigs) > 0 {
		return decoder
	}

	for _, codec := range astiav.Codecs() {
		if codec.ID() == codecID && codec.IsDecoder() {
			hwConfigs = codec.HardwareConfigs()
			if len(hwConfigs) > 0 {
				return codec
			}
		}
	}

	return nil
}

[asticode]

First off, why is the latest 7.0 branch of ffmpeg-rockchip not compatible with go-astiav? What errors are you seeing when compiling? 🤔

[DarkiT]

Apologies, I may not have described the issue accurately. Besides using go-astiav, I also use gocv, the CGO binding for OpenCV. During my last compilation, I encountered an error, but I can't remember exactly where it occurred. I will recompile ffmpeg-rockchip 7.0 for testing later and post my error logs.

[asticode]

Second question, why do you need to provide a pkt *av.Packet to your Decode() functions, can't you get packets with go-astiav?

Also, do you confirm that if you return right at the beginning of your Decode() functions (and do nothing inside it), memore doesn't increase when you run your program?

[DarkiT]

As a Go beginner, the configuration options for FFmpeg present a steep learning curve for me, so I decided to use a third-party RTSP client to handle the stream pulling. After extracting the packets, it's easier to process them using go-astiav. This is because the deepch/vdk library has many learning examples.

When not using SoftwareScaleContext to transcode into images, decoding the packet (pkt *av.Packet) into FFmpeg frames with go-astiav runs very stably. On the RK3588 platform, CPU usage is at 1.2%, and memory usage is at 0.9% of 16GB. However, as soon as I use SoftwareScaleContext, while it does successfully encode the frame into a JPG image, memory usage increases rapidly until it exhausts all the available memory. You previously suggested I use Filters, which I have tried, but with my current skill level, I don’t fully understand it yet, and the code is not working properly.

[asticode]

As a Go beginner, the configuration options for FFmpeg present a steep learning curve for me, so I decided to use a third-party RTSP client to handle the stream pulling. After extracting the packets, it's easier to process them using go-astiav. This is because the deepch/vdk library has many learning examples.

Understood 👍 FYI if you're trying to read an rtsp stream you can take a look at the demuxing/decoding example and replace *input here with your rtsp url 👍 You would then be able to get packets that you could send to your decoder

When not using SoftwareScaleContext to transcode into images, decoding the packet (pkt *av.Packet) into FFmpeg frames with go-astiav runs very stably. On the RK3588 platform, CPU usage is at 1.2%, and memory usage is at 0.9% of 16GB. However, as soon as I use SoftwareScaleContext, while it does successfully encode the frame into a JPG image, memory usage increases rapidly until it exhausts all the available memory. You previously suggested I use Filters, which I have tried, but with my current skill level, I don’t fully understand it yet, and the code is not working properly.

It seems you've found where the memory leak was coming from. I've suggested 2 options to fix the issue. But let me know whether you still want some help on how to fix your code using filters

[DarkiT]

After updating the code to ffmpeg 7.0 and go-astiav, I encountered the same issue again. I now confirm that the memory leak occurs at CreateSoftwareScaleContext. I found others online who faced the same problem and provided their solutions, but I'm not sure if these would be effective for go-astiav as well.

Issue with sws_getContext Function Not Releasing Memory

After repeated investigations, I discovered a memory leak issue with the sws_getContext function. Strangely, even though sws_freeContext(pa->sws_ctx) has been called in the program to release the memory, it still leaks.

Later, I found a function called sws_getCachedContext(). After replacing it, the memory leak was resolved!!!

sws_getCachedContext

/**
 * Check if context can be reused; otherwise, reallocate a new one.
 *
 * If context is NULL, just calls sws_getContext() to get a new
 * context. Otherwise, checks if the parameters are the same as those
 * already saved in the context. If that is the case, it returns the current
 * context. Otherwise, it frees the context and gets a new context with
 * the new parameters.
 *
 * Be warned that srcFilter and dstFilter are not checked; they
 * are assumed to remain the same.
 */
struct SwsContext *sws_getCachedContext(struct SwsContext *context,
                                        int srcW, int srcH, enum AVPixelFormat srcFormat,
                                        int dstW, int dstH, enum AVPixelFormat dstFormat,
                                        int flags, SwsFilter *srcFilter,
                                        SwsFilter *dstFilter, const double *param);

/* Parameter Descriptions */
int srcW, /* Width of the input image */
int srcH, /* Height of the input image */
enum AVPixelFormat srcFormat, /* Pixel format of the input image */
int dstW, /* Width of the output image */
int dstH, /* Height of the output image */
enum AVPixelFormat dstFormat, /* Pixel format of the output image */
int flags, /* Choose scaling algorithm (valid only when input and output image sizes differ), usually choose SWS_FAST_BILINEAR */
SwsFilter *srcFilter, /* Filter information for the input image, use NULL if not needed */
SwsFilter *dstFilter, /* Filter information for the output image, use NULL if not needed */
const double *param /* Parameters required for specific scaling algorithms (?), defaults to NULL */

[DarkiT]

Yes, I added this logic, but even if I remove it, the memory still keeps increasing. Eventually, I confirmed that the memory growth is happening in this method.

if err := d.swsCtx.ScaleFrame(d.frame, d.dstFrame); err != nil {
	return nil, err
}

[DarkiT]

[asticode]

Can you share the full code so that I try to reproduce on my machine? I've tried reproducing the issue with the scaling example but I can't 🤔

[DarkiT]

I think I've found the cause, and it's not caused by go-astiav. The root of the problem is that I used gocv to convert the image into a gocv.Mat, but in the test code, I didn't release the gocv.Mat in time. I just discovered this issue while copying the code to you.

func (d *Decoder) NRGBAToMat(img *image.NRGBA) (gocv.Mat, error) {
	bounds := img.Bounds()
	width, height := bounds.Dx(), bounds.Dy()

	mat := gocv.NewMatWithSize(height, width, gocv.MatTypeCV8UC3)
	defer mat.Close() //I didn't add this line, so the memory kept increasing.

	nrgbaBytes := img.Pix

	matBytes, err := mat.DataPtrUint8()
	if err != nil {
		return gocv.Mat{}, err
	}

	j := 0
	for i := 0; i < len(nrgbaBytes); i += 4 {
		matBytes[j] = nrgbaBytes[i]     // R
		matBytes[j+1] = nrgbaBytes[i+1] // G
		matBytes[j+2] = nrgbaBytes[i+2] // B
		j += 3
	}

	return mat, nil
}

Thank you for your patient answers.

[asticode]

No worries! Glad you found an answer ❤️