[FFmpeg-devel] avformat: AMQP: add option delivery_mode

Signed-off-by: Levis Florian <levis.florian@gmail.com>
---
 doc/protocols.texi    | 13 +++++++++++++
 libavformat/libamqp.c |  6 +++++-
 2 files changed, 18 insertions(+), 1 deletion(-)

Hi Levis, 

Thanks for your patch.

On Mon, 15. Jun 11:16, Levis Florian wrote:
> Signed-off-by: Levis Florian <levis.florian@gmail.com>
> ---
>  doc/protocols.texi    | 13 +++++++++++++
>  libavformat/libamqp.c |  6 +++++-
>  2 files changed, 18 insertions(+), 1 deletion(-)
> 
> diff --git a/doc/protocols.texi b/doc/protocols.texi
> index 7aa758541c..336246e67a 100644
> --- a/doc/protocols.texi
> +++ b/doc/protocols.texi
> @@ -109,6 +109,19 @@ the received message may be truncated causing decoding errors.
>  The timeout in seconds during the initial connection to the broker. The
>  default value is rw_timeout, or 5 seconds if rw_timeout is not set.
>  
> +@item delivery_mode @var{mode}
> +Sets the delivery mode.
> +

> +The following values are recognized:
> +@table @samp
> +@item persistent
> +Delivery mode set to "persistent" (2). This is the default value.
> +
> +@item non-persistent
> +Delivery mode set to "non-persistent" (1)
> +
> +@end table
> +

Please add more details on trade-offs between modes.

