LPMS bottleneck fix

ffmpeg/api_test.go

@@ -0,0 +1,90 @@
+package ffmpeg
+
+import (
+	"testing"
+)
+
+func TestTranscoderAPI_InvalidFile(t *testing.T) {
+	// Test the following file open results on input: fail, success, fail, success
+
+	tc := NewTranscoder()
+	defer tc.StopTranscoder()
+	in := &TranscodeOptionsIn{}
+	out := []TranscodeOptions{TranscodeOptions{
+		Oname:        "-",
+		AudioEncoder: ComponentOptions{Name: "copy"},
+		VideoEncoder: ComponentOptions{Name: "drop"},
+		Muxer:        ComponentOptions{Name: "null"},
+	}}
+
+	// fail # 1
+	in.Fname = "none"
+	_, err := tc.Transcode(in, out)
+	if err == nil || err.Error() != "No such file or directory" {
+		t.Error("Expected 'No such file or directory', got ", err)
+	}
+
+	// success # 1
+	in.Fname = "../transcoder/test.ts"
+	_, err = tc.Transcode(in, out)
+	if err != nil {
+		t.Error(err)
+	}
+
+	// fail # 2
+	in.Fname = "none"
+	_, err = tc.Transcode(in, out)
+	if err == nil || err.Error() != "No such file or directory" {
+		t.Error("Expected 'No such file or directory', got ", err)
+	}
+
+	// success # 2
+	in.Fname = "../transcoder/test.ts"
+	_, err = tc.Transcode(in, out)
+	if err != nil {
+		t.Error(err)
+	}
+
+	// Now check invalid output filename
+	out[0].Muxer = ComponentOptions{Name: "md5"}
+	out[0].Oname = "/not/really/anywhere"
+	_, err = tc.Transcode(in, out)
+	if err == nil {
+		t.Error(err)
+	}
+
+}
+
+func TestTranscoderAPI_Stopped(t *testing.T) {
+
+	// Test stopped transcoder
+	tc := NewTranscoder()
+	tc.StopTranscoder()
+	in := &TranscodeOptionsIn{}
+	_, err := tc.Transcode(in, nil)
+	if err != ErrTranscoderStp {
+		t.Errorf("Unexpected error; wanted %v but got %v", ErrTranscoderStp, err)
+	}
+
+	// test somehow munged transcoder handle
+	tc2 := NewTranscoder()
+	tc2.handle = nil // technically this leaks memory ... OK for test
+	_, err = tc2.Transcode(in, nil)
+	if err != ErrTranscoderStp {
+		t.Errorf("Unexpected error; wanted %v but got %v", ErrTranscoderStp, err)
+	}
+}
+
+func TestTranscoderAPI_TooManyOutputs(t *testing.T) {
+
+	out := make([]TranscodeOptions, 11)
+	for i, _ := range out {
+		out[i].VideoEncoder = ComponentOptions{Name: "drop"}
+	}
+	in := &TranscodeOptionsIn{}
+	tc := NewTranscoder()
+	_, err := tc.Transcode(in, out)
+	if err == nil || err.Error() != "Too many outputs" {
+		t.Error("Expected 'Too many outputs', got ", err)
+	}
+}

ffmpeg/ffmpeg.go

@@ -8,6 +8,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"strings"
+	"sync"
 	"unsafe"
 )
 
@@ -19,6 +20,7 @@ import "C"
 var ErrTranscoderRes = errors.New("TranscoderInvalidResolution")
 var ErrTranscoderHw = errors.New("TranscoderInvalidHardware")
 var ErrTranscoderInp = errors.New("TranscoderInvalidInput")
+var ErrTranscoderStp = errors.New("TranscoderStopped")
 
 type Acceleration int
 
@@ -33,6 +35,12 @@ type ComponentOptions struct {
 	Opts map[string]string
 }
 
