From patchwork Thu Aug 29 18:34:30 2019 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Andriy Gelman X-Patchwork-Id: 14796 Return-Path: X-Original-To: patchwork@ffaux-bg.ffmpeg.org Delivered-To: patchwork@ffaux-bg.ffmpeg.org Received: from ffbox0-bg.mplayerhq.hu (ffbox0-bg.ffmpeg.org [79.124.17.100]) by ffaux.localdomain (Postfix) with ESMTP id 5846244A06F for ; Thu, 29 Aug 2019 21:34:40 +0300 (EEST) Received: from [127.0.1.1] (localhost [127.0.0.1]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTP id 3A294680D0B; Thu, 29 Aug 2019 21:34:40 +0300 (EEST) X-Original-To: ffmpeg-devel@ffmpeg.org Delivered-To: ffmpeg-devel@ffmpeg.org Received: from mail-qt1-f174.google.com (mail-qt1-f174.google.com [209.85.160.174]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTPS id B709A680261 for ; Thu, 29 Aug 2019 21:34:32 +0300 (EEST) Received: by mail-qt1-f174.google.com with SMTP id g4so4790275qtq.7 for ; Thu, 29 Aug 2019 11:34:32 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=gmail.com; s=20161025; h=date:from:to:subject:message-id:reply-to:mime-version :content-disposition:user-agent; bh=e2M691OGkxOtGXpaO3iP6sMufZ1Cpjw6V0TenfMMzDA=; b=EalqDEc78D9lSP0+hCAcna3T12vY/UOfTQmDPXU2to+FEH3osFbgp0qZ4D2lAbFhDV Gcx8p6a5LeU94J1ee7e6xR1zJX67vw/AjtbQWR8vqT8LSJGGdQUG5G9A57t6pySQgqa3 2TziMCk//WLGcbF5h1f+nNaqK35cZz7c1wTaNhGEipEZxjqbAN+5fAXuHkN609EMnLPL eSPeCLTxnvSwWK8C8FQLll4JRFhYQeArWSC3ObRhfIwVAei86V3YUCgA6Dy49vGl+qAv P2kZn5Yl0O5g9mEt1WY4CGFNzSl4WW/zykLyPILks5k+UkpMZkkNeGk/2Tss1RA4TpU7 XSXA== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20161025; h=x-gm-message-state:date:from:to:subject:message-id:reply-to :mime-version:content-disposition:user-agent; bh=e2M691OGkxOtGXpaO3iP6sMufZ1Cpjw6V0TenfMMzDA=; b=S7u0iQw/h1cyzudtXt7WSdMyYZdS1+2x0SlCZXqnpMvHGNkbBWPIz382VNcY6qsWXV BbLVXUSA+NWeb8xQEYZfbxrNFQz1wdLTm4Cb2gjq1IVYYmgHGSiLny9xGPreTSMjo1qe BkXW3H3rvqjt61Hmlm1rhbc7jFUx6rdXXYDtHXWyhMEzifjKiFOUNV/0HzN2TxKx17oG RnQzeUXV6cshLjfubTzgRAlqMRV2ZLKijnAtPN0aKS7Gs2YCH7zSNHAyI73hYxWZ2i14 qNJwNtjsmmeCMfbDiAJtMfT0+pLYk7vVUZHSiSY1kQugVnLnPmamZMO577+3Wdxb+Asz l6Sg== X-Gm-Message-State: APjAAAWwqLRkJ70RYPwE+QUbckygqWXoHts+MpqwfT9CBC+lFzGv1Frd TXXwdZjBUCBLvZ8oQ7NZbJ6Awlym X-Google-Smtp-Source: APXvYqx0KTo4GKiLO2FszKOAy1+2aNNCTEMWlai0Rp7MYcPzYg61DXP/VZ+7VrN5ZNIDP9R1MrLDgQ== X-Received: by 2002:ac8:2d83:: with SMTP id p3mr11627157qta.237.1567103671047; Thu, 29 Aug 2019 11:34:31 -0700 (PDT) Received: from manj (c-71-232-27-28.hsd1.ma.comcast.net. [71.232.27.28]) by smtp.gmail.com with ESMTPSA id z72sm1308581qka.115.2019.08.29.11.34.30 for (version=TLS1_3 cipher=TLS_AES_256_GCM_SHA384 bits=256/256); Thu, 29 Aug 2019 11:34:30 -0700 (PDT) Date: Thu, 29 Aug 2019 14:34:30 -0400 From: Andriy Gelman To: ffmpeg-devel@ffmpeg.org Message-ID: <20190829183430.7bvwklitrkv3druq@manj> MIME-Version: 1.0 Content-Disposition: inline User-Agent: NeoMutt/20180716-1344-11488f-dirty Subject: [FFmpeg-devel] [PATCH v4] Add ZeroMQ as protocol option X-BeenThere: ffmpeg-devel@ffmpeg.org X-Mailman-Version: 2.1.20 Precedence: list List-Id: FFmpeg development discussions and patches List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , Reply-To: FFmpeg development discussions and patches Errors-To: ffmpeg-devel-bounces@ffmpeg.org Sender: "ffmpeg-devel" Changes in v4: - Use polling instead of non-blocking option for socket read/write operations. - Added pkt_size, timeout_send, timeout_recv options. Updated documentation for new options. - Removed redundant memcpy now that pkt_size is set. - Check that context is successfully created. - Added minimum version requirement for libzmq in configure (set to 4.3.1) - Removed link to ubuntu compilation guide from documentation. - Use zmq_errno() to get errno. - Updated Changelog and bumped micro version number. - Added author name. Thanks, Andriy From ca5aa5149b97152b15457b2f40c753c80e68e3de Mon Sep 17 00:00:00 2001 From: Andriy Gelman Date: Tue, 30 Jul 2019 14:39:32 -0400 Subject: [PATCH] libavformat: Add ZeroMQ as a protocol option Currently multiple clients are only supported by using a multicast destination address. An alternative is to stream to a server which re-distributes the content. This commit adds ZeroMQ as a protocol option, which allows multiple clients to connect to a single ffmpeg instance. --- Changelog | 1 + configure | 4 +- doc/general.texi | 1 + doc/protocols.texi | 48 +++++++++++ libavformat/Makefile | 1 + libavformat/libzmq.c | 184 ++++++++++++++++++++++++++++++++++++++++ libavformat/protocols.c | 1 + libavformat/version.h | 2 +- 8 files changed, 240 insertions(+), 2 deletions(-) create mode 100644 libavformat/libzmq.c diff --git a/Changelog b/Changelog index f1cddc4274..1ccaca0b60 100644 --- a/Changelog +++ b/Changelog @@ -7,6 +7,7 @@ version : - Intel QSV-accelerated VP9 decoding - support for TrueHD in mp4 - Supoort AMD AMF encoder on Linux (via Vulkan) +- ZeroMQ protocol version 4.2: diff --git a/configure b/configure index 691ae9bd4f..2cc6323534 100755 --- a/configure +++ b/configure @@ -3413,6 +3413,8 @@ libsrt_protocol_deps="libsrt" libsrt_protocol_select="network" libssh_protocol_deps="libssh" libtls_conflict="openssl gnutls mbedtls" +libzmq_protocol_deps="libzmq" +libzmq_protocol_select="network" # filters afftdn_filter_deps="avcodec" @@ -6323,7 +6325,7 @@ enabled libxavs && require libxavs "stdint.h xavs.h" xavs_encoder_enco enabled libxavs2 && require_pkg_config libxavs2 "xavs2 >= 1.3.0" "stdint.h xavs2.h" xavs2_api_get enabled libxvid && require libxvid xvid.h xvid_global -lxvidcore enabled libzimg && require_pkg_config libzimg "zimg >= 2.7.0" zimg.h zimg_get_api_version -enabled libzmq && require_pkg_config libzmq libzmq zmq.h zmq_ctx_new +enabled libzmq && require_pkg_config libzmq "libzmq >= 4.3.1" zmq.h zmq_ctx_new enabled libzvbi && require_pkg_config libzvbi zvbi-0.2 libzvbi.h vbi_decoder_new && { test_cpp_condition libzvbi.h "VBI_VERSION_MAJOR > 0 || VBI_VERSION_MINOR > 2 || VBI_VERSION_MINOR == 2 && VBI_VERSION_MICRO >= 28" || enabled gpl || die "ERROR: libzvbi requires version 0.2.28 or --enable-gpl."; } diff --git a/doc/general.texi b/doc/general.texi index d0c3525e02..2744c238cf 100644 --- a/doc/general.texi +++ b/doc/general.texi @@ -1339,6 +1339,7 @@ performance on systems without hardware floating point support). @item TCP @tab X @item TLS @tab X @item UDP @tab X +@item ZMQ @tab E @end multitable @code{X} means that the protocol is supported. diff --git a/doc/protocols.texi b/doc/protocols.texi index 3e4e7af3d4..4ce33d60da 100644 --- a/doc/protocols.texi +++ b/doc/protocols.texi @@ -1728,4 +1728,52 @@ Timeout in ms. Create the Unix socket in listening mode. @end table +@section libzmq + +ZeroMQ asynchronous messaging library. + +This library supports unicast streaming to multiple clients without relying on +an external server. + +The required syntax for streaming or connecting to a stream is: +@example +zmq:tcp://ip-address:port +@end example + +Example: +Create a localhost stream on port 5555: +@example +ffmpeg -re -i input -f mpegts zmq:tcp://127.0.0.1:5555 +@end example + +Multiple clients may connect to the stream using: +@example +ffplay zmq:tcp://127.0.0.1:5555 +@end example + +Streaming to multiple clients is implemented using a ZeroMQ Pub-Sub pattern. +The server side binds to a port and publishes data. Clients connect to the +server (via IP address/port) and subscribe to the stream. The order in which +the server and client start generally does not matter. + +ffmpeg must be compiled with the --enable-libzmq option to support +this protocol. + +Options can be set on the @command{ffmpeg}/@command{ffplay} command +line. The following options are supported: + +@table @option + +@item pkt_size +Forces the maximum packet size for sending/receiving data. The default value is 32,768 bytes. On the server side, this sets the maximum size of sent packets via ZeroMQ. On the clients, it sets an internal buffer size for receiving packets. Note that pkt_size on the clients should be equal to or greater than pkt_size on the server. Otherwise the received message may be truncated causing decoding errors. + +@item timeout_send +Polling timeout (in milliseconds) on the server side. The default value is 0 (non-blocking operation). A value of -1 means that the polling operation will block infinitly until the packet can be sent. + +@item timeout_recv +Polling timeout (in milliseconds) on the client side. The default value is 0 (non-blocking operation). A value of -1 means that the polling operation will block infintly until a packet can be received. + +@end table + + @c man end PROTOCOLS diff --git a/libavformat/Makefile b/libavformat/Makefile index a434b005a4..efa3a112ae 100644 --- a/libavformat/Makefile +++ b/libavformat/Makefile @@ -631,6 +631,7 @@ OBJS-$(CONFIG_LIBRTMPTE_PROTOCOL) += librtmp.o OBJS-$(CONFIG_LIBSMBCLIENT_PROTOCOL) += libsmbclient.o OBJS-$(CONFIG_LIBSRT_PROTOCOL) += libsrt.o OBJS-$(CONFIG_LIBSSH_PROTOCOL) += libssh.o +OBJS-$(CONFIG_LIBZMQ_PROTOCOL) += libzmq.o # libavdevice dependencies OBJS-$(CONFIG_IEC61883_INDEV) += dv.o diff --git a/libavformat/libzmq.c b/libavformat/libzmq.c new file mode 100644 index 0000000000..fb25ae4428 --- /dev/null +++ b/libavformat/libzmq.c @@ -0,0 +1,184 @@ +/* + * ZeroMQ Protocol + * Copyright (c) 2019 Andriy Gelman + * + * This file is part of FFmpeg. + * + * FFmpeg is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * FFmpeg is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with FFmpeg; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA + */ + +#include +#include "url.h" +#include "network.h" +#include "libavutil/avstring.h" +#include "libavutil/opt.h" + +#define ZMQ_STRERROR zmq_strerror(zmq_errno()) + +typedef struct ZMQContext { + const AVClass *class; + void *context; + void *socket; + int timeout_send; + int timeout_recv; + int pkt_size; + int pkt_size_overflow; /*keep track of the largest packet during overflow*/ +} ZMQContext; + +#define OFFSET(x) offsetof(ZMQContext, x) +#define D AV_OPT_FLAG_DECODING_PARAM +#define E AV_OPT_FLAG_ENCODING_PARAM +static const AVOption options[] = { + { "timeout_send", "timeout (milliseconds) of socket send", OFFSET(timeout_send), AV_OPT_TYPE_INT, { .i64 = 0 }, -1, INT_MAX, .flags = E }, + { "timeout_recv", "timeout (milliseconds) of socket read", OFFSET(timeout_recv), AV_OPT_TYPE_INT, { .i64 = 0 }, -1, INT_MAX, .flags = D }, + { "pkt_size", "Maximum send/read packet size", OFFSET(pkt_size), AV_OPT_TYPE_INT, { .i64 = 32768 }, -1, INT_MAX, .flags = D | E }, + { NULL } +}; + +static int ff_zmq_open(URLContext *h, const char *uri, int flags) +{ + int ret; + ZMQContext *s = h->priv_data; + s->pkt_size_overflow = 0; + h->is_streamed = 1; + + if (s->pkt_size > 0) + h->max_packet_size = s->pkt_size; + + s->context = zmq_ctx_new(); + if (!s->context) { + /*errno not set on failure during zmq_ctx_new()*/ + av_log(h, AV_LOG_ERROR, "Error occured during zmq_ctx_new()\n"); + return AVERROR_EXTERNAL; + } + + av_strstart(uri, "zmq:", &uri); + + /*publish during write*/ + if (h->flags & AVIO_FLAG_WRITE) { + s->socket = zmq_socket(s->context, ZMQ_PUB); + if (!s->socket) { + av_log(h, AV_LOG_ERROR, "Error occured during zmq_socket(): %s\n", ZMQ_STRERROR); + zmq_ctx_destroy(s->context); + return AVERROR_EXTERNAL; + } + + ret = zmq_bind(s->socket, uri); + if (ret == -1) { + av_log(h, AV_LOG_ERROR, "Error occured during zmq_bind(): %s\n", ZMQ_STRERROR); + zmq_close(s->socket); + zmq_ctx_destroy(s->context); + return AVERROR_EXTERNAL; + } + } + + /*subscribe for read*/ + if (h->flags & AVIO_FLAG_READ) { + s->socket = zmq_socket(s->context, ZMQ_SUB); + if (!s->socket) { + av_log(h, AV_LOG_ERROR, "Error occured during zmq_socket(): %s\n", ZMQ_STRERROR); + zmq_ctx_destroy(s->context); + return AVERROR_EXTERNAL; + } + + zmq_setsockopt(s->socket, ZMQ_SUBSCRIBE, "", 0); + ret = zmq_connect(s->socket, uri); + if (ret == -1) { + av_log(h, AV_LOG_ERROR, "Error occured during zmq_connect(): %s\n", ZMQ_STRERROR); + zmq_close(s->socket); + zmq_ctx_destroy(s->context); + return AVERROR_EXTERNAL; + } + } + return 0; +} + +static int ff_zmq_write(URLContext *h, const unsigned char *buf, int size) +{ + int ret; + ZMQContext *s = h->priv_data; + + zmq_pollitem_t items = { s->socket, 0, ZMQ_POLLOUT, 0 }; + + ret = zmq_poll(&items, 1, s->timeout_send); + if (ret == -1) { + av_log(h, AV_LOG_ERROR, "Error occured during zmq_poll(): %s\n", ZMQ_STRERROR); + return AVERROR_EXTERNAL; + } + if (!(items.revents & ZMQ_POLLOUT)) + return AVERROR(EAGAIN); + + ret = zmq_send(s->socket, buf, size, 0); + if (ret == -1) { + av_log(h, AV_LOG_ERROR, "Error occured during zmq_send(): %s\n", ZMQ_STRERROR); + return AVERROR_EXTERNAL; + } + return ret; /*number of bytes sent*/ +} + +static int ff_zmq_read(URLContext *h, unsigned char *buf, int size) +{ + int ret; + ZMQContext *s = h->priv_data; + + zmq_pollitem_t items = { s->socket, 0, ZMQ_POLLIN, 0 }; + + ret = zmq_poll(&items, 1, s->timeout_recv); + if (ret == -1) { + av_log(h, AV_LOG_ERROR, "Error occured during zmq_poll(): %s\n", ZMQ_STRERROR); + return AVERROR_EXTERNAL; + } + if (!(items.revents & ZMQ_POLLIN)) + return AVERROR(EAGAIN); + + ret = zmq_recv(s->socket, buf, size, 0); + if (ret == -1) { + av_log(h, AV_LOG_ERROR, "Error occured during zmq_recv(): %s\n", ZMQ_STRERROR); + return AVERROR_EXTERNAL; + } + + if (ret > size) { + s->pkt_size_overflow = FFMAX(s->pkt_size_overflow, ret); + av_log(h, AV_LOG_WARNING, "Message exceeds available space in the buffer. Message will be truncated. Setting -pkt_size %d may resolve the issue.\n", s->pkt_size_overflow); + ret = size; + } + return ret; /*number of bytes read*/ +} + +static int ff_zmq_close(URLContext *h) +{ + ZMQContext *s = h->priv_data; + zmq_close(s->socket); + zmq_ctx_destroy(s->context); + return 0; +} + +static const AVClass zmq_context_class = { + .class_name = "zmq", + .item_name = av_default_item_name, + .option = options, + .version = LIBAVUTIL_VERSION_INT, +}; + +const URLProtocol ff_libzmq_protocol = { + .name = "zmq", + .url_close = ff_zmq_close, + .url_open = ff_zmq_open, + .url_read = ff_zmq_read, + .url_write = ff_zmq_write, + .priv_data_size = sizeof(ZMQContext), + .priv_data_class = &zmq_context_class, + .flags = URL_PROTOCOL_FLAG_NETWORK, +}; diff --git a/libavformat/protocols.c b/libavformat/protocols.c index ad95659795..face5b29b5 100644 --- a/libavformat/protocols.c +++ b/libavformat/protocols.c @@ -68,6 +68,7 @@ extern const URLProtocol ff_librtmpte_protocol; extern const URLProtocol ff_libsrt_protocol; extern const URLProtocol ff_libssh_protocol; extern const URLProtocol ff_libsmbclient_protocol; +extern const URLProtocol ff_libzmq_protocol; #include "libavformat/protocol_list.c" diff --git a/libavformat/version.h b/libavformat/version.h index af0db1e970..edfa73fb97 100644 --- a/libavformat/version.h +++ b/libavformat/version.h @@ -32,7 +32,7 @@ // Major bumping may affect Ticket5467, 5421, 5451(compatibility with Chromium) // Also please add any ticket numbers that you believe might be affected here #define LIBAVFORMAT_VERSION_MAJOR 58 -#define LIBAVFORMAT_VERSION_MINOR 31 +#define LIBAVFORMAT_VERSION_MINOR 32 #define LIBAVFORMAT_VERSION_MICRO 104 #define LIBAVFORMAT_VERSION_INT AV_VERSION_INT(LIBAVFORMAT_VERSION_MAJOR, \