>  @end table
>  
>  @section async
> diff --git a/libavformat/libamqp.c b/libavformat/libamqp.c
> index aaf0e51152..83de2229fb 100644
> --- a/libavformat/libamqp.c
> +++ b/libavformat/libamqp.c
> @@ -39,6 +39,7 @@ typedef struct AMQPContext {
>      int pkt_size;
>      int64_t connection_timeout;
>      int pkt_size_overflow;
> +    int delivery_mode;
>  } AMQPContext;
>  
>  #define STR_LEN           1024
> @@ -52,6 +53,9 @@ static const AVOption options[] = {
>      { "exchange", "Exchange to send/read packets", OFFSET(exchange), AV_OPT_TYPE_STRING, { .str = "amq.direct" }, 0, 0, .flags = D | E },
>      { "routing_key", "Key to filter streams", OFFSET(routing_key), AV_OPT_TYPE_STRING, { .str = "amqp" }, 0, 0, .flags = D | E },
>      { "connection_timeout", "Initial connection timeout", OFFSET(connection_timeout), AV_OPT_TYPE_DURATION, { .i64 = -1 }, -1, INT64_MAX, .flags = D | E},
> +    { "delivery_mode",  "Delivery mode", OFFSET(delivery_mode), AV_OPT_TYPE_INT, { .i64 = AMQP_DELIVERY_PERSISTENT }, 1, 2, .flags = E, "delivery_mode"},
> +    { "persistent",     "persistent delivery mode",     0, AV_OPT_TYPE_CONST, {.i64=AMQP_DELIVERY_PERSISTENT }, 0, 0, E, "delivery_mode" },
> +    { "non-persistent", "non-persistent delivery mode", 0, AV_OPT_TYPE_CONST, {.i64=AMQP_DELIVERY_NONPERSISTENT }, 0, 0, E, "delivery_mode" },
>      { NULL }

Is the goal of the non-persistent option to reduce disk io, or something else?
I ran a quick test to look at disk io, but didn't see a difference.

>  };
>  
> @@ -222,7 +226,7 @@ static int amqp_proto_write(URLContext *h, const unsigned char *buf, int size)
>  
>      props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
>      props.content_type = amqp_cstring_bytes("octet/stream");
> -    props.delivery_mode = 2; /* persistent delivery mode */
> +    props.delivery_mode = s->delivery_mode;
>  
>      ret = amqp_basic_publish(s->conn, DEFAULT_CHANNEL, amqp_cstring_bytes(s->exchange),
>                               amqp_cstring_bytes(s->routing_key), 0, 0,
> -- 
> 2.27.0
> 
> _______________________________________________
> ffmpeg-devel mailing list
> ffmpeg-devel@ffmpeg.org
> https://ffmpeg.org/mailman/listinfo/ffmpeg-devel
> 
> To unsubscribe, visit link above, or email
> ffmpeg-devel-request@ffmpeg.org with subject "unsubscribe".

Thanks,

(re-sending, forgot list, sorry Andriy for double sending)

Hi Andriy,

> Is the goal of the non-persistent option to reduce disk io, or something
else?
> I ran a quick test to look at disk io, but didn't see a difference.

TL;DR
-> it have effect on the broker (ex: RabbitMQ)
-> it should not have effect on the client (ffmpeg)

Explanations :

For RabbitMQ
- if the message is short (default 4 kbytes), it stays in RAM
- if not, it write it on disk, unless :
** the queue is flagged as non durable (queue not recreated if the node
reboot) (not sure about it, can't find it in the doc)
** the message itself is flagged as non-persistent, even if the queue is
durable

The informations I gave above are from RabbitMQ documentation
- part from here https://www.rabbitmq.com/memory-use.html#breakdown-queues
- here too https://www.rabbitmq.com/persistence-conf.html

In my case :
- I read video streams, full fps (25), extracting each frame and sending
them to queue
- I can't use something like webp because encoding is too long (no GPU
available)
- As a result, I have to use JPG, resulting in serialized message ~40kbytes
(If i'm right, RMQ store them as base64 encoded)
- As a result, with delivery_mode=2, the system have a lot of I/O pressure
(lot of iowait)

-> delivery_mode=2 (non-persistent) solved the problem


Regards,
--
Florian LEVIS


Le sam. 20 juin 2020 à 07:36, Andriy Gelman <andriy.gelman@gmail.com> a
écrit :

> Hi Levis,
>
> Thanks for your patch.
>
> On Mon, 15. Jun 11:16, Levis Florian wrote:
> > Signed-off-by: Levis Florian <levis.florian@gmail.com>
> > ---
> >  doc/protocols.texi    | 13 +++++++++++++
> >  libavformat/libamqp.c |  6 +++++-
> >  2 files changed, 18 insertions(+), 1 deletion(-)
> >
> > diff --git a/doc/protocols.texi b/doc/protocols.texi
> > index 7aa758541c..336246e67a 100644
> > --- a/doc/protocols.texi
> > +++ b/doc/protocols.texi
> > @@ -109,6 +109,19 @@ the received message may be truncated causing
> decoding errors.
> >  The timeout in seconds during the initial connection to the broker. The
> >  default value is rw_timeout, or 5 seconds if rw_timeout is not set.
> >
> > +@item delivery_mode @var{mode}
> > +Sets the delivery mode.
> > +
>
> > +The following values are recognized:
> > +@table @samp
> > +@item persistent
> > +Delivery mode set to "persistent" (2). This is the default value.
> > +
> > +@item non-persistent
> > +Delivery mode set to "non-persistent" (1)
> > +
> > +@end table
> > +
>
> Please add more details on trade-offs between modes.
>
> >  @end table
> >
> >  @section async
> > diff --git a/libavformat/libamqp.c b/libavformat/libamqp.c
> > index aaf0e51152..83de2229fb 100644
> > --- a/libavformat/libamqp.c
> > +++ b/libavformat/libamqp.c
> > @@ -39,6 +39,7 @@ typedef struct AMQPContext {
> >      int pkt_size;
> >      int64_t connection_timeout;
> >      int pkt_size_overflow;
> > +    int delivery_mode;
> >  } AMQPContext;
> >
> >  #define STR_LEN           1024
> > @@ -52,6 +53,9 @@ static const AVOption options[] = {
> >      { "exchange", "Exchange to send/read packets", OFFSET(exchange),
> AV_OPT_TYPE_STRING, { .str = "amq.direct" }, 0, 0, .flags = D | E },
> >      { "routing_key", "Key to filter streams", OFFSET(routing_key),
> AV_OPT_TYPE_STRING, { .str = "amqp" }, 0, 0, .flags = D | E },
> >      { "connection_timeout", "Initial connection timeout",
> OFFSET(connection_timeout), AV_OPT_TYPE_DURATION, { .i64 = -1 }, -1,
> INT64_MAX, .flags = D | E},
> > +    { "delivery_mode",  "Delivery mode", OFFSET(delivery_mode),
> AV_OPT_TYPE_INT, { .i64 = AMQP_DELIVERY_PERSISTENT }, 1, 2, .flags = E,
> "delivery_mode"},
> > +    { "persistent",     "persistent delivery mode",     0,
> AV_OPT_TYPE_CONST, {.i64=AMQP_DELIVERY_PERSISTENT }, 0, 0, E,
> "delivery_mode" },
> > +    { "non-persistent", "non-persistent delivery mode", 0,
> AV_OPT_TYPE_CONST, {.i64=AMQP_DELIVERY_NONPERSISTENT }, 0, 0, E,
> "delivery_mode" },
> >      { NULL }
>
> Is the goal of the non-persistent option to reduce disk io, or something
> else?
> I ran a quick test to look at disk io, but didn't see a difference.
>
> >  };
> >
> > @@ -222,7 +226,7 @@ static int amqp_proto_write(URLContext *h, const
> unsigned char *buf, int size)
> >
> >      props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG |
> AMQP_BASIC_DELIVERY_MODE_FLAG;
> >      props.content_type = amqp_cstring_bytes("octet/stream");
> > -    props.delivery_mode = 2; /* persistent delivery mode */
> > +    props.delivery_mode = s->delivery_mode;
> >
> >      ret = amqp_basic_publish(s->conn, DEFAULT_CHANNEL,
> amqp_cstring_bytes(s->exchange),
> >                               amqp_cstring_bytes(s->routing_key), 0, 0,
> > --
> > 2.27.0
> >
> > _______________________________________________
> > ffmpeg-devel mailing list
> > ffmpeg-devel@ffmpeg.org
> > https://ffmpeg.org/mailman/listinfo/ffmpeg-devel
> >
> > To unsubscribe, visit link above, or email
> > ffmpeg-devel-request@ffmpeg.org with subject "unsubscribe".
>
> Thanks,
> --
> Andriy
>

On Sat, 20. Jun 12:28, Florian Levis wrote:
> (re-sending, forgot list, sorry Andriy for double sending)
> 
> Hi Andriy,
> 
> > Is the goal of the non-persistent option to reduce disk io, or something
> else?
> > I ran a quick test to look at disk io, but didn't see a difference.
> 
> TL;DR
> -> it have effect on the broker (ex: RabbitMQ)
> -> it should not have effect on the client (ffmpeg)
> 
> Explanations :
> 
> For RabbitMQ
> - if the message is short (default 4 kbytes), it stays in RAM
> - if not, it write it on disk, unless :
> ** the queue is flagged as non durable (queue not recreated if the node
> reboot) (not sure about it, can't find it in the doc)
> ** the message itself is flagged as non-persistent, even if the queue is
> durable
> 
> The informations I gave above are from RabbitMQ documentation
> - part from here https://www.rabbitmq.com/memory-use.html#breakdown-queues
> - here too https://www.rabbitmq.com/persistence-conf.html
> 
> In my case :
> - I read video streams, full fps (25), extracting each frame and sending
> them to queue
> - I can't use something like webp because encoding is too long (no GPU
> available)
> - As a result, I have to use JPG, resulting in serialized message ~40kbytes
> (If i'm right, RMQ store them as base64 encoded)
> - As a result, with delivery_mode=2, the system have a lot of I/O pressure
> (lot of iowait)
> 
> -> delivery_mode=2 (non-persistent) solved the problem

Thanks for explanation.
I also had to set lazy mode on the queue. Otherwise all messages were in memory for
me.

> > >
> > > +@item delivery_mode @var{mode}
> > > +Sets the delivery mode.
> > > +
> >
> > > +The following values are recognized:
> > > +@table @samp
> > > +@item persistent
> > > +Delivery mode set to "persistent" (2). This is the default value.
> > > +
> > > +@item non-persistent
> > > +Delivery mode set to "non-persistent" (1)
> > > +
> > > +@end table
> > > +

Please mention the difference: i.e. for the persistent mode the messages may be written
to disk depending on server's setup. For the non-persistent mode the queued
messages are in memory.

> > > +    { "delivery_mode",  "Delivery mode", OFFSET(delivery_mode),
> > AV_OPT_TYPE_INT, { .i64 = AMQP_DELIVERY_PERSISTENT }, 1, 2, .flags = E,

> > "delivery_mode"},
> > > +    { "persistent",     "persistent delivery mode",     0,
nit:                            ^capitalize                             

> > AV_OPT_TYPE_CONST, {.i64=AMQP_DELIVERY_PERSISTENT }, 0, 0, E,
                        ^add a space. same for below.

> > "delivery_mode" },
> > > +    { "non-persistent", "non-persistent delivery mode", 0,
> > AV_OPT_TYPE_CONST, {.i64=AMQP_DELIVERY_NONPERSISTENT }, 0, 0, E,

Thanks,