+type Transcoder struct {
+	handle  *C.struct_transcode_thread
+	stopped bool
+	mu      *sync.Mutex
+}
+
 type TranscodeOptionsIn struct {
 	Fname  string
 	Accel  Acceleration
@@ -155,6 +163,17 @@ func Transcode2(input *TranscodeOptionsIn, ps []TranscodeOptions) error {
 }
 
 func Transcode3(input *TranscodeOptionsIn, ps []TranscodeOptions) (*TranscodeResults, error) {
+	t := NewTranscoder()
+	defer t.StopTranscoder()
+	return t.Transcode(input, ps)
+}
+
+func (t *Transcoder) Transcode(input *TranscodeOptionsIn, ps []TranscodeOptions) (*TranscodeResults, error) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.stopped || t.handle == nil {
+		return nil, ErrTranscoderStp
+	}
 	if input == nil {
 		return nil, ErrTranscoderInp
 	}
@@ -198,7 +217,7 @@ func Transcode3(input *TranscodeOptionsIn, ps []TranscodeOptions) (*TranscodeRes
 		filters += fmt.Sprintf("%s='w=if(gte(iw,ih),%d,-2):h=if(lt(iw,ih),%d,-2)'", scale_filter, w, h)
 		if input.Accel != Software && p.Accel == Software {
 			// needed for hw dec -> hw rescale -> sw enc
-			filters = filters + ":format=yuv420p,hwdownload"
+			filters = filters + ",hwdownload,format=nv12"
 		}
 		muxOpts := C.component_opts{
 			opts: newAVOpts(p.Muxer.Opts), // don't free this bc of avformat_write_header API
@@ -207,6 +226,12 @@ func Transcode3(input *TranscodeOptionsIn, ps []TranscodeOptions) (*TranscodeRes
 			muxOpts.name = C.CString(p.Muxer.Name)
 			defer C.free(unsafe.Pointer(muxOpts.name))
 		}
+		// Set some default encoding options
+		if len(p.VideoEncoder.Name) <= 0 && len(p.VideoEncoder.Opts) <= 0 {
+			p.VideoEncoder.Opts = map[string]string{
+				"forced-idr": "1",
+			}
+		}
 		vidOpts := C.component_opts{
 			name: C.CString(encoder),
 			opts: newAVOpts(p.VideoEncoder.Opts),
@@ -230,13 +255,28 @@ func Transcode3(input *TranscodeOptionsIn, ps []TranscodeOptions) (*TranscodeRes
 		params[i] = C.output_params{fname: oname, fps: fps,
 			w: C.int(w), h: C.int(h), bitrate: C.int(bitrate),
 			muxer: muxOpts, audio: audioOpts, video: vidOpts, vfilters: vfilt}
+		defer func(param *C.output_params) {
+			// Work around the ownership rules:
+			// ffmpeg normally takes ownership of the following AVDictionary options
+			// However, if we don't pass these opts to ffmpeg, then we need to free
+			if param.muxer.opts != nil {
+				C.av_dict_free(&param.muxer.opts)
+			}
+			if param.audio.opts != nil {
+				C.av_dict_free(&param.audio.opts)
+			}
+			if param.video.opts != nil {
+				C.av_dict_free(&param.video.opts)
+			}
+		}(&params[i])
 	}
 	var device *C.char
 	if input.Device != "" {
 		device = C.CString(input.Device)
 		defer C.free(unsafe.Pointer(device))
 	}
-	inp := &C.input_params{fname: fname, hw_type: hw_type, device: device}
+	inp := &C.input_params{fname: fname, hw_type: hw_type, device: device,
+		handle: t.handle}
 	results := make([]C.output_results, len(ps))
 	decoded := &C.output_results{}
 	var (
@@ -249,7 +289,7 @@ func Transcode3(input *TranscodeOptionsIn, ps []TranscodeOptions) (*TranscodeRes
 	}
 	ret := int(C.lpms_transcode(inp, paramsPointer, resultsPointer, C.int(len(params)), decoded))
 	if 0 != ret {
-		glog.Infof("Transcoder Return : %v\n", Strerror(ret))
+		glog.Error("Transcoder Return : ", ErrorMap[ret])
 		return nil, ErrorMap[ret]
 	}
 	tr := make([]MediaInfo, len(ps))
@@ -266,6 +306,24 @@ func Transcode3(input *TranscodeOptionsIn, ps []TranscodeOptions) (*TranscodeRes
 	return &TranscodeResults{Encoded: tr, Decoded: dec}, nil
 }
 
+func NewTranscoder() *Transcoder {
+	return &Transcoder{
+		handle: C.lpms_transcode_new(),
+		mu:     &sync.Mutex{},
+	}
+}
+
+func (t *Transcoder) StopTranscoder() {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	if t.stopped {
+		return
+	}
+	C.lpms_transcode_stop(t.handle)
+	t.handle = nil // prevent accidental reuse
+	t.stopped = true
+}
+
 func InitFFmpeg() {
 	C.lpms_init()
 }

ffmpeg/ffmpeg_errors.go

@@ -35,6 +35,7 @@ func error_map() map[int]error {
 	}{
 		{code: C.lpms_ERR_INPUT_PIXFMT, desc: "Unsupported input pixel format"},
 		{code: C.lpms_ERR_FILTERS, desc: "Error initializing filtergraph"},
+		{code: C.lpms_ERR_OUTPUTS, desc: "Too many outputs"},
 	}
 	for _, v := range lpmsErrors {
 		m[int(v.code)] = errors.New(v.desc)