Skip to content
This repository


Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP

Streaming queue size protection #171

wants to merge 1 commit into from

2 participants

Jernej Fijačko Adam Sutton
Jernej Fijačko

Now that all memory leaks in http streaming are cleared there is only one major memory issue left. HTSP streaming has a queue size protection but HTTP streaming doesn't.

When there are clients on a slow connection or if the server has connection problems the streaming queue size can increase indefinitely and sooner or later cause out of memory problems on the server.

This can be very easily simulated with VLC. Just start a HTTP stream and press play/pause in 5-10 seconds intervals. You will see the memory usage of tvheadend increase at the bit rate of the stream.

Here is a massif heap profile of tvheadend (marked is the highest memory usage --> cca. 140Mb):
Tvheadend without queue size protection

Here is a massif heap profile of tvheadend with the queue size protection from these PR (marked is the highest memory usage --> cca. 9Mb):
Tvheadend with queue size protection

Both heap profiles were recorded with the same stream, same player and similar play/pause intervals.

I'm not sure if this is the best solution for this problem. I just wanted to point out that there is a problem.

Adam Sutton

@mikrohard great I'll take a look, I believe at the time I made the previous "fix" I did state that the exact scenario you describe was still an issue that ultimately needed to be solved.

Looks like you might have done the job for me :)

Adam Sutton

@mikrohard OK, I've had a look and I'm going to make some changes. The queue size checking will be optional and configurable by the caller. This will remove the requirement to check size where its not necessary, i.e.:

DVR - since it write to disk should be able to keep up
Timeshift - same as DVR (not in master of course)
HTSP - has its own q limiting
Intermediate queues - anything else that's in the internal streaming chain.

So this currently just leaves the HTTP streaming for which a queue size is specified but can also be overridden by the user (using HTTP GET arg).

Ultimately we should consider (for PKT based streaming) whether we should try and employ a similar methodology to HTSP (frame type dependant q lengths). But that's for another day.

I'll push this all to master shortly.


Jernej Fijačko

Sure... but keep in mind that DVR (and timeshift) could be also written on a USB drive or sd card on a low power (and memory) device. There should always be a protection (even if it's set to a higher value than streaming).

Adam Sutton

@mikrohard it's a fair point, but I'd argue if you're silly enough to try and record/shift to such slow media then you're asking for trouble anyway ;) but we can easily integrate limits later.

However since the only issues reported thus far relate to HTTP streaming that seems the most important place to cover things. And we definitely don't want to add extra queue size checking where its not needed (such as intermediate queues, that will ultimately be bounded by the final Q) and for streaming chains that already have their own limiting (HTSP).

At least with DVR and timeshift they will (mostly) be bounded by the fact there will be intrinsic time limits (though admittedly still more than enough data to kill a system).

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Showing 1 unique commit by 1 author.

Oct 23, 2012
Jernej Fijačko mikrohard Streaming queue size protection 0254545
This page is out of date. Refresh to see the latest.

Showing 2 changed files with 40 additions and 1 deletion. Show diff stats Hide diff stats

  1. +38 1 src/streaming.c
  2. +2 0  src/streaming.h
39 src/streaming.c
@@ -52,7 +52,14 @@ streaming_queue_deliver(void *opauqe, streaming_message_t *sm)
52 52 streaming_queue_t *sq = opauqe;
53 53
54 54 pthread_mutex_lock(&sq->sq_mutex);
55   - TAILQ_INSERT_TAIL(&sq->sq_queue, sm, sm_link);
  55 +
  56 + /* queue size protection */
  57 + int queue_size = streaming_queue_size(&sq->sq_queue);
  58 + if (queue_size > 1500000)
  59 + streaming_msg_free(sm);
  60 + else
  61 + TAILQ_INSERT_TAIL(&sq->sq_queue, sm, sm_link);
  62 +
56 63 pthread_cond_signal(&sq->sq_cond);
57 64 pthread_mutex_unlock(&sq->sq_mutex);
58 65 }
@@ -334,6 +341,36 @@ streaming_queue_clear(struct streaming_message_queue *q)
334 341 /**
335 342 *
336 343 */
  344 +int streaming_queue_size(struct streaming_message_queue *q)
  345 +{
  346 + streaming_message_t *sm;
  347 + int size = 0;
  348 +
  349 + TAILQ_FOREACH(sm, q, sm_link) {
  350 + if (sm->sm_type == SMT_PACKET)
  351 + {
  352 + th_pkt_t *pkt = sm->sm_data;
  353 + if (pkt && pkt->pkt_payload)
  354 + {
  355 + size += pkt->pkt_payload->pb_size;
  356 + }
  357 + }
  358 + else if (sm->sm_type == SMT_MPEGTS)
  359 + {
  360 + pktbuf_t *pkt_payload = sm->sm_data;
  361 + if (pkt_payload)
  362 + {
  363 + size += pkt_payload->pb_size;
  364 + }
  365 + }
  366 + }
  367 + return size;
  368 +}
  369 +
  370 +
  371 +/**
  372 + *
  373 + */
337 374 const char *
338 375 streaming_code2txt(int code)
339 376 {
2  src/streaming.h
@@ -73,6 +73,8 @@ void streaming_queue_init(streaming_queue_t *sq, int reject_filter);
73 73
74 74 void streaming_queue_clear(struct streaming_message_queue *q);
75 75
  76 +int streaming_queue_size(struct streaming_message_queue *q);
  77 +
76 78 void streaming_queue_deinit(streaming_queue_t *sq);
77 79
78 80 void streaming_target_connect(streaming_pad_t *sp, streaming_target_t *st);

Tip: You can add notes to lines in a file. Hover to the left of a line to make a note

Something went wrong with that request. Please try again.