feat: add non-terminal output polling

Author: SaicharanKandukuri

include/fma/client.h

@@ -25,6 +25,7 @@ int fma_client_queue_packet(struct fma_client *client, const void *data,
                             size_t size, int64_t pts_us, uint32_t flags);
 int fma_client_drain(struct fma_client *client);
 int fma_client_flush(struct fma_client *client);
+int fma_client_poll_output(struct fma_client *client, uint32_t timeout_ms);
 int fma_client_release_frame(struct fma_client *client, uint32_t slot);
 int fma_client_receive(struct fma_client *client, struct fma_message *message);
 

include/fma/protocol.h

@@ -10,7 +10,7 @@ extern "C" {
 
 #define FMA_MAGIC UINT32_C(0x31414d46) /* FMA1 on a little-endian wire. */
 #define FMA_PROTOCOL_MAJOR 1u
-#define FMA_PROTOCOL_MINOR 0u
+#define FMA_PROTOCOL_MINOR 1u
 #define FMA_WIRE_HEADER_SIZE 48u
 #define FMA_MAX_PAYLOAD (8u * 1024u * 1024u)
 #define FMA_MAX_FDS 4u
@@ -37,6 +37,8 @@ enum fma_message_type {
     FMA_MSG_OUTPUT_EOS = 15,
     FMA_MSG_CLOSE = 16,
     FMA_MSG_ERROR = 17,
+    FMA_MSG_POLL_OUTPUT = 18,
+    FMA_MSG_POLL_DONE = 19,
 };
 
 enum fma_codec {
@@ -61,6 +63,7 @@ enum fma_packet_flags {
 enum fma_capability_flags {
     FMA_CAP_SHARED_FRAME_POOL = 1u << 0,
     FMA_CAP_CAN_FLUSH = 1u << 1,
+    FMA_CAP_CAN_POLL = 1u << 2,
 };
 
 struct fma_message {

src/android/fake-media-acceld.c

@@ -444,7 +444,8 @@ static int serve_client(int fd) {
             struct fma_capabilities caps = {
                 .decoder_mask = probe_decoders(),
                 .pixel_format_mask = 1u << (FMA_PIXFMT_NV12 - 1u),
-                .flags = FMA_CAP_SHARED_FRAME_POOL | FMA_CAP_CAN_FLUSH,
+                .flags = FMA_CAP_SHARED_FRAME_POOL | FMA_CAP_CAN_FLUSH |
+                         FMA_CAP_CAN_POLL,
                 .max_width = 0,
                 .max_height = 0,
             };
@@ -520,6 +521,23 @@ static int serve_client(int fd) {
                 }
             }
             break;
+        case FMA_MSG_POLL_OUTPUT: {
+            if (!session.started || request.payload_size != 4) {
+                result = send_error(fd, &request, "invalid output poll");
+                break;
+            }
+            uint32_t timeout_ms = fma_get_u32(request.payload);
+            if (timeout_ms > 1000)
+                timeout_ms = 1000;
+            int output = emit_output(fd, &request, &session,
+                                     (int64_t)timeout_ms * 1000, false);
+            if (output < 0)
+                result = send_error(fd, &request, "MediaCodec output poll failed");
+            else
+                result = send_reply(fd, &request, FMA_MSG_POLL_DONE, NULL, 0,
+                                    session.id, -1);
+            break;
+        }
         case FMA_MSG_CLOSE:
             fma_message_release(&request);
             destroy_decoder(&session);

src/client/client.c

@@ -142,6 +142,13 @@ int fma_client_flush(struct fma_client *client) {
     return 0;
 }
 
+int fma_client_poll_output(struct fma_client *client, uint32_t timeout_ms) {
+    uint8_t payload[4];
+    fma_put_u32(payload, timeout_ms);
+    return send_simple(client, FMA_MSG_POLL_OUTPUT, payload, sizeof(payload),
+                       0, 0, NULL);
+}
+
 int fma_client_release_frame(struct fma_client *client, uint32_t slot) {
     uint8_t payload[4];
     fma_put_u32(payload, slot);

src/fake/fake-media-acceld.c

@@ -21,7 +21,7 @@ struct fake_session {
     uint8_t *pool_map;
     size_t pool_bytes;
     uint32_t next_slot;
-    bool slots[16];
+    bool slots[FMA_MAX_SLOTS];
 };
 
 static volatile sig_atomic_t running = 1;
@@ -177,7 +177,8 @@ static int handle_client(int fd) {
                                 FMA_CODEC_BIT(FMA_CODEC_VP9) |
                                 FMA_CODEC_BIT(FMA_CODEC_AV1),
                 .pixel_format_mask = 1u << (FMA_PIXFMT_NV12 - 1u),
-                .flags = FMA_CAP_SHARED_FRAME_POOL | FMA_CAP_CAN_FLUSH,
+                .flags = FMA_CAP_SHARED_FRAME_POOL | FMA_CAP_CAN_FLUSH |
+                         FMA_CAP_CAN_POLL,
                 .max_width = 8192,
                 .max_height = 8192,
             };
@@ -228,6 +229,13 @@ static int handle_client(int fd) {
             result = send_reply(fd, &request, FMA_MSG_OUTPUT_EOS, NULL, 0,
                                 session.id, -1);
             break;
+        case FMA_MSG_POLL_OUTPUT:
+            if (request.payload_size != 4)
+                result = send_error(fd, &request, "invalid poll request");
+            else
+                result = send_reply(fd, &request, FMA_MSG_POLL_DONE, NULL, 0,
+                                    session.id, -1);
+            break;
         case FMA_MSG_CLOSE:
             fma_message_release(&request);
             destroy_pool(&session);

tools/fma-decode.c

@@ -115,7 +115,7 @@ static uint32_t parse_codec(const char *name) {
 static int process_message(struct fma_client *client,
                            const struct fma_frame_pool *pool, uint8_t *pool_map,
                            FILE *output, uint64_t *frames, bool *packet_ack,
-                           bool *eos) {
+                           bool *eos, bool *poll_done) {
     struct fma_message message;
     if (fma_client_receive(client, &message) < 0)
         return -1;
@@ -140,6 +140,8 @@ static int process_message(struct fma_client *client,
         *packet_ack = true;
     } else if (message.type == FMA_MSG_OUTPUT_EOS) {
         *eos = true;
+    } else if (message.type == FMA_MSG_POLL_DONE) {
+        *poll_done = true;
     } else if (message.type == FMA_MSG_ERROR) {
         fprintf(stderr, "daemon: %.*s\n", (int)message.payload_size,
                 message.payload ? (char *)message.payload : "error");
@@ -242,6 +244,7 @@ int main(int argc, char **argv) {
         for (size_t i = 0; i < unit_count; ++i) {
             int64_t pts_us = (int64_t)(i * UINT64_C(1000000) / fps);
             bool packet_ack = false;
+            bool poll_done = false;
             if (fma_client_queue_packet(&client, input + units[i].offset,
                                         units[i].size, pts_us, 0) < 0) {
                 perror("decode");
@@ -251,17 +254,28 @@ int main(int argc, char **argv) {
             input_bytes += units[i].size;
             while (!packet_ack &&
                    process_message(&client, &pool, pool_map, output, &frames,
-                                   &packet_ack, &eos) == 0) {}
+                                   &packet_ack, &eos, &poll_done) == 0) {}
             if (!packet_ack) {
                 perror("decode");
                 failed = true;
                 break;
             }
         }
+        if (!failed && fma_client_poll_output(&client, 0) == 0) {
+            bool packet_ack = false;
+            bool poll_done = false;
+            while (!poll_done &&
+                   process_message(&client, &pool, pool_map, output, &frames,
+                                   &packet_ack, &eos, &poll_done) == 0) {}
+            if (!poll_done)
+                failed = true;
+        }
         if (!failed && fma_client_drain(&client) == 0) {
             bool packet_ack = false;
+            bool poll_done = false;
             while (!eos && process_message(&client, &pool, pool_map, output,
-                                           &frames, &packet_ack, &eos) == 0) {}
+                                           &frames, &packet_ack, &eos,
+                                           &poll_done) == 0) {}
         }
         if (!eos)
             failed = true;