Commit 5019130d authored by Michael Smith's avatar Michael Smith

Max queue length for clients is now

    a) based on total bytes in queue, not total number of buffers in queue
    b) configurable (defaults to 100 kB)

mp3 metadata relaying (inline). Untested.

svn path=/trunk/icecast/; revision=4364
parent 8c14c0e0
...@@ -2,12 +2,12 @@ BUGS ...@@ -2,12 +2,12 @@ BUGS
---- ----
- stats get off? this needs testing more testing. - stats get off? this needs testing more testing.
- autoconf doesn't set HAVE_POLL (still true?)
- some stuff (like 'genre') isn't making it into the stats dump - some stuff (like 'genre') isn't making it into the stats dump
- make install - doesn't install configs? - make install - doesn't install configs?
- logging - bytes send and time listening may both be broken?
FEATURES FEATURES
-------- --------
...@@ -46,7 +46,10 @@ FEATURES ...@@ -46,7 +46,10 @@ FEATURES
- httpp - split out query string for further processing - httpp - split out query string for further processing
- binding to multiple ports (possibly including full ipv6 support) - binding to multiple ports
- option to use ipv6 (equiv to using <bind-address>::</bindaddress>, I think.
- per-mountpoint listener maximums.
...@@ -6,6 +6,7 @@ ...@@ -6,6 +6,7 @@
<clients>100</clients> <clients>100</clients>
<sources>2</sources> <sources>2</sources>
<threadpool>5</threadpool> <threadpool>5</threadpool>
<queue-size>102400</queue-size>
<client-timeout>30</client-timeout> <client-timeout>30</client-timeout>
<header-timeout>15</header-timeout> <header-timeout>15</header-timeout>
<source-timeout>10</source-timeout> <source-timeout>10</source-timeout>
...@@ -42,6 +43,9 @@ ...@@ -42,6 +43,9 @@
<server>127.0.0.1</server> <server>127.0.0.1</server>
<port>8001</port> <port>8001</port>
<mount>/example.ogg</mount> <mount>/example.ogg</mount>
<local-mount>/different.ogg</local-mount>
<relay-shoutcast-metadata>0</relay-shoutcast-metadata>
</relay> </relay>
--> -->
......
...@@ -13,6 +13,7 @@ ...@@ -13,6 +13,7 @@
#define CONFIG_DEFAULT_ADMIN "icemaster@localhost" #define CONFIG_DEFAULT_ADMIN "icemaster@localhost"
#define CONFIG_DEFAULT_CLIENT_LIMIT 256 #define CONFIG_DEFAULT_CLIENT_LIMIT 256
#define CONFIG_DEFAULT_SOURCE_LIMIT 16 #define CONFIG_DEFAULT_SOURCE_LIMIT 16
#define CONFIG_DEFAULT_QUEUE_SIZE_LIMIT (100*1024)
#define CONFIG_DEFAULT_THREADPOOL_SIZE 4 #define CONFIG_DEFAULT_THREADPOOL_SIZE 4
#define CONFIG_DEFAULT_CLIENT_TIMEOUT 30 #define CONFIG_DEFAULT_CLIENT_TIMEOUT 30
#define CONFIG_DEFAULT_HEADER_TIMEOUT 15 #define CONFIG_DEFAULT_HEADER_TIMEOUT 15
...@@ -193,6 +194,7 @@ static void _set_defaults(void) ...@@ -193,6 +194,7 @@ static void _set_defaults(void)
_configuration.admin = CONFIG_DEFAULT_ADMIN; _configuration.admin = CONFIG_DEFAULT_ADMIN;
_configuration.client_limit = CONFIG_DEFAULT_CLIENT_LIMIT; _configuration.client_limit = CONFIG_DEFAULT_CLIENT_LIMIT;
_configuration.source_limit = CONFIG_DEFAULT_SOURCE_LIMIT; _configuration.source_limit = CONFIG_DEFAULT_SOURCE_LIMIT;
_configuration.queue_size_limit = CONFIG_DEFAULT_QUEUE_SIZE_LIMIT;
_configuration.threadpool_size = CONFIG_DEFAULT_THREADPOOL_SIZE; _configuration.threadpool_size = CONFIG_DEFAULT_THREADPOOL_SIZE;
_configuration.client_timeout = CONFIG_DEFAULT_CLIENT_TIMEOUT; _configuration.client_timeout = CONFIG_DEFAULT_CLIENT_TIMEOUT;
_configuration.header_timeout = CONFIG_DEFAULT_HEADER_TIMEOUT; _configuration.header_timeout = CONFIG_DEFAULT_HEADER_TIMEOUT;
...@@ -318,6 +320,10 @@ static void _parse_limits(xmlDocPtr doc, xmlNodePtr node) ...@@ -318,6 +320,10 @@ static void _parse_limits(xmlDocPtr doc, xmlNodePtr node)
tmp = (char *)xmlNodeListGetString(doc, node->xmlChildrenNode, 1); tmp = (char *)xmlNodeListGetString(doc, node->xmlChildrenNode, 1);
_configuration.source_limit = atoi(tmp); _configuration.source_limit = atoi(tmp);
if (tmp) xmlFree(tmp); if (tmp) xmlFree(tmp);
} else if (strcmp(node->name, "queue-size") == 0) {
tmp = (char *)xmlNodeListGetString(doc, node->xmlChildrenNode, 1);
_configuration.queue_size_limit = atoi(tmp);
if (tmp) xmlFree(tmp);
} else if (strcmp(node->name, "threadpool") == 0) { } else if (strcmp(node->name, "threadpool") == 0) {
tmp = (char *)xmlNodeListGetString(doc, node->xmlChildrenNode, 1); tmp = (char *)xmlNodeListGetString(doc, node->xmlChildrenNode, 1);
_configuration.threadpool_size = atoi(tmp); _configuration.threadpool_size = atoi(tmp);
...@@ -425,6 +431,11 @@ static void _parse_relay(xmlDocPtr doc, xmlNodePtr node) ...@@ -425,6 +431,11 @@ static void _parse_relay(xmlDocPtr doc, xmlNodePtr node)
relay->localmount = (char *)xmlNodeListGetString( relay->localmount = (char *)xmlNodeListGetString(
doc, node->xmlChildrenNode, 1); doc, node->xmlChildrenNode, 1);
} }
else if (strcmp(node->name, "relay-shoutcast-metadata") == 0) {
tmp = (char *)xmlNodeListGetString(doc, node->xmlChildrenNode, 1);
relay->mp3metadata = atoi(tmp);
if(tmp) xmlFree(tmp);
}
} while ((node = node->next)); } while ((node = node->next));
} }
......
...@@ -20,6 +20,7 @@ typedef struct _relay_server { ...@@ -20,6 +20,7 @@ typedef struct _relay_server {
int port; int port;
char *mount; char *mount;
char *localmount; char *localmount;
int mp3metadata;
struct _relay_server *next; struct _relay_server *next;
} relay_server; } relay_server;
...@@ -44,6 +45,7 @@ typedef struct ice_config_tag ...@@ -44,6 +45,7 @@ typedef struct ice_config_tag
int client_limit; int client_limit;
int source_limit; int source_limit;
long queue_size_limit;
int threadpool_size; int threadpool_size;
int client_timeout; int client_timeout;
int header_timeout; int header_timeout;
......
...@@ -535,6 +535,7 @@ static void handle_metadata_request(client_t *client) ...@@ -535,6 +535,7 @@ static void handle_metadata_request(client_t *client)
free(state->metadata); free(state->metadata);
state->metadata = strdup(value); state->metadata = strdup(value);
state->metadata_age++; state->metadata_age++;
state->metadata_raw = 0;
thread_mutex_unlock(&(state->lock)); thread_mutex_unlock(&(state->lock));
DEBUG2("Metadata on mountpoint %s changed to \"%s\"", mount, value); DEBUG2("Metadata on mountpoint %s changed to \"%s\"", mount, value);
......
...@@ -80,7 +80,7 @@ int format_generic_write_buf_to_client(format_plugin_t *format, ...@@ -80,7 +80,7 @@ int format_generic_write_buf_to_client(format_plugin_t *format,
client_t *client, unsigned char *buf, int len) client_t *client, unsigned char *buf, int len)
{ {
int ret; int ret;
ret = sock_write_bytes(client->con->sock, buf, len); ret = sock_write_bytes(client->con->sock, buf, len);
if(ret < 0) { if(ret < 0) {
......
...@@ -101,15 +101,19 @@ static int send_metadata(client_t *client, mp3_client_data *client_state, ...@@ -101,15 +101,19 @@ static int send_metadata(client_t *client, mp3_client_data *client_state,
return 0; return 0;
} }
fullmetadata_size = strlen(source_state->metadata) + if(source_state->metadata_raw) {
strlen("StreamTitle='';StreamUrl=''") + 1; fullmetadata_size = strlen(source_state->metadata);
fullmetadata = source_state->metadata;
fullmetadata = alloca(fullmetadata_size); }
else {
fullmetadata_size = strlen(source_state->metadata) +
strlen("StreamTitle='';StreamUrl=''") + 1;
memset(fullmetadata, 0, fullmetadata_size); fullmetadata = alloca(fullmetadata_size);
sprintf(fullmetadata, "StreamTitle='%s';StreamUrl=''", sprintf(fullmetadata, "StreamTitle='%s';StreamUrl=''",
source_state->metadata); source_state->metadata);
}
source_age = source_state->metadata_age; source_age = source_state->metadata_age;
send_metadata = source_age != client_state->metadata_age; send_metadata = source_age != client_state->metadata_age;
...@@ -210,14 +214,104 @@ static int format_mp3_get_buffer(format_plugin_t *self, char *data, ...@@ -210,14 +214,104 @@ static int format_mp3_get_buffer(format_plugin_t *self, char *data,
refbuf_t *refbuf; refbuf_t *refbuf;
mp3_state *state = self->_state; mp3_state *state = self->_state;
if(!data) { /* Set this to NULL in case it doesn't get set to a valid buffer later */
*buffer = NULL; *buffer = NULL;
if(!data)
return 0; return 0;
}
if(state->inline_metadata_interval) { if(state->inline_metadata_interval) {
/* Source is sending metadata, handle it... */
while(len > 0) {
int to_read = state->inline_metadata_interval - state->offset;
if(to_read > 0) {
refbuf_t *old_refbuf = *buffer;
if(to_read > len)
to_read = len;
if(old_refbuf) {
refbuf = refbuf_new(to_read + old_refbuf->len);
memcpy(refbuf->data, old_refbuf->data, old_refbuf->len);
memcpy(refbuf->data+old_refbuf->len, data, to_read);
refbuf_release(old_refbuf);
}
else {
refbuf = refbuf_new(to_read);
memcpy(refbuf->data, data, to_read);
}
*buffer = refbuf;
state->offset += to_read;
data += to_read;
len -= to_read;
}
else if(!state->metadata_length) {
/* Next up is the metadata byte... */
unsigned char byte = data[0];
data++;
len--;
/* According to the "spec"... this byte * 16 */
state->metadata_length = byte * 16;
if(state->metadata_length) {
state->metadata_buffer =
calloc(state->metadata_length + 1, 1);
/* Ensure we have a null-terminator even if the source
* stream is invalid.
*/
state->metadata_buffer[state->metadata_length] = 0;
}
else {
state->offset = 0;
}
state->metadata_offset = 0;
}
else {
/* Metadata to read! */
int readable = state->metadata_length - state->metadata_offset;
if(readable > len)
readable = len;
memcpy(state->metadata_buffer + state->metadata_offset,
data, readable);
data += readable;
len -= readable;
if(state->metadata_offset == state->metadata_length)
{
state->offset = 0;
state->metadata_length = 0;
if(state->metadata_length)
{
thread_mutex_lock(&(state->lock));
free(state->metadata);
state->metadata = state->metadata_buffer;
state->metadata_buffer = NULL;
state->metadata_age++;
state->metadata_raw = 1;
thread_mutex_unlock(&(state->lock));
}
}
}
}
/* Either we got a buffer above (in which case it can be used), or
* we set *buffer to NULL in the prologue, so the return value is
* correct anyway...
*/
return 0; return 0;
} }
else { else {
/* Simple case - no metadata, just dump data directly to a buffer */
refbuf = refbuf_new(len); refbuf = refbuf_new(len);
memcpy(refbuf->data, data, len); memcpy(refbuf->data, data, len);
......
...@@ -9,9 +9,15 @@ ...@@ -9,9 +9,15 @@
typedef struct { typedef struct {
char *metadata; char *metadata;
int metadata_age; int metadata_age;
int metadata_formatted; int metadata_raw;
int inline_metadata_interval;
mutex_t lock; mutex_t lock;
/* These are for inline metadata */
int inline_metadata_interval;
int offset;
int metadata_length;
char *metadata_buffer;
int metadata_offset;
} mp3_state; } mp3_state;
format_plugin_t *format_mp3_get_plugin(http_parser_t *parser); format_plugin_t *format_mp3_get_plugin(http_parser_t *parser);
......
...@@ -83,8 +83,12 @@ refbuf_t *refbuf_queue_remove(refbuf_queue_t **queue) ...@@ -83,8 +83,12 @@ refbuf_t *refbuf_queue_remove(refbuf_queue_t **queue)
refbuf = item->refbuf; refbuf = item->refbuf;
item->refbuf = NULL; item->refbuf = NULL;
if(*queue)
(*queue)->total_length = item->total_length - refbuf->len;
free(item); free(item);
return refbuf; return refbuf;
} }
...@@ -95,6 +99,7 @@ void refbuf_queue_insert(refbuf_queue_t **queue, refbuf_t *refbuf) ...@@ -95,6 +99,7 @@ void refbuf_queue_insert(refbuf_queue_t **queue, refbuf_t *refbuf)
item->refbuf = refbuf; item->refbuf = refbuf;
item->next = *queue; item->next = *queue;
item->total_length = item->next->total_length + item->refbuf->len;
*queue = item; *queue = item;
} }
...@@ -111,5 +116,12 @@ int refbuf_queue_size(refbuf_queue_t **queue) ...@@ -111,5 +116,12 @@ int refbuf_queue_size(refbuf_queue_t **queue)
return size; return size;
} }
int refbuf_queue_length(refbuf_queue_t **queue)
{
if(*queue)
return (*queue)->total_length;
else
return 0;
}
...@@ -17,6 +17,7 @@ typedef struct _refbuf_tag ...@@ -17,6 +17,7 @@ typedef struct _refbuf_tag
typedef struct _refbuf_queue_tag typedef struct _refbuf_queue_tag
{ {
refbuf_t *refbuf; refbuf_t *refbuf;
long total_length;
struct _refbuf_queue_tag *next; struct _refbuf_queue_tag *next;
} refbuf_queue_t; } refbuf_queue_t;
...@@ -31,7 +32,11 @@ void refbuf_release(refbuf_t *self); ...@@ -31,7 +32,11 @@ void refbuf_release(refbuf_t *self);
void refbuf_queue_add(refbuf_queue_t **queue, refbuf_t *refbuf); void refbuf_queue_add(refbuf_queue_t **queue, refbuf_t *refbuf);
refbuf_t *refbuf_queue_remove(refbuf_queue_t **queue); refbuf_t *refbuf_queue_remove(refbuf_queue_t **queue);
void refbuf_queue_insert(refbuf_queue_t **queue, refbuf_t *refbuf); void refbuf_queue_insert(refbuf_queue_t **queue, refbuf_t *refbuf);
/* Size in buffers */
int refbuf_queue_size(refbuf_queue_t **queue); int refbuf_queue_size(refbuf_queue_t **queue);
/* Size in bytes */
int refbuf_queue_length(refbuf_queue_t **queue);
#endif /* __REFBUF_H__ */ #endif /* __REFBUF_H__ */
......
...@@ -66,7 +66,7 @@ void slave_shutdown(void) { ...@@ -66,7 +66,7 @@ void slave_shutdown(void) {
} }
static void create_relay_stream(char *server, int port, static void create_relay_stream(char *server, int port,
char *remotemount, char *localmount) char *remotemount, char *localmount, int mp3)
{ {
sock_t streamsock; sock_t streamsock;
char header[4096]; char header[4096];
...@@ -85,7 +85,13 @@ static void create_relay_stream(char *server, int port, ...@@ -85,7 +85,13 @@ static void create_relay_stream(char *server, int port,
return; return;
} }
con = create_connection(streamsock, NULL); con = create_connection(streamsock, NULL);
sock_write(streamsock, "GET %s HTTP/1.0\r\n\r\n", remotemount); if(mp3) {
sock_write(streamsock, "GET %s HTTP/1.0\r\nIcy-MetaData: 1\r\n",
remotemount);
}
else {
sock_write(streamsock, "GET %s HTTP/1.0\r\n\r\n", remotemount);
}
memset(header, 0, sizeof(header)); memset(header, 0, sizeof(header));
if (util_read_header(con->sock, header, 4096) == 0) { if (util_read_header(con->sock, header, 4096) == 0) {
connection_close(con); connection_close(con);
...@@ -168,7 +174,7 @@ static void *_slave_thread(void *arg) { ...@@ -168,7 +174,7 @@ static void *_slave_thread(void *arg) {
create_relay_stream( create_relay_stream(
config_get_config()->master_server, config_get_config()->master_server,
config_get_config()->master_server_port, config_get_config()->master_server_port,
buf, NULL); buf, NULL, 0);
} }
else else
avl_tree_unlock(global.source_tree); avl_tree_unlock(global.source_tree);
...@@ -184,7 +190,7 @@ static void *_slave_thread(void *arg) { ...@@ -184,7 +190,7 @@ static void *_slave_thread(void *arg) {
avl_tree_unlock(global.source_tree); avl_tree_unlock(global.source_tree);
create_relay_stream(relay->server, relay->port, relay->mount, create_relay_stream(relay->server, relay->port, relay->mount,
relay->localmount); relay->localmount, relay->mp3metadata);
} }
else else
avl_tree_unlock(global.source_tree); avl_tree_unlock(global.source_tree);
......
...@@ -153,6 +153,8 @@ void *source_main(void *arg) ...@@ -153,6 +153,8 @@ void *source_main(void *arg)
int i=0; int i=0;
int suppress_yp = 0; int suppress_yp = 0;
long queue_limit = config_get_config()->queue_size_limit;
timeout = config_get_config()->source_timeout; timeout = config_get_config()->source_timeout;
/* grab a read lock, to make sure we get a chance to cleanup */ /* grab a read lock, to make sure we get a chance to cleanup */
...@@ -447,10 +449,8 @@ void *source_main(void *arg) ...@@ -447,10 +449,8 @@ void *source_main(void *arg)
/* if the client is too slow, its queue will slowly build up. /* if the client is too slow, its queue will slowly build up.
** we need to make sure the client is keeping up with the ** we need to make sure the client is keeping up with the
** data, so we'll kick any client who's queue gets to large. ** data, so we'll kick any client who's queue gets to large.
** the queue_limit might need to be tuned, but should work fine.
** TODO: put queue_limit in a config file
*/ */
if (refbuf_queue_size(&client->queue) > 25) { if (refbuf_queue_length(&client->queue) > queue_limit) {
DEBUG0("Client has fallen too far behind, removing"); DEBUG0("Client has fallen too far behind, removing");
client->con->error = 1; client->con->error = 1;
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment