From patchwork Thu Apr 12 13:35:47 2018 Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit X-Patchwork-Submitter: Stephan Holljes X-Patchwork-Id: 8424 Delivered-To: ffmpegpatchwork@gmail.com Received: by 10.2.1.70 with SMTP id c67csp1910833jad; Thu, 12 Apr 2018 06:44:04 -0700 (PDT) X-Google-Smtp-Source: AIpwx48zB5ylYvjewQHVO0Wc4JJQe9kz3FDusKBDeXXuGIByQlKYpPaZDobi1NT/hRozFp4/Qzwt X-Received: by 10.28.91.203 with SMTP id p194mr857393wmb.52.1523540644523; Thu, 12 Apr 2018 06:44:04 -0700 (PDT) ARC-Seal: i=1; a=rsa-sha256; t=1523540644; cv=none; d=google.com; s=arc-20160816; b=EB/E/iAiNkkIWSoL23WorrbB7YeUH1mUqZLCwEK2v9X1pqGycrIftv6vhRkUN5rH9r auqlTkm4H+Tmn4+GWqeRwASXP5UqBG8czat5SG49KWXNrCbPLGhxaD59z7BPNMCmivjx zIxP661fX1VRCenQRwVDaIHBVFiZtv41EPU4UM3ehbsTTAwylAokRhecngLqjcCV/1rO Id0Rm2K92FPKJv34YjCmKmV1DBxSJF+InFCOF2Fah3wExBXBzc+DV+PpETESYPeISW6Q fHe/eak0vHPgB21m/HWWVmnZ9lGfqmluCLM70aJaKvHuyaw/Dr9JVjLPwp+UYRNB9Ym1 r+Uw== 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=ch9Zr6IuAkgjJLRUOBGlmscN+N77n1Lujqz6NvZ3PLY=; b=fje18IoXtZtL/+l/xxc1hPqoaoqIFsWY+lpDQ5kU9NqdJ2hu4gXKi8627xxs9gcq9l XFIFZ/fpg55UuQ/DsNtet+REt66G/y92lY4X78UgCKqN2vJ2n/KS0DTsGxp/TbTYI0Py Tt8yHfZHpes3OsJkPooaJ4CZipXnC+R2qbCKyxqukID4KSAOrhZ9tXInVCIWpTRcI8Qu vpZDQ/DfofUEd4VWloqKFGkcKrDM4kWby4L0HvC8qc0p8TNKjAwnE4uPv9OkDDPVzN+I 3z3hSv+nKGxUEF4thHkZ3nCX0GvrAcc6O3a8+J4xp+lzNl/NLtYsJ1vmMWhg1mqg7A/K jNwg== ARC-Authentication-Results: i=1; mx.google.com; dkim=neutral (body hash did not verify) header.i=@googlemail.com header.s=20161025 header.b=Yj3fqRoo; 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 v187si2577566wmf.147.2018.04.12.06.44.03; Thu, 12 Apr 2018 06:44:04 -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=Yj3fqRoo; 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 25583689AAF; Thu, 12 Apr 2018 16:43:38 +0300 (EEST) X-Original-To: ffmpeg-devel@ffmpeg.org Delivered-To: ffmpeg-devel@ffmpeg.org Received: from mail-wm0-f68.google.com (mail-wm0-f68.google.com [74.125.82.68]) by ffbox0-bg.mplayerhq.hu (Postfix) with ESMTPS id 20BDC68828D for ; Thu, 12 Apr 2018 16:43:32 +0300 (EEST) Received: by mail-wm0-f68.google.com with SMTP id b21so9757300wme.4 for ; Thu, 12 Apr 2018 06:43:56 -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=iGmylVNi/f2H4D8kL8IzjE65tWAcaX24qi8tHpLWcwQ=; b=Yj3fqRoo37IKdpQH8FRe228ICzEnKfSCcr6FB2WRKo/+KbnrHF5ngKTi5iPRyaqLv4 lyS0d76F5X7muumKh7Y6LPRPtugILug7RfLAwLAmDY2/nSTcK6ds7Vs08w110UVuL4Nm nz7o1gtVIxOwr7iYeAFo2NmfPVSO1usi47HDCsktePnpAn+HMV+NvxfKzVzoV2My/k6P Xfsj+LBdPYmfY+MgQkZ0MUPA6Nyd7YzsqQml0EUKgNQWqOfDwSFxzA21ZlJwRxqKXaip pTWW4tf6DuxZgohNK7F8mpnvBGxJ6Js+gfLXuiqlaFqih6rMqeVa0Helv8tK4tm1nipp LNrw== 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=iGmylVNi/f2H4D8kL8IzjE65tWAcaX24qi8tHpLWcwQ=; b=RYMuonEPTPiM7y85WY+4J2vdP3Ded+iQ25BnpyJZjbP3kUi6YXNeY1e198hDNZlGrE dnScGA6fXux1cgAIX7Qvi2FiSJmqf3nfT2WOU3M3K7w1qAd/410g9W85vRteAK7GZxFM im/RdSKCv+GgBAkkUFt2y1X1w1jmzcckUyt2GbXxF3t6jWhZiIi2+mGlgUlfAb2yRmTj Xj05t9yBzZP1v/UJNkirMFJrZO2TsKNNCv/6iFFeUzEVg8bSYKEdUZ5EtM541oMbXrN5 OzEvMpsfjTi0+w53V+jJ+kEhfIYMeGATlPFq8DF8/8kfEYDNIvyk5NLwWgROtt1ZzFvj H0xg== X-Gm-Message-State: ALQs6tCDltnBuHzFkmEAd+JQgeYo972edtGhjpUOshbnq82FP80zjKFF L+68/CIkgbQ+u67WKRLdQ/S6FA== X-Received: by 10.28.156.2 with SMTP id f2mr759598wme.64.1523540173967; Thu, 12 Apr 2018 06:36:13 -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 y101sm4588731wmh.10.2018.04.12.06.36.12 (version=TLS1_2 cipher=ECDHE-RSA-AES128-GCM-SHA256 bits=128/128); Thu, 12 Apr 2018 06:36:13 -0700 (PDT) From: Stephan Holljes To: ffmpeg-devel@ffmpeg.org Date: Thu, 12 Apr 2018 15:35:47 +0200 Message-Id: <20180412133549.19939-3-klaxa1337@googlemail.com> X-Mailer: git-send-email 2.16.2 In-Reply-To: <20180412133549.19939-1-klaxa1337@googlemail.com> References: <20180412133549.19939-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 | 278 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ publisher.h | 134 +++++++++++++++++++++++++++++ 2 files changed, 412 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..d1ccb95 --- /dev/null +++ b/publisher.c @@ -0,0 +1,278 @@ +#include "publisher.h" +#include "segment.h" +#include + +void client_log(struct Client *c) +{ + char state[64]; + sprintf("State: ", state); + switch(c->state) { + case FREE: + sprintf(state, "FREE"); + break; + case RESERVED: + sprintf(state, "RESERVED"); + break; + case WAIT: + sprintf(state, "WAIT"); + break; + case WRITABLE: + sprintf(state, "WRITABLE"); + break; + case BUSY: + sprintf(state, "BUSY"); + break; + case BUFFER_FULL: + sprintf(state, "BUFFER_FULL"); + break; + default: + sprintf(state, "UNDEFINED"); + break; + } + av_log(NULL, AV_LOG_INFO, "%s\n", state); +} + +void client_disconnect(struct Client *c) +{ + struct Segment *seg; + av_write_trailer(c->ofmt_ctx); + avio_close(c->ofmt_ctx->pb); + avformat_free_context(c->ofmt_ctx); + while(av_fifo_size(c->buffer)) { + av_fifo_generic_read(c->buffer, &seg, sizeof(struct Segment*), NULL); + segment_unref(seg); + } + 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) +{ + 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); + return; + } + segment_ref(seg); + av_fifo_generic_write(c->buffer, &seg, sizeof(struct Segment*), NULL); + client_set_state(c, WRITABLE); +} + +void publisher_init(struct PublisherContext **pub) +{ + int i; + struct PublisherContext *pc = (struct PublisherContext*) 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); + 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); + client_set_state(c, FREE); + } + *pub = pc; +} + +void publisher_push_segment(struct PublisherContext *pub, struct Segment *seg) +{ + struct Segment *drop; + 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); + 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; + 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); + } +} + +void publisher_add_client(struct PublisherContext *pub, AVFormatContext *ofmt_ctx) +{ + int i; + struct Segment *prebuffer_seg; + 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; + 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); + 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); + } + return; +} + +void publisher_freep(struct PublisherContext **pub) +{ + publisher_free(*pub); + *pub = NULL; + return; +} + +void publish(struct PublisherContext *pub) +{ + int i; + struct Segment *seg; + av_log(NULL, AV_LOG_DEBUG, "pub->buffer size: %d\n", av_fifo_size(pub->buffer)); + if (av_fifo_size(pub->buffer) == 0) + return; + av_log(NULL, AV_LOG_DEBUG, "Peeking buffer\n"); + av_fifo_generic_read(pub->buffer, &seg, sizeof(struct Segment*), NULL); + av_log(NULL, AV_LOG_DEBUG, "Peeked buffer\n"); + if (seg) { + pub->current_segment_id = seg->id; + for (i = 0; i < MAX_CLIENTS; i++) { + switch(pub->clients[i].state) { + case BUFFER_FULL: + av_log(NULL, 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 nb_free = 0, nb_reserved = 0, nb_wait = 0, nb_writable = 0, nb_busy = 0, nb_buffer_full = 0, 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; + } + + switch(c->state) { + case FREE: + nb_free++; + continue; + case RESERVED: + nb_reserved++; + continue; + case WAIT: + nb_wait++; + continue; + case WRITABLE: + nb_writable++; + continue; + case BUSY: + nb_busy++; + continue; + case BUFFER_FULL: + nb_buffer_full++; + continue; + default: + continue; + } + } + + + 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", + nb_free, + nb_reserved, + nb_wait, + nb_writable, + nb_busy, + nb_buffer_full, + current_read, + newest_write, + oldest_write); +} diff --git a/publisher.h b/publisher.h new file mode 100644 index 0000000..7646fda --- /dev/null +++ b/publisher.h @@ -0,0 +1,134 @@ +#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 +}; + + +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 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 + 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); + +/** + * 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