From patchwork Tue Apr 17 01:52:31 2018 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Stephan Holljes X-Patchwork-Id: 8472 Delivered-To: ffmpegpatchwork@gmail.com Received: by 10.2.1.85 with SMTP id c82csp1440816jad; Mon, 16 Apr 2018 18:59:31 -0700 (PDT) X-Google-Smtp-Source: AIpwx4+l/lzZyfBC+CbN6jrVf2pdqXixhZGyPnMDbLkX0nF5eWSMqug7CQw0Z5b2cz0IqZ2QLUDa X-Received: by 10.223.209.193 with SMTP id m1mr96701wri.61.1523930371662; Mon, 16 Apr 2018 18:59:31 -0700 (PDT) ARC-Seal: i=1; a=rsa-sha256; t=1523930371; cv=none; d=google.com; s=arc-20160816; b=v0VvV4sASZdWjxCCp5JRHvXGUms8URTO69xdQDSpf/sjqEQAu2YGSiUtgNBhzuCInD w1Sayl8oDj1kYY55mYk37KXtR1t0n+/d8+Cdc4egJtZOc6BZJEL9Xk0XHv+yu9KR8r22 N9RPsXbONbVrZ6eMJyA9R1Z7ty5tE60WqAMep0AK3hV8CvSqr+wqnn+hIfR3/uhBJagW P+lL9Yia7yEMKNS9lHN+3PWK+YV1brkHaGvgkzduuW/PSZCqH0y920MXk/tLNryLcORm 8e0ddg65pqxyzG/+ol3Nl+sdSrLUNg8AdLoPdmaLld1whXXzsytwtcAQW7DXf5vZQ2ID UqjA== ARC-Message-Signature: i=1; a=rsa-sha256; c=relaxed/relaxed; d=google.com; s=arc-20160816; h=sender:errors-to:content-transfer-encoding:mime-version:cc:reply-to :list-subscribe:list-help:list-post:list-archive:list-unsubscribe :list-id:precedence:subject:references:in-reply-to:message-id:date :to:from:dkim-signature:delivered-to:arc-authentication-results; bh=fng5WDKFTPYv7FRu2OUhufoZLYlJpqXxoyztQmGt8SE=; b=oK0guiuwKndaHh2utVk7IDyvi3oL0atXE+z6sjNv7X22uOM9kbLGZ2ztmTRjLoaarn XVaeQH9fcs0c66/8DaqwBXOvc2jLes0BH7oV4FD4ISCrZTpmxqwCm3m93rY+jOPfeEEa TKnsaduJXxGQcwQRmgoFgbd6CW9XFHeRB4hh2QmosuyehGT0EtAYBSw3GZnktBb52g9R I8pJrHhgW3a0splNLOlzLuOpP3TjI1YRY2kLT+R5NajG46mqnKVvy4nHvIYxJFtbIA/Q RBpp+hZFJUsE+fqydApRtc8XpSMhWf2ryTPkRI2vVp26EMMycl7bRFTB71656wJNwVoj fglA== ARC-Authentication-Results: i=1; mx.google.com; dkim=neutral (body hash did not verify) header.i=@googlemail.com header.s=20161025 header.b=OwVw9SvW; spf=pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) smtp.mailfrom=ffmpeg-devel-bounces@ffmpeg.org; dmarc=fail (p=QUARANTINE sp=QUARANTINE dis=NONE) header.from=googlemail.com Return-Path: Received: from ffbox0-bg.mplayerhq.hu (ffbox0-bg.ffmpeg.org. [79.124.17.100]) by mx.google.com with ESMTP id 137si7111281wmm.225.2018.04.16.18.59.31; Mon, 16 Apr 2018 18:59:31 -0700 (PDT) Received-SPF: pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) client-ip=79.124.17.100; Authentication-Results: mx.google.com; dkim=neutral (body hash did not verify) header.i=@googlemail.com header.s=20161025 header.b=OwVw9SvW; spf=pass (google.com: domain of ffmpeg-devel-bounces@ffmpeg.org designates 79.124.17.100 as permitted sender) smtp.mailfrom=ffmpeg-devel-bounces@ffmpeg.org; dmarc=fail (p=QUARANTINE sp=QUARANTINE dis=NONE) header.from=googlemail.com Received: from [127.0.1.1] (localhost [127.0.0.1]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTP id C79FE689EB5; Tue, 17 Apr 2018 04:59:03 +0300 (EEST) X-Original-To: ffmpeg-devel@ffmpeg.org Delivered-To: ffmpeg-devel@ffmpeg.org Received: from mail-wr0-f194.google.com (mail-wr0-f194.google.com [209.85.128.194]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTPS id 7AE2D689A92 for ; Tue, 17 Apr 2018 04:58:57 +0300 (EEST) Received: by mail-wr0-f194.google.com with SMTP id v60so27096672wrc.7 for ; Mon, 16 Apr 2018 18:59:24 -0700 (PDT) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=googlemail.com; s=20161025; h=from:to:cc:subject:date:message-id:in-reply-to:references; bh=FQqMZeLKjOubwsCcyKU9k2Vw3s9Fh9G5KhI0eM9IUAQ=; b=OwVw9SvWkG9YUC2z9dMyg6gdJq/ksgNzvbkN0z5zUtOQOso4753CpaMWN/sZylkVqd jyQHlLS9v8Jyc6Pq8s5Xl6bOcpRZ82mPB1iTtRqKueVkYEs9uF7O6jw8xzm9jh2mpWun 95pKgn8ayR7u7OucTfS1u+sMp03zwYP2GL/9F/+RDz3/ovsoD3ma2cshf2B9O9NEJL3g h1Iq1836KrAMqGSWf9V/XqR0Sgsos9pIcLkH1wzkkQBSw/FTqqMUVPdinu3h0Qs+NqDF zr6g6MP+D1yuCkUOIiJovq7RP0+7J6QRe/sRkEeYCZEAEDoCFcUgavyzH9ro21WaLhEo xrCw== X-Google-DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed; d=1e100.net; s=20161025; h=x-gm-message-state:from:to:cc:subject:date:message-id:in-reply-to :references; bh=FQqMZeLKjOubwsCcyKU9k2Vw3s9Fh9G5KhI0eM9IUAQ=; b=JhqzIAz7yKQsi6hxjOlUIAC+H2vIXDnoFeGtabjVjj3d2AMN3OseHl8YdIsQ0FGE9u hXfn5FhLI40s/BZH1IDpneZieDLtJUXbfx8UbCEyzxKNa1w7q43KjBupnmkJmIkV+btf PQS4nA+qYiXULOtkyyR79pBh6qDq6FN2GdlzcN9+I3aCyQnJcWEouQ1iWFxMcsfEpR0W 3mWzXSHGIgeNIv/eRHkaL9xgKsTvPrLgGYuKntF8OeXhyjt59jKiwYJUcaz5sxvDh4E7 SsE2IHWPPS9petY/8mDt/Yse7UuF5EYwyOgXcRsLWyzadoQNgiST9RLxWPwzxcQO3Ek9 p+8w== X-Gm-Message-State: ALQs6tBoQgv4ZuWjtTFNmGz7aDXHTcqOLQXMXhvx1R599+BCWxYC8JOh KDEDjte1DTC46GTHwOFcqBTxPg== X-Received: by 10.28.69.93 with SMTP id s90mr327087wma.71.1523929978122; Mon, 16 Apr 2018 18:52:58 -0700 (PDT) Received: from localhost.localdomain (HSI-KBW-046-005-002-245.hsi8.kabel-badenwuerttemberg.de. [46.5.2.245]) by smtp.gmail.com with ESMTPSA id b47sm11845123wrg.13.2018.04.16.18.52.57 (version=TLS1_2 cipher=ECDHE-RSA-AES128-GCM-SHA256 bits=128/128); Mon, 16 Apr 2018 18:52:57 -0700 (PDT) From: Stephan Holljes To: ffmpeg-devel@ffmpeg.org Date: Tue, 17 Apr 2018 03:52:31 +0200 Message-Id: <20180417015233.31983-3-klaxa1337@googlemail.com> X-Mailer: git-send-email 2.16.2 In-Reply-To: <20180417015233.31983-1-klaxa1337@googlemail.com> References: <20180412133549.19939-1-klaxa1337@googlemail.com> <20180417015233.31983-1-klaxa1337@googlemail.com> Subject: [FFmpeg-devel] [PATCH 2/4] ffserver: Implement publisher 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 Cc: Stephan Holljes MIME-Version: 1.0 Errors-To: ffmpeg-devel-bounces@ffmpeg.org Sender: "ffmpeg-devel" --- publisher.c | 301 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ publisher.h | 156 +++++++++++++++++++++++++++++++ 2 files changed, 457 insertions(+) create mode 100644 publisher.c create mode 100644 publisher.h diff --git a/publisher.c b/publisher.c new file mode 100644 index 0000000..65dca70 --- /dev/null +++ b/publisher.c @@ -0,0 +1,301 @@ +/* + * 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 "publisher.h" +#include "segment.h" +#include + +void client_log(struct Client *c) +{ + av_log(NULL, AV_LOG_INFO, "State: "); + switch(c->state) { + case FREE: + av_log(NULL, AV_LOG_INFO, "FREE\n"); + break; + case RESERVED: + av_log(NULL, AV_LOG_INFO, "RESERVED\n"); + break; + case WAIT: + av_log(NULL, AV_LOG_INFO, "WAIT\n"); + break; + case WRITABLE: + av_log(NULL, AV_LOG_INFO, "WRITABLE\n"); + break; + case BUSY: + av_log(NULL, AV_LOG_INFO, "BUSY\n"); + break; + case BUFFER_FULL: + av_log(NULL, AV_LOG_INFO, "BUFFER_FULL\n"); + break; + default: + av_log(NULL, AV_LOG_INFO, "UNKOWN\n"); + break; + } +} + +void client_disconnect(struct Client *c, int write_trailer) +{ + struct Segment *seg; + if (write_trailer) + av_write_trailer(c->ofmt_ctx); + avio_close(c->ofmt_ctx->pb); + avformat_free_context(c->ofmt_ctx); + pthread_mutex_lock(&c->buffer_lock); + while(av_fifo_size(c->buffer)) { + av_fifo_generic_read(c->buffer, &seg, sizeof(struct Segment*), NULL); + segment_unref(seg); + } + pthread_mutex_unlock(&c->buffer_lock); + c->ofmt_ctx = NULL; + client_set_state(c, FREE); + c->current_segment_id = -1; +} + +void client_set_state(struct Client *c, enum State state) +{ + pthread_mutex_lock(&c->state_lock); + c->state = state; + pthread_mutex_unlock(&c->state_lock); +} + +void client_push_segment(struct Client *c, struct Segment *seg) +{ + pthread_mutex_lock(&c->buffer_lock); + if (av_fifo_space(c->buffer) == 0) { + av_log(NULL, AV_LOG_WARNING, "Client buffer full, dropping Segment.\n"); + client_set_state(c, BUFFER_FULL); + pthread_mutex_unlock(&c->buffer_lock); + return; + } + segment_ref(seg); + av_fifo_generic_write(c->buffer, &seg, sizeof(struct Segment*), NULL); + pthread_mutex_unlock(&c->buffer_lock); + client_set_state(c, WRITABLE); +} + +void publisher_init(struct PublisherContext **pub) +{ + int i; + struct PublisherContext *pc = (struct PublisherContext*) av_malloc(sizeof(struct PublisherContext)); + pc->nb_threads = 4; + pc->current_segment_id = -1; + pc->shutdown = 0; + pc->buffer = av_fifo_alloc_array(sizeof(struct Segment), MAX_SEGMENTS); + pc->fs_buffer = av_fifo_alloc_array(sizeof(struct Segment), MAX_SEGMENTS); + pthread_mutex_init(&pc->buffer_lock, NULL); + pthread_mutex_init(&pc->fs_buffer_lock, NULL); + for (i = 0; i < MAX_CLIENTS; i++) { + struct Client *c = &pc->clients[i]; + c->buffer = av_fifo_alloc_array(sizeof(struct Segment), MAX_SEGMENTS); + c->id = i; + c->current_segment_id = -1; + pthread_mutex_init(&c->state_lock, NULL); + pthread_mutex_init(&c->buffer_lock, NULL); + client_set_state(c, FREE); + } + *pub = pc; +} + +void publisher_push_segment(struct PublisherContext *pub, struct Segment *seg) +{ + struct Segment *drop; + pthread_mutex_lock(&pub->buffer_lock); + pthread_mutex_lock(&pub->fs_buffer_lock); + av_fifo_generic_write(pub->buffer, &seg, sizeof(struct Segment*), NULL); + segment_ref(seg); + if (av_fifo_size(pub->fs_buffer) >= BUFFER_SEGMENTS * sizeof(struct Segment*)) { + av_fifo_generic_read(pub->fs_buffer, &drop, sizeof(struct Segment*), NULL); + segment_unref(drop); + } + av_fifo_generic_write(pub->fs_buffer, &seg, sizeof(struct Segment*), NULL); + pthread_mutex_unlock(&pub->buffer_lock); + pthread_mutex_unlock(&pub->fs_buffer_lock); + segment_ref(seg); +} + +int publisher_reserve_client(struct PublisherContext *pub) +{ + int i; + for (i = 0; i < MAX_CLIENTS; i++) { + switch(pub->clients[i].state) { + case FREE: + client_set_state(&pub->clients[i], RESERVED); + return 0; + default: + continue; + } + } + return 1; +} + +void publisher_cancel_reserve(struct PublisherContext *pub) +{ + int i; + for (i = 0; i < MAX_CLIENTS; i++) { + switch(pub->clients[i].state) { + case RESERVED: + client_set_state(&pub->clients[i], FREE); + return; + default: + continue; + } + } + return; +} + +void client_push_prebuffer(struct PublisherContext *pub, struct Client *c) +{ + int off; + int size; + struct Segment *seg; + pthread_mutex_lock(&pub->fs_buffer_lock); + size = av_fifo_size(pub->fs_buffer); + for (off = 0; off < size; off += sizeof(struct Segment*)) { + av_fifo_generic_peek_at(pub->fs_buffer, &seg, off, sizeof(struct Segment*), NULL); + client_push_segment(c, seg); + } + pthread_mutex_unlock(&pub->fs_buffer_lock); +} + +void publisher_add_client(struct PublisherContext *pub, AVFormatContext *ofmt_ctx) +{ + int i; + for (i = 0; i < MAX_CLIENTS; i++) { + switch(pub->clients[i].state) { + case RESERVED: + pub->clients[i].ofmt_ctx = ofmt_ctx; + client_set_state(&pub->clients[i], WRITABLE); + client_push_prebuffer(pub, &pub->clients[i]); + return; + default: + continue; + } + } +} + +void publisher_free(struct PublisherContext *pub) +{ + int i; + struct Segment *seg; + pthread_mutex_lock(&pub->buffer_lock); + while(av_fifo_size(pub->buffer)) { + av_fifo_generic_read(pub->buffer, &seg, sizeof(struct Segment*), NULL); + segment_unref(seg); + } + av_fifo_freep(&pub->buffer); + pthread_mutex_unlock(&pub->buffer_lock); + + pthread_mutex_lock(&pub->fs_buffer_lock); + while(av_fifo_size(pub->fs_buffer)) { + av_fifo_generic_read(pub->fs_buffer, &seg, sizeof(struct Segment*), NULL); + segment_unref(seg); + } + av_fifo_freep(&pub->fs_buffer); + for (i = 0; i < MAX_CLIENTS; i++) { + av_fifo_freep(&pub->clients[i].buffer); + } + pthread_mutex_unlock(&pub->fs_buffer_lock); + av_free(pub); + return; +} + +void publisher_freep(struct PublisherContext **pub) +{ + publisher_free(*pub); + *pub = NULL; + return; +} + +void publish(struct PublisherContext *pub) +{ + int i; + struct Segment *seg; + char filename[128] = {0}; + pthread_mutex_lock(&pub->buffer_lock); + av_log(NULL, AV_LOG_DEBUG, "pub->buffer size: %d\n", av_fifo_size(pub->buffer)); + if (av_fifo_size(pub->buffer) == 0) { + pthread_mutex_unlock(&pub->buffer_lock); + return; + } + av_fifo_generic_read(pub->buffer, &seg, sizeof(struct Segment*), NULL); + pthread_mutex_unlock(&pub->buffer_lock); + if (seg) { + pub->current_segment_id = seg->id; + snprintf(filename, 127, "segment-%04d.mkv", seg->id); + segment_save(seg, filename); + client_log(pub->clients); + + for (i = 0; i < MAX_CLIENTS; i++) { + switch(pub->clients[i].state) { + case BUFFER_FULL: + av_log(pub, AV_LOG_WARNING, "Dropping segment for client %d, buffer full.\n", i); + continue; + case WAIT: + case WRITABLE: + client_push_segment(&pub->clients[i], seg); + default: + continue; + } + } + segment_unref(seg); + } +} + +void publisher_gen_status_json(struct PublisherContext *pub, char *status) +{ + int states[STATE_NB] = {0}; + int current_read = 0, newest_write = 0, oldest_write = 0; + int i; + struct Client *c; + + current_read = pub->current_segment_id; + oldest_write = current_read; + + for (i = 0; i < MAX_CLIENTS; i++) { + c = &pub->clients[i]; + if (c->current_segment_id > 0 && c->current_segment_id < oldest_write) { + oldest_write = c->current_segment_id; + } + if (c->current_segment_id > newest_write) { + newest_write = c->current_segment_id; + } + states[c->state]++; + } + + + snprintf(status, 4095, + "{\n\t\"free\": %d,\n" + "\t\"reserved\": %d,\n" + "\t\"wait\": %d,\n" + "\t\"writable\": %d,\n" + "\t\"busy\": %d,\n" + "\t\"buffer_full\": %d,\n" + "\t\"current_read\": %d,\n" + "\t\"newest_write\": %d,\n" + "\t\"oldest_write\": %d\n" + "}\n", + states[FREE], + states[RESERVED], + states[WAIT], + states[WRITABLE], + states[BUSY], + states[BUFFER_FULL], + current_read, + newest_write, + oldest_write); +} diff --git a/publisher.h b/publisher.h new file mode 100644 index 0000000..2508837 --- /dev/null +++ b/publisher.h @@ -0,0 +1,156 @@ +/* + * 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 + */ + +#ifndef PUBLISHER_H +#define PUBLISHER_H + +#include +#include +#include +#include "segment.h" + +#define MAX_CLIENTS 16 +#define MAX_SEGMENTS 16 +#define BUFFER_SEGMENTS 10 + +/* Client State enum */ + +enum State { + FREE, // no client connected + RESERVED, // reserved for a client that just connected + WAIT, // up to date, no new Segments to write + WRITABLE, // buffer is not full, new Segments can be pushed + BUSY, // currently writing to this client + BUFFER_FULL, // client buffer is full, new Segments will be dropped + STATE_NB +}; + + +struct Client { + AVFormatContext *ofmt_ctx; // writable AVFormatContext, basically our tcp connection to the client + AVFifoBuffer *buffer; // Client buffer of Segment references + enum State state; + pthread_mutex_t buffer_lock; + pthread_mutex_t state_lock; + int id; + int current_segment_id; // The stream-based id of the segment that has last been worked on. +}; + +struct PublisherContext { + struct Client clients[MAX_CLIENTS]; // currently compile-time configuration, easly made dynamic with malloc? + AVFifoBuffer *buffer; // publisher buffer for new Segments + AVFifoBuffer *fs_buffer; // fast start buffer + pthread_mutex_t buffer_lock; + pthread_mutex_t fs_buffer_lock; + int nb_threads; + int current_segment_id; + int shutdown; // indicate shutdown, gracefully close client connections and files and exit +}; + +/** + * Log a client's stats to the console. + * + * @param c pointer to the client to print + */ +void client_log(struct Client *c); + +/** + * Disconnect a client. + * + * @param c pointer to the client to disconnect. + */ +void client_disconnect(struct Client *c, int write_trailer); + +/** + * Set a client's state. Note: This is protected by mutex locks. + * + * @param c pointer to the client to set the state of + * @param state the state to set the client to + */ +void client_set_state(struct Client *c, enum State state); + +/** + * Allocate and initialize a PublisherContext + * + * @param pub pointer to a pointer to a PublisherContext. It will be allocated and initialized. + */ +void publisher_init(struct PublisherContext **pub); + +/** + * Push a Segment to a PublisherContext. + * + * @param pub pointer to a PublisherContext + * @param seg pointer to the Segment to add + */ +void publisher_push_segment(struct PublisherContext *pub, struct Segment *seg); + +/** + * Reserve a slot in the client struct of a PublisherContext. May fail if the number + * of maximum clients has been reached. + * + * @param pub pointer to a PublisherContext + * @return 0 in case of success, 1 in case of failure + */ +int publisher_reserve_client(struct PublisherContext *pub); + +/** + * Cancel a single reservation. This can be used if a client spot was reserved, but the client + * unexpectedly disconnects or sends an invalid request. + * + * @param pub pointer to a PublisherContext + */ +void publisher_cancel_reserve(struct PublisherContext *pub); + +/** + * Add a client by its ofmt_ctx. This initializes an element in the client struct of the PublisherContext + * that has been reserved prior to calling this function. + * + * @param pub pointer to a PublisherContext + * @param ofmt_ctx AVFormatContext of a client + */ +void publisher_add_client(struct PublisherContext *pub, AVFormatContext *ofmt_ctx); + +/** + * Free buffers and associated client buffers. + * + * @param pub pointer to the PublisherContext to free + */ +void publisher_free(struct PublisherContext *pub); + +/** + * Free buffers and associated client buffers and set *pub to NULL. + * + * @param pub pointer to the PublisherContext pointer to free + */ +void publisher_freep(struct PublisherContext **pub); + +/** + * Signal to the PublisherContext to check its buffer and publish pending Segments. + * + * @param pub pointer to a PublisherContext + */ +void publish(struct PublisherContext *pub); + +/** + * Print the current client and file reading status to a json string. + * @param pub pointer to a PublisherContext + * @param status string of at least 4096 bytes size. + */ +void publisher_gen_status_json(struct PublisherContext *pub, char *status); + +#endif // PUBLISHER_H