New pipeline, fix video issues and transcoding stalls

lib/src/androidTest/java/com/otaliastudios/transcoder/integration/IssuesTests.kt

@@ -1,5 +1,6 @@
 package com.otaliastudios.transcoder.integration
 
+import android.media.MediaFormat
 import android.media.MediaMetadataRetriever.METADATA_KEY_DURATION
 import android.media.MediaMetadataRetriever
 import androidx.test.ext.junit.runners.AndroidJUnit4
@@ -66,7 +67,7 @@ class IssuesTests {
     }
 
 
-    @Test(timeout = 3000)
+    @Test(timeout = 5000)
     fun issue137() = with(Helper(137)) {
         transcode {
             // addDataSource(ClipDataSource(input("main.mp3"), 0L, 200_000L))
@@ -93,7 +94,7 @@ class IssuesTests {
         Unit
     }
 
-    @Test(timeout = 3000)
+    @Test(timeout = 5000)
     fun issue184() = with(Helper(184)) {
         transcode {
             addDataSource(TrackType.VIDEO, input("transcode.3gp"))

lib/src/main/java/com/otaliastudios/transcoder/internal/Codecs.kt

@@ -1,16 +1,19 @@
 package com.otaliastudios.transcoder.internal
 
 import android.media.MediaCodec
+import android.media.MediaCodecInfo
+import android.media.MediaCodecList
 import android.media.MediaFormat
+import android.opengl.EGL14
 import android.view.Surface
+import com.otaliastudios.opengl.core.EglCore
+import com.otaliastudios.opengl.surface.EglOffscreenSurface
+import com.otaliastudios.opengl.surface.EglWindowSurface
 import com.otaliastudios.transcoder.common.TrackStatus
 import com.otaliastudios.transcoder.common.TrackType
-import com.otaliastudios.transcoder.internal.media.MediaFormatProvider
+import com.otaliastudios.transcoder.internal.media.MediaFormatConstants
 import com.otaliastudios.transcoder.internal.utils.Logger
 import com.otaliastudios.transcoder.internal.utils.TrackMap
-import com.otaliastudios.transcoder.internal.utils.trackMapOf
-import com.otaliastudios.transcoder.source.DataSource
-import com.otaliastudios.transcoder.strategy.TrackStrategy
 
 /**
  * Encoders are shared between segments. This is not strictly needed but it is more efficient
@@ -25,6 +28,16 @@ internal class Codecs(
         private val current: TrackMap<Int>
 ) {
 
+    internal class Surface(
+        val context: EglCore,
+        val window: EglWindowSurface,
+    ) {
+        fun release() {
+            window.release()
+            context.release()
+        }
+    }
+
     private val log = Logger("Codecs")
 
     val encoders = object : TrackMap<Pair<MediaCodec, Surface?>> {
@@ -40,9 +53,28 @@ internal class Codecs(
 
         private val lazyVideo by lazy {
             val format = tracks.outputFormats.video
+            val width = format.getInteger(MediaFormat.KEY_WIDTH)
+            val height = format.getInteger(MediaFormat.KEY_HEIGHT)
+            log.i("Destination video surface size: ${width}x${height} @ ${format.getInteger(MediaFormatConstants.KEY_ROTATION_DEGREES)}")
+            log.i("Destination video format: $format")
+
+            val allCodecs = MediaCodecList(MediaCodecList.REGULAR_CODECS)
+            val videoEncoders = allCodecs.codecInfos.filter { it.isEncoder && it.supportedTypes.any { it.startsWith("video/") } }
+            log.i("Available encoders: ${videoEncoders.joinToString { "${it.name} (${it.supportedTypes.joinToString()})" }}")
+
+            // Could consider MediaCodecList(MediaCodecList.REGULAR_CODECS).findEncoderForFormat(format)
+            // But it's trickier, for example, format should not include frame rate on API 21 and maybe other quirks.
             val codec = MediaCodec.createEncoderByType(format.getString(MediaFormat.KEY_MIME)!!)
             codec.configure(format, null, null, MediaCodec.CONFIGURE_FLAG_ENCODE)
-            codec to codec.createInputSurface()
+            log.i("Selected encoder ${codec.name}")
+            val surface = codec.createInputSurface()
+
+            log.i("Creating OpenGL context on ${Thread.currentThread()} (${surface.isValid})")
+            val eglContext = EglCore(EGL14.EGL_NO_CONTEXT, EglCore.FLAG_RECORDABLE)
+            val eglWindow = EglWindowSurface(eglContext, surface, true)
+            eglWindow.makeCurrent()
+
+            codec to Surface(eglContext, eglWindow)
         }
 
         override fun get(type: TrackType) = when (type) {

lib/src/main/java/com/otaliastudios/transcoder/internal/Segment.kt

@@ -3,7 +3,6 @@ package com.otaliastudios.transcoder.internal
 import com.otaliastudios.transcoder.common.TrackType
 import com.otaliastudios.transcoder.internal.pipeline.Pipeline
 import com.otaliastudios.transcoder.internal.pipeline.State
-import com.otaliastudios.transcoder.internal.utils.Logger
 
 internal class Segment(
         val type: TrackType,
@@ -27,7 +26,7 @@ internal class Segment(
     fun needsSleep(): Boolean {
         when(val s = state ?: return false) {
             is State.Ok -> return false
-            is State.Wait -> return s.sleep
+            is State.Failure -> return s.sleep
         }
     }
 

lib/src/main/java/com/otaliastudios/transcoder/internal/Segments.kt

@@ -21,7 +21,7 @@ internal class Segments(
 
     fun hasNext(type: TrackType): Boolean {
         if (!sources.has(type)) return false
-        log.v("hasNext($type): segment=${current.getOrNull(type)} lastIndex=${sources.getOrNull(type)?.lastIndex} canAdvance=${current.getOrNull(type)?.canAdvance()}")
+        // log.v("hasNext($type): segment=${current.getOrNull(type)} lastIndex=${sources.getOrNull(type)?.lastIndex} canAdvance=${current.getOrNull(type)?.canAdvance()}")
         val segment = current.getOrNull(type) ?: return true // not started
         val lastIndex = sources.getOrNull(type)?.lastIndex ?: return false // no track!
         return segment.canAdvance() || segment.index < lastIndex

lib/src/main/java/com/otaliastudios/transcoder/internal/audio/AudioEngine.kt

@@ -6,11 +6,8 @@ import android.view.Surface
 import com.otaliastudios.transcoder.internal.audio.remix.AudioRemixer
 import com.otaliastudios.transcoder.internal.codec.*
 import com.otaliastudios.transcoder.internal.pipeline.*
-import com.otaliastudios.transcoder.internal.utils.Logger
-import com.otaliastudios.transcoder.internal.utils.trackMapOf
 import com.otaliastudios.transcoder.resample.AudioResampler
 import com.otaliastudios.transcoder.stretch.AudioStretcher
-import java.util.concurrent.atomic.AtomicInteger
 import kotlin.math.ceil
 import kotlin.math.floor
 
@@ -30,18 +27,18 @@ internal class AudioEngine(
     private val MediaFormat.sampleRate get() = getInteger(KEY_SAMPLE_RATE)
     private val MediaFormat.channels get() = getInteger(KEY_CHANNEL_COUNT)
 
+    private val chunks = ChunkQueue(log)
+    private var readyToDrain = false
     private lateinit var rawFormat: MediaFormat
-    private lateinit var chunks: ChunkQueue
     private lateinit var remixer: AudioRemixer
 
     override fun handleSourceFormat(sourceFormat: MediaFormat): Surface? = null
 
     override fun handleRawFormat(rawFormat: MediaFormat) {
         log.i("handleRawFormat($rawFormat)")
-        check(!::rawFormat.isInitialized) { "handleRawFormat called twice: ${this.rawFormat} => $rawFormat"}
         this.rawFormat = rawFormat
-        remixer = AudioRemixer[rawFormat.channels, targetFormat.channels]
-        chunks = ChunkQueue(log, rawFormat.sampleRate, rawFormat.channels)
+        this.remixer = AudioRemixer[rawFormat.channels, targetFormat.channels]
+        this.readyToDrain = true
     }
 
     override fun enqueueEos(data: DecoderData) {
@@ -59,19 +56,24 @@ internal class AudioEngine(
     }
 
     override fun drain(): State<EncoderData> {
+        if (!readyToDrain) {
+            log.i("drain(): not ready, waiting...")
+            return State.Retry(false)
+        }
         if (chunks.isEmpty()) {
             // nothing was enqueued
             log.i("drain(): no chunks, waiting...")
-            return State.Wait(false)
+            return State.Retry(false)
         }
         val (outBytes, outId) = next.buffer() ?: return run {
             // dequeueInputBuffer failed
             log.i("drain(): no next buffer, waiting...")
-            State.Wait(true)
+            State.Retry(true)
         }
         val outBuffer = outBytes.asShortBuffer()
         return chunks.drain(
-                eos = State.Eos(EncoderData(outBytes, outId, 0))
+            eos = State.Eos(EncoderData(outBytes, outId, 0)),
+            format = rawFormat
         ) { inBuffer, timeUs, stretch ->
             val outSize = outBuffer.remaining()
             val inSize = inBuffer.remaining()

lib/src/main/java/com/otaliastudios/transcoder/internal/audio/chunks.kt

@@ -1,6 +1,10 @@
 package com.otaliastudios.transcoder.internal.audio
 
+import android.media.MediaFormat
+import android.media.MediaFormat.KEY_CHANNEL_COUNT
+import android.media.MediaFormat.KEY_SAMPLE_RATE
 import com.otaliastudios.transcoder.internal.utils.Logger
+import java.nio.ByteBuffer
 import java.nio.ShortBuffer
 
 private data class Chunk(
@@ -20,12 +24,9 @@ private data class Chunk(
  * big enough to contain the full processed size, in which case we want to consume only
  * part of the input buffer and keep it available for the next cycle.
  */
-internal class ChunkQueue(
-    private val log: Logger,
-    private val sampleRate: Int,
-    private val channels: Int
-) {
+internal class ChunkQueue(private val log: Logger) {
     private val queue = ArrayDeque<Chunk>()
+    private val pool = ShortBufferPool()
 
     fun isEmpty() = queue.isEmpty()
 
@@ -44,7 +45,7 @@ internal class ChunkQueue(
         queue.addLast(Chunk.Eos)
     }
 
-    fun <T> drain(eos: T, action: (buffer: ShortBuffer, timeUs: Long, timeStretch: Double) -> T): T {
+    fun <T> drain(format: MediaFormat, eos: T, action: (buffer: ShortBuffer, timeUs: Long, timeStretch: Double) -> T): T {
         val head = queue.removeFirst()
         if (head === Chunk.Eos) return eos
 
@@ -54,9 +55,17 @@ internal class ChunkQueue(
         // Action can reduce the limit for any reason. Restore it before comparing sizes.
         head.buffer.limit(limit)
         if (head.buffer.hasRemaining()) {
+            // We could technically hold onto the same chunk, but in practice it's better to
+            // release input buffers back to the decoder otherwise it can get stuck
             val consumed = size - head.buffer.remaining()
+            val sampleRate = format.getInteger(KEY_SAMPLE_RATE)
+            val channelCount = format.getInteger(KEY_CHANNEL_COUNT)
+            val buffer = pool.take(head.buffer)
+            head.release()
             queue.addFirst(head.copy(
-                    timeUs = shortsToUs(consumed, sampleRate, channels)
+                timeUs = shortsToUs(consumed, sampleRate, channelCount),
+                release = { pool.give(buffer) },
+                buffer = buffer
             ))
             log.v("[ChunkQueue] partially handled chunk at ${head.timeUs}us, ${head.buffer.remaining()} bytes left (${queue.size})")
         } else {
@@ -67,3 +76,27 @@ internal class ChunkQueue(
         return result
     }
 }
+
+
+class ShortBufferPool {
+    private val pool = mutableListOf<ShortBuffer>()
+
+    fun take(original: ShortBuffer): ShortBuffer {
+        val needed = original.remaining()
+        val index = pool.indexOfFirst { it.capacity() >= needed }
+        val memory = when {
+            index >= 0 -> pool.removeAt(index)
+            else -> ByteBuffer.allocateDirect(needed.coerceAtLeast(1024))
+                .order(original.order())
+                .asShortBuffer()
+        }
+        memory.put(original)
+        memory.flip()
+        return memory
+    }
+
+    fun give(buffer: ShortBuffer) {
+        buffer.clear()
+        pool.add(buffer)
+    }
+}

lib/src/main/java/com/otaliastudios/transcoder/internal/codec/Decoder.kt

@@ -10,11 +10,7 @@ import com.otaliastudios.transcoder.internal.data.ReaderData
 import com.otaliastudios.transcoder.internal.pipeline.Channel
 import com.otaliastudios.transcoder.internal.pipeline.QueuedStep
 import com.otaliastudios.transcoder.internal.pipeline.State
-import com.otaliastudios.transcoder.internal.utils.Logger
-import com.otaliastudios.transcoder.internal.utils.trackMapOf
 import java.nio.ByteBuffer
-import java.util.concurrent.atomic.AtomicInteger
-import kotlin.properties.Delegates
 import kotlin.properties.Delegates.observable
 
 
@@ -70,7 +66,7 @@ internal class Decoder(
             val buf = checkNotNull(codec.getInputBuffer(id)) { "inputBuffer($id) should not be null." }
             buf to id
         } else {
-            log.i("buffer() failed. dequeuedInputs=$dequeuedInputs dequeuedOutputs=$dequeuedOutputs")
+            log.i("buffer() failed with $id. dequeuedInputs=$dequeuedInputs dequeuedOutputs=$dequeuedOutputs")
             null
         }
     }
@@ -96,7 +92,7 @@ internal class Decoder(
         return when (result) {
             INFO_TRY_AGAIN_LATER -> {
                 log.i("drain(): got INFO_TRY_AGAIN_LATER, waiting.")
-                State.Wait(true)
+                State.Retry(true)
             }
             INFO_OUTPUT_FORMAT_CHANGED -> {
                 log.i("drain(): got INFO_OUTPUT_FORMAT_CHANGED, handling format and retrying. format=${codec.outputFormat}")
@@ -126,7 +122,7 @@ internal class Decoder(
                 } else {
                     // frame was dropped, no need to sleep
                     codec.releaseOutputBuffer(result, false)
-                    State.Wait(false)
+                    State.Retry(false)
                 }.also {
                     log.v("drain(): returning $it")
                 }

lib/src/main/java/com/otaliastudios/transcoder/internal/codec/Encoder.kt

@@ -3,19 +3,15 @@ package com.otaliastudios.transcoder.internal.codec
 import android.media.MediaCodec
 import android.media.MediaCodec.*
 import android.view.Surface
+import com.otaliastudios.opengl.surface.EglWindowSurface
 import com.otaliastudios.transcoder.common.TrackType
-import com.otaliastudios.transcoder.common.trackType
 import com.otaliastudios.transcoder.internal.Codecs
 import com.otaliastudios.transcoder.internal.data.WriterChannel
 import com.otaliastudios.transcoder.internal.data.WriterData
 import com.otaliastudios.transcoder.internal.pipeline.Channel
 import com.otaliastudios.transcoder.internal.pipeline.QueuedStep
 import com.otaliastudios.transcoder.internal.pipeline.State
-import com.otaliastudios.transcoder.internal.utils.Logger
-import com.otaliastudios.transcoder.internal.utils.trackMapOf
 import java.nio.ByteBuffer
-import java.util.concurrent.atomic.AtomicInteger
-import kotlin.properties.Delegates
 import kotlin.properties.Delegates.observable
 
 internal data class EncoderData(
@@ -27,15 +23,15 @@ internal data class EncoderData(
 }
 
 internal interface EncoderChannel : Channel {
-    val surface: Surface?
+    val surface: Codecs.Surface?
     fun buffer(): Pair<ByteBuffer, Int>?
 }
 
 internal class Encoder(
-        private val codec: MediaCodec,
-        override val surface: Surface?,
-        ownsCodecStart: Boolean,
-        private val ownsCodecStop: Boolean,
+    private val codec: MediaCodec,
+    override val surface: Codecs.Surface?,
+    ownsCodecStart: Boolean,
+    private val ownsCodecStop: Boolean,
 ) : QueuedStep<EncoderData, EncoderChannel, WriterData, WriterChannel>(
     when (surface) {
         null -> "AudioEncoder"
@@ -44,13 +40,12 @@ internal class Encoder(
 ), EncoderChannel {
 
     constructor(codecs: Codecs, type: TrackType) : this(
-            codecs.encoders[type].first,
-            codecs.encoders[type].second,
-            codecs.ownsEncoderStart[type],
-            codecs.ownsEncoderStop[type]
+        codecs.encoders[type].first,
+        codecs.encoders[type].second,
+        codecs.ownsEncoderStart[type],
+        codecs.ownsEncoderStop[type]
     )
 
-    private val type = if (surface != null) TrackType.VIDEO else TrackType.AUDIO
     private var dequeuedInputs by observable(0) { _, _, _ -> printDequeued() }
     private var dequeuedOutputs by observable(0) { _, _, _ -> printDequeued() }
     private fun printDequeued() {
@@ -61,9 +56,8 @@ internal class Encoder(
 
     private var info = BufferInfo()
 
-
     init {
-        log.i("Encoder: ownsStart=$ownsCodecStart ownsStop=$ownsCodecStop")
+        log.i("ownsStart=$ownsCodecStart ownsStop=$ownsCodecStop")
         if (ownsCodecStart) {
             codec.start()
         }
@@ -116,7 +110,7 @@ internal class Encoder(
                     State.Eos(WriterData(buffer, 0L, 0) {})
                 } else {
                     log.i("Can't dequeue output buffer: INFO_TRY_AGAIN_LATER")
-                    State.Wait(true)
+                    State.Retry(true)
                 }
             }
             INFO_OUTPUT_FORMAT_CHANGED -> {