1
0
mirror of https://github.com/FFmpeg/FFmpeg.git synced 2024-12-23 12:43:46 +02:00

udp: switch custom fifo to libavutil fifo.

This also allows us to drop the semaphore as our fifo is lock free.
Signed-off-by: Michael Niedermayer <michaelni@gmx.at>
This commit is contained in:
Michael Niedermayer 2011-05-13 17:16:15 +02:00
parent 593bb98d61
commit 03f8f39769

View File

@ -29,13 +29,13 @@
#include "avformat.h" #include "avformat.h"
#include "avio_internal.h" #include "avio_internal.h"
#include "libavutil/parseutils.h" #include "libavutil/parseutils.h"
#include "libavutil/fifo.h"
#include <unistd.h> #include <unistd.h>
#include "internal.h" #include "internal.h"
#include "network.h" #include "network.h"
#include "os_support.h" #include "os_support.h"
#include "url.h" #include "url.h"
#include <pthread.h> #include <pthread.h>
#include <semaphore.h>
#include <sys/resource.h> #include <sys/resource.h>
#include <sys/time.h> #include <sys/time.h>
@ -56,15 +56,11 @@ typedef struct {
int is_connected; int is_connected;
/* Circular Buffer variables for use in UDP receive code */ /* Circular Buffer variables for use in UDP receive code */
unsigned char *circular_buffer;
int circular_buffer_head;
int circular_buffer_size; int circular_buffer_size;
int circular_buffer_tail; AVFifoBuffer *fifo;
int circular_buffer_available;
int circular_buffer_available_max; int circular_buffer_available_max;
int circular_buffer_error; int circular_buffer_error;
pthread_t circular_buffer_thread; pthread_t circular_buffer_thread;
sem_t circular_buffer_semaphore;
} UDPContext; } UDPContext;
#define UDP_TX_BUF_SIZE 32768 #define UDP_TX_BUF_SIZE 32768
@ -345,31 +341,28 @@ static void *circular_buffer_task( void *_URLContext)
continue; continue;
/* How much do we have left to the end of the buffer */ /* How much do we have left to the end of the buffer */
left = s->circular_buffer_size-s->circular_buffer_head;
/* Whats the minimum we can read so that we dont comletely fill the buffer */ /* Whats the minimum we can read so that we dont comletely fill the buffer */
sem_wait( &s->circular_buffer_semaphore); left = av_fifo_space(s->fifo);
left = FFMIN( left, s->circular_buffer_size-s->circular_buffer_available); left = FFMIN(left, s->fifo->end - s->fifo->wptr);
sem_post( &s->circular_buffer_semaphore );
/* No Space left, error, what do we do now */ /* No Space left, error, what do we do now */
if( !left) { if( !left) {
av_log(h, AV_LOG_ERROR, "circular_buffer: OVERRUN\n"); av_log(h, AV_LOG_ERROR, "circular_buffer: OVERRUN\n");
s->circular_buffer_error = EIO; s->circular_buffer_error = EIO;
return NULL; return NULL;
} }
len = recv(s->udp_fd, s->circular_buffer+s->circular_buffer_head, left, 0);
len = recv(s->udp_fd, s->fifo->wptr, left, 0);
if (len < 0) { if (len < 0) {
if (ff_neterrno() != AVERROR(EAGAIN) && ff_neterrno() != AVERROR(EINTR)) { if (ff_neterrno() != AVERROR(EAGAIN) && ff_neterrno() != AVERROR(EINTR)) {
s->circular_buffer_error = EIO; s->circular_buffer_error = EIO;
return NULL; return NULL;
} }
} }
s->circular_buffer_head += len; s->fifo->wptr += len;
sem_wait( &s->circular_buffer_semaphore); if (s->fifo->wptr >= s->fifo->end)
s->circular_buffer_available += len; s->fifo->wptr = s->fifo->buffer;
s->circular_buffer_available_max = FFMAX( s->circular_buffer_available_max, s->circular_buffer_available); s->fifo->wndx += len;
sem_post( &s->circular_buffer_semaphore );
if( s->circular_buffer_head>=s->circular_buffer_size)
s->circular_buffer_head -= s->circular_buffer_size;
} }
return NULL; return NULL;
@ -516,11 +509,7 @@ static int udp_open(URLContext *h, const char *uri, int flags)
if (!is_output && s->circular_buffer_size) { if (!is_output && s->circular_buffer_size) {
/* start the task going */ /* start the task going */
s->circular_buffer = av_malloc( s->circular_buffer_size); s->fifo = av_fifo_alloc(s->circular_buffer_size);
if (sem_init( &s->circular_buffer_semaphore, PTHREAD_PROCESS_PRIVATE, 1 )) {
av_log(h, AV_LOG_ERROR, "sem_init failed\n");
goto fail;
}
if (pthread_create(&s->circular_buffer_thread, NULL, circular_buffer_task, h)) { if (pthread_create(&s->circular_buffer_thread, NULL, circular_buffer_task, h)) {
av_log(h, AV_LOG_ERROR, "pthread_create failed\n"); av_log(h, AV_LOG_ERROR, "pthread_create failed\n");
goto fail; goto fail;
@ -531,10 +520,7 @@ static int udp_open(URLContext *h, const char *uri, int flags)
fail: fail:
if (udp_fd >= 0) if (udp_fd >= 0)
closesocket(udp_fd); closesocket(udp_fd);
if (s->circular_buffer) { av_fifo_free(s->fifo);
sem_destroy( &s->circular_buffer_semaphore);
av_free( s->circular_buffer);
}
av_free(s); av_free(s);
return AVERROR(EIO); return AVERROR(EIO);
} }
@ -551,30 +537,12 @@ static int udp_read(URLContext *h, uint8_t *buf, int size)
if (s->circular_buffer_thread) { if (s->circular_buffer_thread) {
do { do {
sem_wait( &s->circular_buffer_semaphore ); avail = av_fifo_size(s->fifo);
avail = s->circular_buffer_available;
sem_post( &s->circular_buffer_semaphore );
if (avail) { // >=size) { if (avail) { // >=size) {
// Maximum amount available // Maximum amount available
size = FFMIN( avail, size); size = FFMIN( avail, size);
// Whats left till the end of the circular buffer av_fifo_generic_read(s->fifo, buf, size, NULL);
left = s->circular_buffer_size-s->circular_buffer_tail;
// How much do we need, all?
left = FFMIN( left, size);
// Get the first block
memcpy( buf, s->circular_buffer+s->circular_buffer_tail, left);
// Have we any more, this will be from the start of the buffer
if (size-left)
memcpy( buf+left, s->circular_buffer, size-left);
// Check for the tail wrapping around
s->circular_buffer_tail += size;
if( s->circular_buffer_tail>=s->circular_buffer_size)
s->circular_buffer_tail -= s->circular_buffer_size;
// Update the available amount
sem_wait( &s->circular_buffer_semaphore );
s->circular_buffer_available -= size;
sem_post( &s->circular_buffer_semaphore );
return size; return size;
} }
else { else {
@ -627,11 +595,8 @@ static int udp_close(URLContext *h)
if (s->is_multicast && (h->flags & AVIO_FLAG_READ)) if (s->is_multicast && (h->flags & AVIO_FLAG_READ))
udp_leave_multicast_group(s->udp_fd, (struct sockaddr *)&s->dest_addr); udp_leave_multicast_group(s->udp_fd, (struct sockaddr *)&s->dest_addr);
closesocket(s->udp_fd); closesocket(s->udp_fd);
if (s->circular_buffer) { av_log( h, AV_LOG_INFO, "circular_buffer_info max:%d%%\r\n", (s->circular_buffer_available_max*100)/s->circular_buffer_size);
sem_destroy( &s->circular_buffer_semaphore); av_fifo_free(s->fifo);
av_free( s->circular_buffer);
av_log( h, AV_LOG_INFO, "circular_buffer_info max:%d%%\r\n", (s->circular_buffer_available_max*100)/s->circular_buffer_size);
}
av_free(s); av_free(s);
return 0; return 0;
} }