aniani/icecast-server
1
0
Fork 0
mirror of https://gitlab.xiph.org/xiph/icecast-server.git synced 2024-09-29 04:25:55 -04:00
Code

kh13. connection_t/client_t merge, client scheduling changes. intro content

Browse Source 
from listener_add now possible.

svn path=/icecast/branches/kh/icecast/; revision=16524
This commit is contained in:
Karl Heyes 2009-08-28 20:19:49 +00:00
parent 9565edee8e
commit 749691af2b
19 changed files with 521 additions and 371 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
src/admin.c
View File

@ -188,7 +188,7 @@ xmlDocPtr admin_build_sourcelist (const char *mount)
if (source->client) if (source->client)
{ {
snprintf (buf, sizeof(buf), "%lu", snprintf (buf, sizeof(buf), "%lu",
(unsigned long)(now - source->client->con->con_time)); (unsigned long)(now - source->client->connection.con_time));
xmlNewChild (srcnode, NULL, XMLSTR("Connected"), XMLSTR(buf)); xmlNewChild (srcnode, NULL, XMLSTR("Connected"), XMLSTR(buf));
} }
xmlNewChild (srcnode, NULL, XMLSTR("content-type"), xmlNewChild (srcnode, NULL, XMLSTR("content-type"),
@ -289,8 +289,10 @@ void admin_mount_request (client_t *client, const char *uri)
} }
else else
{ {
thread_mutex_lock (&source->lock);
if (source_available (source) == 0) if (source_available (source) == 0)
{ {
thread_mutex_unlock (&source->lock);
avl_tree_unlock (global.source_tree); avl_tree_unlock (global.source_tree);
INFO1("Received admin command on unavailable mount \"%s\"", mount); INFO1("Received admin command on unavailable mount \"%s\"", mount);
client_send_400 (client, "Source is not available"); client_send_400 (client, "Source is not available");
@ -453,6 +455,7 @@ static void command_move_clients(client_t *client, source_t *source,
} }
if (!parameters_passed) { if (!parameters_passed) {
doc = admin_build_sourcelist(source->mount); doc = admin_build_sourcelist(source->mount);
thread_mutex_unlock (&source->lock);
admin_send_response(doc, client, response, "moveclients.xsl"); admin_send_response(doc, client, response, "moveclients.xsl");
xmlFreeDoc(doc); xmlFreeDoc(doc);
return; return;
@ -468,6 +471,7 @@ static void command_move_clients(client_t *client, source_t *source,
snprintf (buf, sizeof(buf), "Clients moved from %s to %s", snprintf (buf, sizeof(buf), "Clients moved from %s to %s",
source->mount, dest_source); source->mount, dest_source);
thread_mutex_unlock (&source->lock);
xmlNewChild(node, NULL, XMLSTR("message"), XMLSTR(buf)); xmlNewChild(node, NULL, XMLSTR("message"), XMLSTR(buf));
xmlNewChild(node, NULL, XMLSTR("return"), XMLSTR("1")); xmlNewChild(node, NULL, XMLSTR("return"), XMLSTR("1"));
@ -613,10 +617,10 @@ static void add_listener_node (xmlNodePtr srcnode, client_t *listener)
xmlNodePtr node = xmlNewChild (srcnode, NULL, XMLSTR("listener"), NULL); xmlNodePtr node = xmlNewChild (srcnode, NULL, XMLSTR("listener"), NULL);
snprintf (buf, sizeof (buf), "%lu", listener->con->id); snprintf (buf, sizeof (buf), "%lu", listener->connection.id);
xmlSetProp (node, XMLSTR("id"), XMLSTR(buf)); xmlSetProp (node, XMLSTR("id"), XMLSTR(buf));
xmlNewChild (node, NULL, XMLSTR("ip"), XMLSTR(listener->con->ip)); xmlNewChild (node, NULL, XMLSTR("ip"), XMLSTR(listener->connection.ip));
useragent = httpp_getvar (listener->parser, "user-agent"); useragent = httpp_getvar (listener->parser, "user-agent");
if (useragent) if (useragent)
@ -626,13 +630,16 @@ static void add_listener_node (xmlNodePtr srcnode, client_t *listener)
xmlFree (str); xmlFree (str);
} }
snprintf (buf, sizeof (buf), "%ld", (long)(source->client->queue_pos - listener->queue_pos)); if (listener->flags & CLIENT_ACTIVE)
snprintf (buf, sizeof (buf), "%ld", (long)(source->client->queue_pos - listener->queue_pos));
else
snprintf (buf, sizeof (buf), "0");
xmlNewChild (node, NULL, XMLSTR("lag"), XMLSTR(buf)); xmlNewChild (node, NULL, XMLSTR("lag"), XMLSTR(buf));
if (listener->worker) if (listener->worker)
{ {
snprintf (buf, sizeof (buf), "%lu", snprintf (buf, sizeof (buf), "%lu",
(unsigned long)(listener->worker->current_time.tv_sec - listener->con->con_time)); (unsigned long)(listener->worker->current_time.tv_sec - listener->connection.con_time));
xmlNewChild (node, NULL, XMLSTR("connected"), XMLSTR(buf)); xmlNewChild (node, NULL, XMLSTR("connected"), XMLSTR(buf));
} }
if (listener->username) if (listener->username)
@ -654,15 +661,12 @@ void admin_source_listeners (source_t *source, xmlNodePtr srcnode)
if (source == NULL) if (source == NULL)
return; return;
thread_mutex_lock (&source->lock);
listener = source->client_list; listener = source->client_list;
while (listener) while (listener)
{ {
add_listener_node (srcnode, listener); add_listener_node (srcnode, listener);
listener = listener->next; listener = listener->next;
} }
thread_mutex_unlock (&source->lock);
} }
@ -734,20 +738,19 @@ static void command_show_listeners(client_t *client, source_t *source,
else else
{ {
client_t *listener; client_t *listener;
thread_mutex_lock (&source->lock);
listener = source->client_list; listener = source->client_list;
while (listener) while (listener)
{ {
if (listener->con->id == id) if (listener->connection.id == id)
{ {
add_listener_node (srcnode, listener); add_listener_node (srcnode, listener);
break; break;
} }
listener = listener->next; listener = listener->next;
} }
thread_mutex_unlock (&source->lock);
} }
thread_mutex_unlock (&source->lock);
admin_send_response(doc, client, response, "listclients.xsl"); admin_send_response(doc, client, response, "listclients.xsl");
xmlFreeDoc(doc); xmlFreeDoc(doc);
@ -871,6 +874,7 @@ static void command_manageauth(client_t *client, source_t *source,
node = xmlNewDocNode(doc, NULL, XMLSTR("icestats"), NULL); node = xmlNewDocNode(doc, NULL, XMLSTR("icestats"), NULL);
srcnode = xmlNewChild(node, NULL, XMLSTR("source"), NULL); srcnode = xmlNewChild(node, NULL, XMLSTR("source"), NULL);
xmlSetProp(srcnode, XMLSTR "mount", XMLSTR(source->mount)); xmlSetProp(srcnode, XMLSTR "mount", XMLSTR(source->mount));
thread_mutex_unlock (&source->lock);
if (message) { if (message) {
msgnode = xmlNewChild(node, NULL, XMLSTR("iceresponse"), NULL); msgnode = xmlNewChild(node, NULL, XMLSTR("iceresponse"), NULL);
@ -890,6 +894,7 @@ static void command_manageauth(client_t *client, source_t *source,
return; return;
} while (0); } while (0);
thread_mutex_unlock (&source->lock);
config_release_config (); config_release_config ();
client_send_400 (client, "missing parameter"); client_send_400 (client, "missing parameter");
} }
@ -908,6 +913,7 @@ static void command_kill_source(client_t *client, source_t *source,
source->flags &= ~SOURCE_RUNNING; source->flags &= ~SOURCE_RUNNING;
thread_mutex_unlock (&source->lock);
admin_send_response(doc, client, response, "response.xsl"); admin_send_response(doc, client, response, "response.xsl");
xmlFreeDoc(doc); xmlFreeDoc(doc);
} }
@ -926,7 +932,6 @@ static void command_kill_client(client_t *client, source_t *source,
id = atoi(idtext); id = atoi(idtext);
thread_mutex_lock (&source->lock);
listener = source_find_client(source, id); listener = source_find_client(source, id);
doc = xmlNewDoc(XMLSTR("1.0")); doc = xmlNewDoc(XMLSTR("1.0"));
@ -939,7 +944,7 @@ static void command_kill_client(client_t *client, source_t *source,
/* This tags it for removal on the next iteration of the main source /* This tags it for removal on the next iteration of the main source
* loop * loop
*/ */
listener->con->error = 1; listener->connection.error = 1;
snprintf(buf, sizeof(buf), "Client %d removed", id); snprintf(buf, sizeof(buf), "Client %d removed", id);
xmlNewChild(node, NULL, XMLSTR("message"), XMLSTR(buf)); xmlNewChild(node, NULL, XMLSTR("message"), XMLSTR(buf));
xmlNewChild(node, NULL, XMLSTR("return"), XMLSTR("1")); xmlNewChild(node, NULL, XMLSTR("return"), XMLSTR("1"));
@ -958,13 +963,30 @@ static void command_kill_client(client_t *client, source_t *source,
static void command_fallback(client_t *client, source_t *source, static void command_fallback(client_t *client, source_t *source,
int response) int response)
{ {
const char *fallback; char *mount = strdup (source->mount);
mount_proxy *mountinfo;
ice_config_t *config;
thread_mutex_unlock (&source->lock);
DEBUG0("Got fallback request"); DEBUG0("Got fallback request");
config = config_grab_config();
mountinfo = config_find_mount (config, mount);
free (mount);
if (mountinfo)
{
const char *fallback;
char buffer[200];
COMMAND_REQUIRE(client, "fallback", fallback);
COMMAND_REQUIRE(client, "fallback", fallback); xmlFree (mountinfo->fallback_mount);
mountinfo->fallback_mount = (char *)xmlCharStrdup (fallback);
client_send_400 (client, "not implemented"); snprintf (buffer, sizeof (buffer), "Fallback for \"%s\" configured", mountinfo->mountname);
config_release_config ();
html_success (client, buffer);
return;
}
config_release_config ();
client_send_400 (client, "no mount details available");
} }
static void command_metadata(client_t *client, source_t *source, static void command_metadata(client_t *client, source_t *source,
@ -989,10 +1011,8 @@ static void command_metadata(client_t *client, source_t *source,
COMMAND_OPTIONAL(client, "artwork", artwork); COMMAND_OPTIONAL(client, "artwork", artwork);
COMMAND_OPTIONAL(client, "charset", charset); COMMAND_OPTIONAL(client, "charset", charset);
thread_mutex_lock (&source->lock);
plugin = source->format; plugin = source->format;
if (source->client && strcmp (client->con->ip, source->client->con->ip) != 0) if (source->client && strcmp (client->connection.ip, source->client->connection.ip) != 0)
if (response == RAW && connection_check_admin_pass (client->parser) == 0) if (response == RAW && connection_check_admin_pass (client->parser) == 0)
same_ip = 0; same_ip = 0;
@ -1067,12 +1087,13 @@ static void command_shoutcast_metadata(client_t *client, source_t *source)
if ((source->flags & SOURCE_SHOUTCAST_COMPAT) == 0) if ((source->flags & SOURCE_SHOUTCAST_COMPAT) == 0)
{ {
thread_mutex_unlock (&source->lock);
ERROR0 ("illegal change of metadata on non-shoutcast compatible stream"); ERROR0 ("illegal change of metadata on non-shoutcast compatible stream");
client_send_400 (client, "illegal metadata call"); client_send_400 (client, "illegal metadata call");
return; return;
} }
if (source->client && strcmp (client->con->ip, source->client->con->ip) != 0) if (source->client && strcmp (client->connection.ip, source->client->connection.ip) != 0)
if (connection_check_admin_pass (client->parser) == 0) if (connection_check_admin_pass (client->parser) == 0)
same_ip = 0; same_ip = 0;
@ -1181,6 +1202,8 @@ void command_list_mounts(client_t *client, int response)
{ {
DEBUG0("List mounts request"); DEBUG0("List mounts request");
client_set_queue (client, NULL);
client->refbuf = refbuf_new (PER_CLIENT_REFBUF_SIZE);
if (response == TEXT) if (response == TEXT)
{ {
redirector_update (client); redirector_update (client);
@ -1214,6 +1237,7 @@ static void command_updatemetadata(client_t *client, source_t *source,
xmlDocPtr doc; xmlDocPtr doc;
xmlNodePtr node, srcnode; xmlNodePtr node, srcnode;
thread_mutex_unlock (&source->lock);
doc = xmlNewDoc(XMLSTR("1.0")); doc = xmlNewDoc(XMLSTR("1.0"));
node = xmlNewDocNode(doc, NULL, XMLSTR("icestats"), NULL); node = xmlNewDocNode(doc, NULL, XMLSTR("icestats"), NULL);
srcnode = xmlNewChild(node, NULL, XMLSTR("source"), NULL); srcnode = xmlNewChild(node, NULL, XMLSTR("source"), NULL);

47
src/auth.c
View File

@ -189,7 +189,7 @@ static void auth_client_free (auth_client *auth_user)
client_t *client = auth_user->client; client_t *client = auth_user->client;
if (client->respcode) if (client->respcode)
client->con->error = 1; client->connection.error = 1;
client_send_401 (client, auth_user->auth->realm); client_send_401 (client, auth_user->auth->realm);
client->flags |= CLIENT_ACTIVE; client->flags |= CLIENT_ACTIVE;
auth_user->client = NULL; auth_user->client = NULL;
@ -207,7 +207,7 @@ static int is_listener_connected (client_t *client)
int ret = 1; int ret = 1;
if (client) if (client)
{ {
if (sock_active (client->con->sock) == 0) if (sock_active (client->connection.sock) == 0)
ret = 0; ret = 0;
} }
return ret; return ret;
@ -338,7 +338,19 @@ static void *auth_run_thread (void *arg)
auth_user->handler = handler->id; auth_user->handler = handler->id;
if (auth_user->process) if (auth_user->process)
{
worker_t *worker = NULL;
if (auth_user->client)
worker = auth_user->client->worker;
auth_user->process (auth_user); auth_user->process (auth_user);
if (worker)
{
/* wakeup worker for new client */
thread_mutex_lock (&worker->lock);
thread_cond_signal (&worker->cond);
thread_mutex_unlock (&worker->lock);
}
}
else else
ERROR0 ("client auth process not set"); ERROR0 ("client auth process not set");
@ -360,20 +372,27 @@ void move_listener (client_t *client, struct _fbinfo *finfo)
DEBUG1 ("moving listener to %s", finfo->mount); DEBUG1 ("moving listener to %s", finfo->mount);
avl_tree_rlock (global.source_tree); avl_tree_rlock (global.source_tree);
source = source_find_mount (finfo->mount);
if (source && source_available (source)) source = source_find_mount (finfo->mount);
while (source)
{ {
thread_mutex_lock (&source->lock); thread_mutex_lock (&source->lock);
avl_tree_unlock (global.source_tree); if (source_available (source))
source_setup_listener (source, client); {
thread_mutex_unlock (&source->lock); avl_tree_unlock (global.source_tree);
} source_setup_listener (source, client);
else thread_mutex_unlock (&source->lock);
{ return;
avl_tree_unlock (global.source_tree); }
fserve_setup_client_fb (client, finfo); if (source->fallback.mount)
{
source_t *prev = source;
source = source_find_mount (prev->fallback.mount);
thread_mutex_unlock (&prev->lock);
}
} }
avl_tree_unlock (global.source_tree);
fserve_setup_client_fb (client, finfo);
} }
@ -441,10 +460,12 @@ static int auth_postprocess_listener (auth_client *auth_user)
ice_config_t *config; ice_config_t *config;
mount_proxy *mountinfo; mount_proxy *mountinfo;
const char *mount = auth_user->mount; const char *mount = auth_user->mount;
worker_t *worker;
if (client == NULL) if (client == NULL)
return -1; return -1;
worker = client->worker;
if ((client->flags & CLIENT_AUTHENTICATED) == 0) if ((client->flags & CLIENT_AUTHENTICATED) == 0)
{ {
/* auth failed so do we place the listener elsewhere */ /* auth failed so do we place the listener elsewhere */
@ -557,6 +578,8 @@ int auth_release_listener (client_t *client, const char *mount, mount_proxy *mou
return 1; return 1;
} }
client->flags &= ~CLIENT_AUTHENTICATED; client->flags &= ~CLIENT_AUTHENTICATED;
client_destroy (client);
return 1;
} }
client_send_404 (client, NULL); client_send_404 (client, NULL);
return 0; return 0;

62
src/auth_url.c
View File

@ -114,6 +114,11 @@ typedef struct {
} auth_url; } auth_url;
struct build_intro_contents
{
refbuf_t *head, **tailp;
};
static void auth_url_clear(auth_t *self) static void auth_url_clear(auth_t *self)
{ {
auth_url *url; auth_url *url;
@ -174,7 +179,7 @@ static int handle_returned_header (void *ptr, size_t size, size_t nmemb, void *s
{ {
unsigned int limit = 0; unsigned int limit = 0;
sscanf ((char *)ptr+url->timelimit_header_len, "%u\r\n", &limit); sscanf ((char *)ptr+url->timelimit_header_len, "%u\r\n", &limit);
client->con->discon_time = time(NULL) + limit; client->connection.discon_time = time(NULL) + limit;
} }
if (strncasecmp (ptr, "icecast-slave: 1", 16) == 0) if (strncasecmp (ptr, "icecast-slave: 1", 16) == 0)
client->flags |= CLIENT_IS_SLAVE; client->flags |= CLIENT_IS_SLAVE;
@ -206,10 +211,26 @@ static int handle_returned_header (void *ptr, size_t size, size_t nmemb, void *s
return (int)bytes; return (int)bytes;
} }
/* capture returned data, but don't do anything with it */
static int handle_returned_data (void *ptr, size_t size, size_t nmemb, void *stream) static int handle_returned_data (void *ptr, size_t size, size_t nmemb, void *stream)
{ {
return (int)(size*nmemb); auth_client *auth_user = stream;
unsigned bytes = size * nmemb;
client_t *client = auth_user->client;
if (client && client->respcode == 0)
{
refbuf_t *n, *r = client->refbuf;
struct build_intro_contents *x = (void*)r->data;
client->flags |= CLIENT_HAS_INTRO_CONTENT;
n = refbuf_new (bytes);
memcpy (n->data, ptr, bytes);
*x->tailp = n;
x->tailp = &n->next;
}
return (int)(bytes);
} }
@ -218,7 +239,7 @@ static auth_result url_remove_listener (auth_client *auth_user)
client_t *client = auth_user->client; client_t *client = auth_user->client;
auth_url *url = auth_user->auth->state; auth_url *url = auth_user->auth->state;
auth_thread_data *atd = auth_user->thread_data; auth_thread_data *atd = auth_user->thread_data;
time_t duration = time(NULL) - client->con->con_time; time_t duration = time(NULL) - client->connection.con_time;
char *username, *password, *mount, *server, *ipaddr; char *username, *password, *mount, *server, *ipaddr;
const char *qargs; const char *qargs;
char *userpwd = NULL, post [4096]; char *userpwd = NULL, post [4096];
@ -241,12 +262,12 @@ static auth_result url_remove_listener (auth_client *auth_user)
qargs = httpp_getvar (client->parser, HTTPP_VAR_QUERYARGS); qargs = httpp_getvar (client->parser, HTTPP_VAR_QUERYARGS);
snprintf (post, sizeof post, "%s%s", auth_user->mount, qargs ? qargs : ""); snprintf (post, sizeof post, "%s%s", auth_user->mount, qargs ? qargs : "");
mount = util_url_escape (post); mount = util_url_escape (post);
ipaddr = util_url_escape (client->con->ip); ipaddr = util_url_escape (client->connection.ip);
snprintf (post, sizeof (post), snprintf (post, sizeof (post),
"action=listener_remove&server=%s&port=%d&client=%lu&mount=%s" "action=listener_remove&server=%s&port=%d&client=%lu&mount=%s"
"&user=%s&pass=%s&ip=%s&duration=%lu", "&user=%s&pass=%s&ip=%s&duration=%lu",
server, auth_user->port, client->con->id, mount, username, server, auth_user->port, client->connection.id, mount, username,
password, ipaddr, (long unsigned)duration); password, ipaddr, (long unsigned)duration);
free (ipaddr); free (ipaddr);
free (server); free (server);
@ -280,6 +301,7 @@ static auth_result url_remove_listener (auth_client *auth_user)
curl_easy_setopt (atd->curl, CURLOPT_URL, url->removeurl); curl_easy_setopt (atd->curl, CURLOPT_URL, url->removeurl);
curl_easy_setopt (atd->curl, CURLOPT_POSTFIELDS, post); curl_easy_setopt (atd->curl, CURLOPT_POSTFIELDS, post);
curl_easy_setopt (atd->curl, CURLOPT_WRITEHEADER, auth_user); curl_easy_setopt (atd->curl, CURLOPT_WRITEHEADER, auth_user);
curl_easy_setopt (atd->curl, CURLOPT_WRITEDATA, auth_user);
DEBUG1 ("...handler %d sending request", auth_user->handler); DEBUG1 ("...handler %d sending request", auth_user->handler);
if (curl_easy_perform (atd->curl)) if (curl_easy_perform (atd->curl))
@ -304,6 +326,7 @@ static auth_result url_add_listener (auth_client *auth_user)
char *user_agent, *username, *password; char *user_agent, *username, *password;
char *mount, *ipaddr, *server; char *mount, *ipaddr, *server;
ice_config_t *config; ice_config_t *config;
struct build_intro_contents *x;
char *userpwd = NULL, post [4096]; char *userpwd = NULL, post [4096];
if (url->addurl == NULL) if (url->addurl == NULL)
@ -330,12 +353,12 @@ static auth_result url_add_listener (auth_client *auth_user)
qargs = httpp_getvar (client->parser, HTTPP_VAR_QUERYARGS); qargs = httpp_getvar (client->parser, HTTPP_VAR_QUERYARGS);
snprintf (post, sizeof post, "%s%s", auth_user->mount, qargs ? qargs : ""); snprintf (post, sizeof post, "%s%s", auth_user->mount, qargs ? qargs : "");
mount = util_url_escape (post); mount = util_url_escape (post);
ipaddr = util_url_escape (client->con->ip); ipaddr = util_url_escape (client->connection.ip);
snprintf (post, sizeof (post), snprintf (post, sizeof (post),
"action=listener_add&server=%s&port=%d&client=%lu&mount=%s" "action=listener_add&server=%s&port=%d&client=%lu&mount=%s"
"&user=%s&pass=%s&ip=%s&agent=%s", "&user=%s&pass=%s&ip=%s&agent=%s",
server, port, client->con->id, mount, username, server, port, client->connection.id, mount, username,
password, ipaddr, user_agent); password, ipaddr, user_agent);
free (server); free (server);
free (mount); free (mount);
@ -370,7 +393,12 @@ static auth_result url_add_listener (auth_client *auth_user)
curl_easy_setopt (atd->curl, CURLOPT_URL, url->addurl); curl_easy_setopt (atd->curl, CURLOPT_URL, url->addurl);
curl_easy_setopt (atd->curl, CURLOPT_POSTFIELDS, post); curl_easy_setopt (atd->curl, CURLOPT_POSTFIELDS, post);
curl_easy_setopt (atd->curl, CURLOPT_WRITEHEADER, auth_user); curl_easy_setopt (atd->curl, CURLOPT_WRITEHEADER, auth_user);
curl_easy_setopt (atd->curl, CURLOPT_WRITEDATA, auth_user);
atd->errormsg[0] = '\0'; atd->errormsg[0] = '\0';
/* setup in case intro data is returned */
x = (void *)client->refbuf->data;
x->head = NULL;
x->tailp = &x->head;
DEBUG1 ("handler %d sending request", auth_user->handler); DEBUG1 ("handler %d sending request", auth_user->handler);
res = curl_easy_perform (atd->curl); res = curl_easy_perform (atd->curl);
@ -395,7 +423,11 @@ static auth_result url_add_listener (auth_client *auth_user)
} }
/* we received a response, lets see what it is */ /* we received a response, lets see what it is */
if (client->flags & CLIENT_AUTHENTICATED) if (client->flags & CLIENT_AUTHENTICATED)
{
if (client->flags & CLIENT_HAS_INTRO_CONTENT)
client->refbuf->next = x->head;
return AUTH_OK; return AUTH_OK;
}
if (atoi (atd->errormsg) == 403) if (atoi (atd->errormsg) == 403)
{ {
client_send_403 (client, atd->errormsg+4); client_send_403 (client, atd->errormsg+4);
@ -419,8 +451,8 @@ static void url_stream_start (auth_client *auth_user)
server = util_url_escape (auth_user->hostname); server = util_url_escape (auth_user->hostname);
mount = util_url_escape (auth_user->mount); mount = util_url_escape (auth_user->mount);
if (client && client->con) if (client && client->connection.ip)
ipaddr = util_url_escape (client->con->ip); ipaddr = util_url_escape (client->connection.ip);
else else
ipaddr = strdup(""); ipaddr = strdup("");
@ -442,6 +474,7 @@ static void url_stream_start (auth_client *auth_user)
curl_easy_setopt (atd->curl, CURLOPT_URL, url->stream_start); curl_easy_setopt (atd->curl, CURLOPT_URL, url->stream_start);
curl_easy_setopt (atd->curl, CURLOPT_POSTFIELDS, post); curl_easy_setopt (atd->curl, CURLOPT_POSTFIELDS, post);
curl_easy_setopt (atd->curl, CURLOPT_WRITEHEADER, auth_user); curl_easy_setopt (atd->curl, CURLOPT_WRITEHEADER, auth_user);
curl_easy_setopt (atd->curl, CURLOPT_WRITEDATA, auth_user);
DEBUG1 ("handler %d sending request", auth_user->handler); DEBUG1 ("handler %d sending request", auth_user->handler);
if (curl_easy_perform (atd->curl)) if (curl_easy_perform (atd->curl))
@ -460,8 +493,8 @@ static void url_stream_end (auth_client *auth_user)
server = util_url_escape (auth_user->hostname); server = util_url_escape (auth_user->hostname);
mount = util_url_escape (auth_user->mount); mount = util_url_escape (auth_user->mount);
if (client && client->con) if (client && client->connection.ip)
ipaddr = util_url_escape (client->con->ip); ipaddr = util_url_escape (client->connection.ip);
else else
ipaddr = strdup(""); ipaddr = strdup("");
@ -483,6 +516,7 @@ static void url_stream_end (auth_client *auth_user)
curl_easy_setopt (atd->curl, CURLOPT_URL, url->stream_end); curl_easy_setopt (atd->curl, CURLOPT_URL, url->stream_end);
curl_easy_setopt (atd->curl, CURLOPT_POSTFIELDS, post); curl_easy_setopt (atd->curl, CURLOPT_POSTFIELDS, post);
curl_easy_setopt (atd->curl, CURLOPT_WRITEHEADER, auth_user); curl_easy_setopt (atd->curl, CURLOPT_WRITEHEADER, auth_user);
curl_easy_setopt (atd->curl, CURLOPT_WRITEDATA, auth_user);
DEBUG1 ("handler %d sending request", auth_user->handler); DEBUG1 ("handler %d sending request", auth_user->handler);
if (curl_easy_perform (atd->curl)) if (curl_easy_perform (atd->curl))
@ -511,13 +545,14 @@ static void url_stream_auth (auth_client *auth_user)
curl_easy_setopt (atd->curl, CURLOPT_URL, url->stream_auth); curl_easy_setopt (atd->curl, CURLOPT_URL, url->stream_auth);
curl_easy_setopt (atd->curl, CURLOPT_POSTFIELDS, post); curl_easy_setopt (atd->curl, CURLOPT_POSTFIELDS, post);
curl_easy_setopt (atd->curl, CURLOPT_WRITEHEADER, auth_user); curl_easy_setopt (atd->curl, CURLOPT_WRITEHEADER, auth_user);
curl_easy_setopt (atd->curl, CURLOPT_WRITEDATA, auth_user);
if (strcmp (auth_user->mount, httpp_getvar (client->parser, HTTPP_VAR_URI)) != 0) if (strcmp (auth_user->mount, httpp_getvar (client->parser, HTTPP_VAR_URI)) != 0)
admin = "&admin=1"; admin = "&admin=1";
mount = util_url_escape (auth_user->mount); mount = util_url_escape (auth_user->mount);
host = util_url_escape (auth_user->hostname); host = util_url_escape (auth_user->hostname);
user = util_url_escape (client->username); user = util_url_escape (client->username);
pass = util_url_escape (client->password); pass = util_url_escape (client->password);
ipaddr = util_url_escape (client->con->ip); ipaddr = util_url_escape (client->connection.ip);
snprintf (post, sizeof (post), snprintf (post, sizeof (post),
"action=stream_auth&mount=%s&ip=%s&server=%s&port=%d&user=%s&pass=%s%s", "action=stream_auth&mount=%s&ip=%s&server=%s&port=%d&user=%s&pass=%s%s",
@ -560,7 +595,6 @@ static void *alloc_thread_data (auth_t *auth)
curl_easy_setopt (atd->curl, CURLOPT_USERAGENT, atd->server_id); curl_easy_setopt (atd->curl, CURLOPT_USERAGENT, atd->server_id);
curl_easy_setopt (atd->curl, CURLOPT_HEADERFUNCTION, handle_returned_header); curl_easy_setopt (atd->curl, CURLOPT_HEADERFUNCTION, handle_returned_header);
curl_easy_setopt (atd->curl, CURLOPT_WRITEFUNCTION, handle_returned_data); curl_easy_setopt (atd->curl, CURLOPT_WRITEFUNCTION, handle_returned_data);
curl_easy_setopt (atd->curl, CURLOPT_WRITEDATA, atd);
curl_easy_setopt (atd->curl, CURLOPT_NOSIGNAL, 1L); curl_easy_setopt (atd->curl, CURLOPT_NOSIGNAL, 1L);
curl_easy_setopt (atd->curl, CURLOPT_TIMEOUT, 6L); curl_easy_setopt (atd->curl, CURLOPT_TIMEOUT, 6L);
#ifdef CURLOPT_PASSWDFUNCTION #ifdef CURLOPT_PASSWDFUNCTION

151
src/client.c
View File

@ -45,34 +45,28 @@
int worker_count; int worker_count;
/* create a client_t with the provided connection and parser details. Return /* Return client_t ready for use. The provided socket can be SOCK_ERROR to
* client_t ready for use. Should be called with global lock held. * allocate a dummy client_t. Must be called with global lock held.
*/ */
client_t *client_create (connection_t *con, http_parser_t *parser) client_t *client_create (sock_t sock)
{ {
client_t *client = (client_t *)calloc(1, sizeof(client_t)); client_t *client = calloc (1, sizeof (client_t));
if (client == NULL) if (sock != SOCK_ERROR)
abort();
global.clients++;
if (con && con->serversock != SOCK_ERROR)
{ {
int i; refbuf_t *r;
for (i=0; i < global.server_sockets; i++) if (connection_init (&client->connection, sock) < 0)
{ {
if (global.serversock[i] == con->serversock) free (client);
{ return NULL;
client->server_conn = global.server_conn[i];
client->server_conn->refcount++;
}
} }
r = refbuf_new (PER_CLIENT_REFBUF_SIZE);
r->len = 0;
client->shared_data = r;
client->flags |= CLIENT_ACTIVE;
} }
global.clients++;
stats_event_args (NULL, "clients", "%d", global.clients); stats_event_args (NULL, "clients", "%d", global.clients);
client->con = con;
client->parser = parser;
client->pos = 0;
return client; return client;
} }
@ -82,6 +76,11 @@ void client_destroy(client_t *client)
if (client == NULL) if (client == NULL)
return; return;
if (client->worker)
{
WARN0 ("client still on worker thread");
return;
}
/* release the buffer now, as the buffer could be on the source queue /* release the buffer now, as the buffer could be on the source queue
* and may of disappeared after auth completes */ * and may of disappeared after auth completes */
if (client->refbuf) if (client->refbuf)
@ -99,8 +98,7 @@ void client_destroy(client_t *client)
if (client->respcode && client->parser) if (client->respcode && client->parser)
logging_access(client); logging_access(client);
if (client->con) connection_close (&client->connection);
connection_close(client->con);
if (client->parser) if (client->parser)
httpp_destroy (client->parser); httpp_destroy (client->parser);
@ -124,6 +122,7 @@ void client_destroy(client_t *client)
/* helper function for reading data from a client */ /* helper function for reading data from a client */
int client_read_bytes (client_t *client, void *buf, unsigned len) int client_read_bytes (client_t *client, void *buf, unsigned len)
{ {
int (*con_read)(struct connection_tag *handle, void *buf, size_t len) = connection_read;
int bytes; int bytes;
if (client->refbuf && client->pos < client->refbuf->len) if (client->refbuf && client->pos < client->refbuf->len)
@ -135,9 +134,13 @@ int client_read_bytes (client_t *client, void *buf, unsigned len)
client->pos += remaining; client->pos += remaining;
return remaining; return remaining;
} }
bytes = client->con->read (client->con, buf, len); #ifdef HAVE_OPENSSL
if (client->connection.ssl)
con_read = connection_read_ssl;
#endif
bytes = con_read (&client->connection, buf, len);
if (bytes == -1 && client->con->error) if (bytes == -1 && client->connection.error)
DEBUG0 ("reading from connection has failed"); DEBUG0 ("reading from connection has failed");
return bytes; return bytes;
@ -146,6 +149,8 @@ int client_read_bytes (client_t *client, void *buf, unsigned len)
void client_send_302(client_t *client, const char *location) void client_send_302(client_t *client, const char *location)
{ {
client_set_queue (client, NULL);
client->refbuf = refbuf_new (PER_CLIENT_REFBUF_SIZE);
snprintf (client->refbuf->data, PER_CLIENT_REFBUF_SIZE, snprintf (client->refbuf->data, PER_CLIENT_REFBUF_SIZE,
"HTTP/1.0 302 Temporarily Moved\r\n" "HTTP/1.0 302 Temporarily Moved\r\n"
"Content-Type: text/html\r\n" "Content-Type: text/html\r\n"
@ -158,6 +163,8 @@ void client_send_302(client_t *client, const char *location)
void client_send_400(client_t *client, char *message) { void client_send_400(client_t *client, char *message) {
client_set_queue (client, NULL);
client->refbuf = refbuf_new (PER_CLIENT_REFBUF_SIZE);
snprintf (client->refbuf->data, PER_CLIENT_REFBUF_SIZE, snprintf (client->refbuf->data, PER_CLIENT_REFBUF_SIZE,
"HTTP/1.0 400 Bad Request\r\n" "HTTP/1.0 400 Bad Request\r\n"
"Content-Type: text/html\r\n\r\n" "Content-Type: text/html\r\n\r\n"
@ -175,7 +182,7 @@ void client_send_401 (client_t *client, const char *realm)
if (realm == NULL) if (realm == NULL)
realm = config->server_id; realm = config->server_id;
refbuf_release (client->refbuf); client_set_queue (client, NULL);
client->refbuf = refbuf_new (500); client->refbuf = refbuf_new (500);
snprintf (client->refbuf->data, 500, snprintf (client->refbuf->data, 500,
"HTTP/1.0 401 Authentication Required\r\n" "HTTP/1.0 401 Authentication Required\r\n"
@ -193,6 +200,8 @@ void client_send_403(client_t *client, const char *reason)
{ {
if (reason == NULL) if (reason == NULL)
reason = "Forbidden"; reason = "Forbidden";
client_set_queue (client, NULL);
client->refbuf = refbuf_new (PER_CLIENT_REFBUF_SIZE);
snprintf (client->refbuf->data, PER_CLIENT_REFBUF_SIZE, snprintf (client->refbuf->data, PER_CLIENT_REFBUF_SIZE,
"HTTP/1.0 403 %s\r\n" "HTTP/1.0 403 %s\r\n"
"Content-Type: text/html\r\n\r\n", reason); "Content-Type: text/html\r\n\r\n", reason);
@ -228,13 +237,15 @@ void client_send_404(client_t *client, const char *message)
"<b>%s</b>\r\n", message); "<b>%s</b>\r\n", message);
client->respcode = 404; client->respcode = 404;
client->refbuf->len = strlen (client->refbuf->data); client->refbuf->len = strlen (client->refbuf->data);
fserve_setup_client (client, NULL);
} }
fserve_setup_client (client, NULL);
} }
void client_send_416(client_t *client) void client_send_416(client_t *client)
{ {
client_set_queue (client, NULL);
client->refbuf = refbuf_new (PER_CLIENT_REFBUF_SIZE);
snprintf (client->refbuf->data, PER_CLIENT_REFBUF_SIZE, snprintf (client->refbuf->data, PER_CLIENT_REFBUF_SIZE,
"HTTP/1.0 416 Request Range Not Satisfiable\r\n\r\n"); "HTTP/1.0 416 Request Range Not Satisfiable\r\n\r\n");
client->respcode = 416; client->respcode = 416;
@ -246,42 +257,37 @@ void client_send_416(client_t *client)
/* helper function for sending the data to a client */ /* helper function for sending the data to a client */
int client_send_bytes (client_t *client, const void *buf, unsigned len) int client_send_bytes (client_t *client, const void *buf, unsigned len)
{ {
#ifdef HAVE_AIO int (*con_send)(struct connection_tag *handle, const void *buf, size_t len) = connection_send;
int ret, err; int ret;
struct aiocb *aiocbp = &client->aio; #ifdef HAVE_OPENSSL
if (client->connection.ssl)
con_send = connection_send_ssl;
#endif
ret = con_send (&client->connection, buf, len);
if (client->pending_io == 0) if (client->connection.error)
{
memset (aiocbp, 0 , sizeof (struct aiocb));
aiocbp->aio_fildes = client->con->sock;
aiocbp->aio_buf = (void*)buf; /* only read from */
aiocbp->aio_nbytes = len;
if (aio_write (aiocbp) < 0)
return -1;
client->pending_io = 1;
}
if ((err = aio_error (aiocbp)) == EINPROGRESS)
return -1;
ret = aio_return (aiocbp);
if (ret < 0)
sock_set_error (err); /* make sure errno gets set */
client->pending_io = 0;
#else
int ret = client->con->send (client->con, buf, len);
if (client->con->error)
DEBUG0 ("Client connection died"); DEBUG0 ("Client connection died");
return ret; return ret;
#endif
} }
void client_set_queue (client_t *client, refbuf_t *refbuf) void client_set_queue (client_t *client, refbuf_t *refbuf)
{ {
refbuf_t *to_release = client->refbuf; refbuf_t *to_release = client->refbuf;
if (to_release && client->flags & CLIENT_HAS_INTRO_CONTENT)
{
refbuf_t *intro = to_release->next;
while (intro)
{
refbuf_t *r = intro->next;
intro->next = NULL;
refbuf_release (intro);
intro = r;
}
to_release->next = NULL;
client->flags &= ~CLIENT_HAS_INTRO_CONTENT;
}
client->refbuf = refbuf; client->refbuf = refbuf;
if (refbuf) if (refbuf)
refbuf_addref (client->refbuf); refbuf_addref (client->refbuf);
@ -293,12 +299,11 @@ void client_set_queue (client_t *client, refbuf_t *refbuf)
worker_t *find_least_busy_handler (void) worker_t *find_least_busy_handler (void)
{ {
worker_t *handler, *min = NULL; worker_t *min = workers;
if (workers) if (workers && workers->next)
{ {
min = workers; worker_t *handler = workers->next;
handler = workers->next;
DEBUG2 ("handler %p has %d clients", min, min->count); DEBUG2 ("handler %p has %d clients", min, min->count);
while (handler) while (handler)
{ {
@ -320,13 +325,14 @@ void client_change_worker (client_t *client, worker_t *dest_worker)
*this_worker->current_p = client->next_on_worker; *this_worker->current_p = client->next_on_worker;
this_worker->count--; this_worker->count--;
thread_mutex_unlock (&this_worker->lock); thread_mutex_unlock (&this_worker->lock);
client->next_on_worker = NULL;
thread_mutex_lock (&dest_worker->lock); thread_mutex_lock (&dest_worker->lock);
if (dest_worker->running) if (dest_worker->running)
{ {
client->worker = dest_worker; client->worker = dest_worker;
client->next_on_worker = dest_worker->clients; *dest_worker->last_p = client;
dest_worker->clients = client; dest_worker->last_p = &client->next_on_worker;
dest_worker->count++; dest_worker->count++;
client->flags |= CLIENT_HAS_CHANGED_THREAD; client->flags |= CLIENT_HAS_CHANGED_THREAD;
// make client inactive so that the destination thread does not run it straight away // make client inactive so that the destination thread does not run it straight away
@ -348,8 +354,8 @@ void client_add_worker (client_t *client)
thread_rwlock_unlock (&workers_lock); thread_rwlock_unlock (&workers_lock);
client->schedule_ms = handler->time_ms; client->schedule_ms = handler->time_ms;
client->next_on_worker = handler->clients; *handler->last_p = client;
handler->clients = client; handler->last_p = &client->next_on_worker;
client->worker = handler; client->worker = handler;
++handler->count; ++handler->count;
if (handler->wakeup_ms - handler->time_ms > 15) if (handler->wakeup_ms - handler->time_ms > 15)
@ -405,6 +411,8 @@ void *worker (void *arg)
client->flags &= ~CLIENT_HAS_CHANGED_THREAD; client->flags &= ~CLIENT_HAS_CHANGED_THREAD;
client->flags |= CLIENT_ACTIVE; client->flags |= CLIENT_ACTIVE;
client = *prevp; client = *prevp;
if (client == NULL)
handler->last_p = prevp;
continue; continue;
} }
if (ret < 0) if (ret < 0)
@ -417,6 +425,8 @@ void *worker (void *arg)
if (client->ops->release) if (client->ops->release)
client->ops->release (client); client->ops->release (client);
client = *prevp; client = *prevp;
if (client == NULL)
handler->last_p = prevp;
continue; continue;
} }
} }
@ -444,6 +454,7 @@ static void worker_start (void)
thread_mutex_create (&handler->lock); thread_mutex_create (&handler->lock);
thread_cond_create (&handler->cond); thread_cond_create (&handler->cond);
thread_rwlock_wlock (&workers_lock); thread_rwlock_wlock (&workers_lock);
handler->last_p = &handler->clients;
handler->next = workers; handler->next = workers;
workers = handler; workers = handler;
worker_count++; worker_count++;
@ -454,7 +465,8 @@ static void worker_start (void)
static void worker_stop (void) static void worker_stop (void)
{ {
worker_t *handler = workers; worker_t *handler = workers;
client_t *clients = NULL; client_t *clients = NULL, **last;
int count;
if (handler == NULL) if (handler == NULL)
return; return;
@ -471,8 +483,8 @@ static void worker_stop (void)
thread_sleep (10000); thread_sleep (10000);
thread_mutex_lock (&handler->lock); thread_mutex_lock (&handler->lock);
clients = handler->clients; clients = handler->clients;
handler->clients = NULL; last = handler->last_p;
handler->count = 0; count = handler->count;
thread_cond_signal (&handler->cond); thread_cond_signal (&handler->cond);
thread_mutex_unlock (&handler->lock); thread_mutex_unlock (&handler->lock);
if (clients) if (clients)
@ -481,18 +493,9 @@ static void worker_stop (void)
WARN0 ("clients left unprocessed"); WARN0 ("clients left unprocessed");
else else
{ {
// move clients to another worker
client_t *endp = clients;
int count = 0;
thread_mutex_lock (&workers->lock); thread_mutex_lock (&workers->lock);
while (endp->next_on_worker) *workers->last_p = clients;
{ workers->last_p = last;
endp = endp->next_on_worker;
count++;
}
endp->next_on_worker = workers->clients;
workers->clients = clients;
workers->count += count; workers->count += count;
thread_mutex_unlock (&workers->lock); thread_mutex_unlock (&workers->lock);
} }

10
src/client.h
View File

@ -35,7 +35,7 @@ struct _worker_t
mutex_t lock; mutex_t lock;
cond_t cond; cond_t cond;
client_t *clients; client_t *clients;
client_t **current_p; client_t **current_p, **last_p;
thread_type *thread; thread_type *thread;
struct timespec current_time; struct timespec current_time;
uint64_t time_ms; uint64_t time_ms;
@ -69,8 +69,9 @@ struct _client_tag
client_t *next_on_worker; client_t *next_on_worker;
/* the client's connection */ /* the clients connection */
connection_t *con; connection_t connection;
/* the client's http headers */ /* the client's http headers */
http_parser_t *parser; http_parser_t *parser;
@ -116,7 +117,7 @@ struct _client_tag
client_t *next; /* for use with grouping similar clients */ client_t *next; /* for use with grouping similar clients */
}; };
client_t *client_create (connection_t *con, http_parser_t *parser); client_t *client_create (sock_t sock);
void client_destroy(client_t *client); void client_destroy(client_t *client);
void client_send_504(client_t *client, char *message); void client_send_504(client_t *client, char *message);
void client_send_416(client_t *client); void client_send_416(client_t *client);
@ -142,6 +143,7 @@ void workers_adjust (int new_count);
#define CLIENT_IS_SLAVE (004) #define CLIENT_IS_SLAVE (004)
#define CLIENT_HAS_CHANGED_THREAD (010) #define CLIENT_HAS_CHANGED_THREAD (010)
#define CLIENT_NO_CONTENT_LENGTH (020) #define CLIENT_NO_CONTENT_LENGTH (020)
#define CLIENT_HAS_INTRO_CONTENT (040)
#define CLIENT_FORMAT_BIT (01000) #define CLIENT_FORMAT_BIT (01000)
#endif /* __CLIENT_H__ */ #endif /* __CLIENT_H__ */

184
src/connection.c
View File

@ -31,6 +31,7 @@
#ifndef _WIN32 #ifndef _WIN32
#include <sys/socket.h> #include <sys/socket.h>
#include <netinet/in.h> #include <netinet/in.h>
#include <netdb.h>
#endif #endif
#include "compat.h" #include "compat.h"
@ -96,7 +97,7 @@ typedef struct
static time_t now; static time_t now;
static spin_t _connection_lock; static spin_t _connection_lock;
static volatile unsigned long _current_id = 0; static volatile unsigned long _current_id = 0;
static volatile thread_type *conn_tid; thread_type *conn_tid;
static int ssl_ok; static int ssl_ok;
#ifdef HAVE_OPENSSL #ifdef HAVE_OPENSSL
@ -249,7 +250,7 @@ static void get_ssl_certificate (ice_config_t *config)
/* handlers for reading and writing a connection_t when there is ssl /* handlers for reading and writing a connection_t when there is ssl
* configured on the listening port * configured on the listening port
*/ */
static int connection_read_ssl (connection_t *con, void *buf, size_t len) int connection_read_ssl (connection_t *con, void *buf, size_t len)
{ {
int bytes = SSL_read (con->ssl, buf, len); int bytes = SSL_read (con->ssl, buf, len);
@ -266,7 +267,7 @@ static int connection_read_ssl (connection_t *con, void *buf, size_t len)
return bytes; return bytes;
} }
static int connection_send_ssl (connection_t *con, const void *buf, size_t len) int connection_send_ssl (connection_t *con, const void *buf, size_t len)
{ {
int bytes = SSL_write (con->ssl, buf, len); int bytes = SSL_write (con->ssl, buf, len);
@ -298,7 +299,7 @@ static void get_ssl_certificate (ice_config_t *config)
/* handlers (default) for reading and writing a connection_t, no encrpytion /* handlers (default) for reading and writing a connection_t, no encrpytion
* used just straight access to the socket * used just straight access to the socket
*/ */
static int connection_read (connection_t *con, void *buf, size_t len) int connection_read (connection_t *con, void *buf, size_t len)
{ {
int bytes = sock_read_bytes (con->sock, buf, len); int bytes = sock_read_bytes (con->sock, buf, len);
if (bytes == 0) if (bytes == 0)
@ -308,7 +309,7 @@ static int connection_read (connection_t *con, void *buf, size_t len)
return bytes; return bytes;
} }
static int connection_send (connection_t *con, const void *buf, size_t len) int connection_send (connection_t *con, const void *buf, size_t len)
{ {
int bytes = sock_write_bytes (con->sock, buf, len); int bytes = sock_write_bytes (con->sock, buf, len);
if (bytes < 0) if (bytes < 0)
@ -416,30 +417,49 @@ static int accept_ip_address (char *ip)
} }
connection_t *connection_create (sock_t sock, sock_t serversock, char *ip) int connection_init (connection_t *con, sock_t sock)
{ {
connection_t *con;
con = (connection_t *)calloc(1, sizeof(connection_t));
if (con) if (con)
{ {
con->sock = sock; struct sockaddr_storage sa;
con->serversock = serversock; socklen_t slen = sizeof (sa);
con->read = connection_read;
con->send = connection_send;
con->con_time = time(NULL); con->con_time = time(NULL);
con->id = _next_connection_id(); con->id = _next_connection_id();
con->ip = ip; con->discon_time = con->con_time + header_timeout;
con->sock = sock;
if (sock == SOCK_ERROR)
return 0;
if (getpeername (sock, (struct sockaddr *)&sa, &slen) == 0)
{
char *ip;
#ifdef HAVE_GETNAMEINFO
char buffer [200] = "unknown";
getnameinfo ((struct sockaddr *)&sa, slen, buffer, 200, NULL, 0, NI_NUMERICHOST);
ip = strdup (buffer);
#else
int len = 30;
ip = malloc (len);
strncpy (ip, inet_ntoa (sa.sin_addr), len);
#endif
if (accept_ip_address (ip))
{
con->ip = ip;
return 0;
}
free (ip);
}
memset (con, 0, sizeof (connection_t));
} }
return con; return -1;
} }
/* prepare connection for interacting over a SSL connection /* prepare connection for interacting over a SSL connection
*/ */
void connection_uses_ssl (connection_t *con) void connection_uses_ssl (connection_t *con)
{ {
#ifdef HAVE_OPENSSL #ifdef HAVE_OPENSSL
con->read = connection_read_ssl;
con->send = connection_send_ssl;
con->ssl = SSL_new (ssl_ctx); con->ssl = SSL_new (ssl_ctx);
SSL_set_accept_state (con->ssl); SSL_set_accept_state (con->ssl);
SSL_set_fd (con->ssl, con->sock); SSL_set_fd (con->ssl, con->sock);
@ -529,42 +549,57 @@ static sock_t wait_for_serversock(int timeout)
#endif #endif
} }
static connection_t *_accept_connection(int duration)
{
sock_t sock, serversock;
char *ip;
serversock = wait_for_serversock (duration); static client_t *accept_client (int duration)
{
client_t *client;
sock_t sock, serversock = wait_for_serversock (duration);
if (serversock == SOCK_ERROR) if (serversock == SOCK_ERROR)
return NULL; return NULL;
now = time(NULL); sock = sock_accept (serversock, NULL, 0);
/* malloc enough room for a full IP address (including ipv6) */ if (sock == SOCK_ERROR)
ip = (char *)malloc(MAX_ADDR_LEN);
sock = sock_accept(serversock, ip, MAX_ADDR_LEN);
if (sock != SOCK_ERROR)
{ {
connection_t *con = NULL; if (sock_recoverable (sock_error()))
/* Make any IPv4 mapped IPv6 address look like a normal IPv4 address */ return NULL;
if (strncmp (ip, "::ffff:", 7) == 0) WARN2 ("accept() failed with error %d: %s", sock_error(), strerror(sock_error()));
memmove (ip, ip+7, strlen (ip+7)+1); thread_sleep (500000);
return NULL;
if (accept_ip_address (ip))
con = connection_create (sock, serversock, ip);
if (con)
return con;
sock_close (sock);
} }
else global_lock ();
client = client_create (sock);
if (client)
{ {
if (!sock_recoverable(sock_error())) connection_t *con = &client->connection;
int i;
for (i=0; i < global.server_sockets; i++)
{ {
WARN2("accept() failed with error %d: %s", sock_error(), strerror(sock_error())); if (global.serversock[i] == serversock)
thread_sleep (500000); {
client->server_conn = global.server_conn[i];
client->server_conn->refcount++;
if (client->server_conn->ssl && ssl_ok)
connection_uses_ssl (con);
if (client->server_conn->shoutcast_compat)
client->ops = &shoutcast_source_ops;
else
client->ops = &http_request_ops;
break;
}
} }
global_unlock ();
stats_event_inc (NULL, "connections");
if (sock_set_blocking (con->sock, 0) || sock_set_nodelay (con->sock))
{
WARN0 ("failed to set tcp options on client connection, dropping");
client_destroy (client);
client = NULL;
}
return client;
} }
free (ip); global_unlock ();
sock_close (sock);
return NULL; return NULL;
} }
@ -577,7 +612,8 @@ static int shoutcast_source_client (client_t *client)
{ {
do do
{ {
if (client->con->error || client->con->discon_time <= client->worker->current_time.tv_sec) connection_t *con = &client->connection;
if (con->error || con->discon_time <= client->worker->current_time.tv_sec)
break; break;
if (client->shared_data) /* need to get password first */ if (client->shared_data) /* need to get password first */
@ -593,7 +629,7 @@ static int shoutcast_source_client (client_t *client)
break; break;
ret = client_read_bytes (client, buf, remaining); ret = client_read_bytes (client, buf, remaining);
if (ret == 0 || client->con->error) if (ret == 0 || con->error)
break; break;
if (ret < 0) if (ret < 0)
return 0; return 0;
@ -650,7 +686,7 @@ static int http_client_request (client_t *client)
refbuf_t *refbuf = client->shared_data; refbuf_t *refbuf = client->shared_data;
int remaining = PER_CLIENT_REFBUF_SIZE - 1 - refbuf->len, ret = -1; int remaining = PER_CLIENT_REFBUF_SIZE - 1 - refbuf->len, ret = -1;
if (remaining && client->con->discon_time > client->worker->current_time.tv_sec) if (remaining && client->connection.discon_time > client->worker->current_time.tv_sec)
{ {
char *buf = refbuf->data + refbuf->len; char *buf = refbuf->data + refbuf->len;
@ -701,7 +737,7 @@ static int http_client_request (client_t *client)
} while (0); } while (0);
client->refbuf = client->shared_data; client->refbuf = client->shared_data;
client->shared_data = NULL; client->shared_data = NULL;
client->con->discon_time = 0; client->connection.discon_time = 0;
client->parser = httpp_create_parser(); client->parser = httpp_create_parser();
httpp_initialize (client->parser, NULL); httpp_initialize (client->parser, NULL);
if (httpp_parse (client->parser, refbuf->data, refbuf->len)) if (httpp_parse (client->parser, refbuf->data, refbuf->len))
@ -751,7 +787,7 @@ static int http_client_request (client_t *client)
/* invalid http request */ /* invalid http request */
return -1; return -1;
} }
if (ret && client->con->error == 0) if (ret && client->connection.error == 0)
{ {
client->schedule_ms = client->worker->time_ms + 100; client->schedule_ms = client->worker->time_ms + 100;
return 0; return 0;
@ -763,9 +799,8 @@ static int http_client_request (client_t *client)
} }
void *connection_thread (void *arg) static void *connection_thread (void *arg)
{ {
connection_t *con;
ice_config_t *config; ice_config_t *config;
connection_running = 1; connection_running = 1;
@ -780,37 +815,9 @@ void *connection_thread (void *arg)
while (connection_running) while (connection_running)
{ {
con = _accept_connection (333); client_t *client = accept_client (333);
if (client)
if (con)
{ {
client_t *client = NULL;
refbuf_t *r;
global_lock();
client = client_create (con, NULL);
global_unlock();
if (client->server_conn->ssl && ssl_ok)
connection_uses_ssl (client->con);
if (sock_set_blocking (client->con->sock, 0) || sock_set_nodelay (client->con->sock))
{
WARN0 ("failed to set tcp options on client connection, dropping");
client_destroy (client);
continue;
}
if (client->server_conn->shoutcast_compat)
client->ops = &shoutcast_source_ops;
else
client->ops = &http_request_ops;
r = refbuf_new (PER_CLIENT_REFBUF_SIZE);
r->len = 0;
client->shared_data = r;
client->flags |= CLIENT_ACTIVE;
client->con->discon_time = time(NULL) + header_timeout;
/* do a small delay here so the client has chance to send the request after /* do a small delay here so the client has chance to send the request after
* getting a connect. This also prevents excessively large number of new * getting a connect. This also prevents excessively large number of new
* listeners from joining at the same time */ * listeners from joining at the same time */
@ -840,11 +847,10 @@ void connection_thread_shutdown ()
{ {
if (conn_tid) if (conn_tid)
{ {
thread_type *tid = (thread_type*)conn_tid;;
conn_tid = NULL;
INFO0("shutting down connection thread");
connection_running = 0; connection_running = 0;
thread_join (tid); INFO0("shutting down connection thread");
thread_join (conn_tid);
conn_tid = NULL;
} }
} }
@ -1315,12 +1321,14 @@ int connection_setup_sockets (ice_config_t *config)
void connection_close(connection_t *con) void connection_close(connection_t *con)
{ {
sock_close(con->sock); if (con->con_time)
if (con->ip) free(con->ip); {
if (con->host) free(con->host); sock_close(con->sock);
free(con->ip);
#ifdef HAVE_OPENSSL #ifdef HAVE_OPENSSL
if (con->ssl) { SSL_shutdown (con->ssl); SSL_free (con->ssl); } if (con->ssl) { SSL_shutdown (con->ssl); SSL_free (con->ssl); }
#endif #endif
free(con); memset (con, 0, sizeof (connection_t));
}
} }

14
src/connection.h
View File

@ -21,9 +21,9 @@
#endif #endif
struct source_tag; struct source_tag;
struct ice_config_tag;
typedef struct connection_tag connection_t; typedef struct connection_tag connection_t;
#include "cfgfile.h"
#include "compat.h" #include "compat.h"
#include "httpp/httpp.h" #include "httpp/httpp.h"
#include "net/sock.h" #include "net/sock.h"
@ -37,17 +37,13 @@ struct connection_tag
uint64_t sent_bytes; uint64_t sent_bytes;
sock_t sock; sock_t sock;
sock_t serversock;
int error; int error;
#ifdef HAVE_OPENSSL #ifdef HAVE_OPENSSL
SSL *ssl; /* SSL handler */ SSL *ssl; /* SSL handler */
#endif #endif
int (*send)(struct connection_tag *handle, const void *buf, size_t len);
int (*read)(struct connection_tag *handle, void *buf, size_t len);
char *ip; char *ip;
char *host;
}; };
#ifdef HAVE_OPENSSL #ifdef HAVE_OPENSSL
@ -61,9 +57,15 @@ void connection_thread_startup();
void connection_thread_shutdown(); void connection_thread_shutdown();
int connection_setup_sockets (struct ice_config_tag *config); int connection_setup_sockets (struct ice_config_tag *config);
void connection_close(connection_t *con); void connection_close(connection_t *con);
connection_t *connection_create (sock_t sock, sock_t serversock, char *ip); int connection_init (connection_t *con, sock_t sock);
int connection_complete_source (struct source_tag *source, int response); int connection_complete_source (struct source_tag *source, int response);
void connection_uses_ssl (connection_t *con); void connection_uses_ssl (connection_t *con);
#ifdef HAVE_OPENSSL
int connection_read_ssl (connection_t *con, void *buf, size_t len);
int connection_send_ssl (connection_t *con, const void *buf, size_t len);
#endif
int connection_read (connection_t *con, void *buf, size_t len);
int connection_send (connection_t *con, const void *buf, size_t len);
void connection_thread_shutdown_req (void); void connection_thread_shutdown_req (void);
int connection_check_pass (http_parser_t *parser, const char *user, const char *pass); int connection_check_pass (http_parser_t *parser, const char *user, const char *pass);

52
src/format.c
View File

@ -96,12 +96,29 @@ int format_file_read (client_t *client, FILE *intro)
{ {
refbuf_t *refbuf = client->refbuf; refbuf_t *refbuf = client->refbuf;
if (intro == NULL) if (refbuf == NULL)
return -1; return -1;
if (client->pos == refbuf->len) if (client->pos == refbuf->len)
{ {
size_t bytes; size_t bytes;
if (client->flags & CLIENT_HAS_INTRO_CONTENT)
{
if (refbuf->next)
{
client->refbuf = refbuf->next;
refbuf->next = NULL;
refbuf_release (refbuf);
client->pos = 0;
return 0;
}
client_set_queue (client, NULL);
client->flags &= ~CLIENT_HAS_INTRO_CONTENT;
client->intro_offset = client->connection.sent_bytes;
return -1;
}
if (intro == NULL)
return -1;
if (fseek (intro, client->intro_offset, SEEK_SET) < 0) if (fseek (intro, client->intro_offset, SEEK_SET) < 0)
return -1; return -1;
bytes = fread (refbuf->data, 1, PER_CLIENT_REFBUF_SIZE, intro); bytes = fread (refbuf->data, 1, PER_CLIENT_REFBUF_SIZE, intro);
@ -123,7 +140,7 @@ int format_generic_write_to_client (client_t *client)
const char *buf = refbuf->data + client->pos; const char *buf = refbuf->data + client->pos;
unsigned int len = refbuf->len - client->pos; unsigned int len = refbuf->len - client->pos;
if (len > 4096) /* make sure we don't send huge amounts in one go */ if (len > 5000) /* make sure we don't send huge amounts in one go */
len = 4096; len = 4096;
ret = client_send_bytes (client, buf, len); ret = client_send_bytes (client, buf, len);
@ -134,25 +151,23 @@ int format_generic_write_to_client (client_t *client)
} }
int format_prepare_headers (source_t *source, client_t *client) int format_general_headers (source_t *source, client_t *client)
{ {
unsigned remaining; unsigned remaining = 4096 - client->refbuf->len;
char *ptr; char *ptr = client->refbuf->data + client->refbuf->len;
int bytes; int bytes;
int bitrate_filtered = 0; int bitrate_filtered = 0;
avl_node *node; avl_node *node;
ice_config_t *config; ice_config_t *config;
DEBUG0 ("processing listener headers"); if (client->respcode == 0)
remaining = client->refbuf->len; {
ptr = client->refbuf->data; bytes = snprintf (ptr, remaining, "HTTP/1.0 200 OK\r\n"
client->respcode = 200; "Content-Type: %s\r\n", source->format->contenttype);
remaining -= bytes;
bytes = snprintf (ptr, remaining, "HTTP/1.0 200 OK\r\n" ptr += bytes;
"Content-Type: %s\r\n", source->format->contenttype); client->respcode = 200;
}
remaining -= bytes;
ptr += bytes;
/* iterate through source http headers and send to client */ /* iterate through source http headers and send to client */
avl_tree_rlock (source->parser->vars); avl_tree_rlock (source->parser->vars);
@ -227,10 +242,7 @@ int format_prepare_headers (source_t *source, client_t *client)
remaining -= bytes; remaining -= bytes;
ptr += bytes; ptr += bytes;
client->refbuf->len -= remaining; client->refbuf->len = 4096 - remaining;
if (source->format->create_client_data) return 0;
if (source->format->create_client_data (source, client) < 0)
bytes = -1;
return bytes;
} }

2
src/format.h
View File

@ -64,7 +64,7 @@ int format_get_plugin(format_type_t type, struct source_tag *source);
int format_generic_write_to_client (client_t *client); int format_generic_write_to_client (client_t *client);
int format_file_read (client_t *client, FILE *fp); int format_file_read (client_t *client, FILE *fp);
int format_prepare_headers (struct source_tag *source, client_t *client); int format_general_headers (struct source_tag *source, client_t *client);
void format_send_general_headers(format_plugin_t *format, void format_send_general_headers(format_plugin_t *format,
struct source_tag *source, client_t *client); struct source_tag *source, client_t *client);

24
src/format_mp3.c
View File

@ -382,14 +382,14 @@ static int send_stream_metadata (client_t *client, refbuf_t *refbuf, unsigned in
{ {
client_mp3->metadata_offset += (ret - remaining); client_mp3->metadata_offset += (ret - remaining);
client->flags |= CLIENT_IN_METADATA; client->flags |= CLIENT_IN_METADATA;
client->schedule_ms += 100; client->schedule_ms += 300;
} }
client_mp3->since_meta_block = 0; client_mp3->since_meta_block = 0;
client->pos += remaining; client->pos += remaining;
client->queue_pos += remaining; client->queue_pos += remaining;
return ret; return ret;
} }
client->schedule_ms += 100; client->schedule_ms += 300;
if (ret > 0) if (ret > 0)
{ {
client_mp3->since_meta_block += ret; client_mp3->since_meta_block += ret;
@ -409,7 +409,7 @@ static int send_stream_metadata (client_t *client, refbuf_t *refbuf, unsigned in
} }
if (ret > 0) if (ret > 0)
client_mp3->metadata_offset += ret; client_mp3->metadata_offset += ret;
client->schedule_ms += 100; client->schedule_ms += 300;
client->flags |= CLIENT_IN_METADATA; client->flags |= CLIENT_IN_METADATA;
return ret > 0 ? ret : 0; return ret > 0 ? ret : 0;
@ -476,6 +476,8 @@ static int format_mp3_write_buf_to_client (client_t *client)
ret = 0; ret = 0;
} while (0); } while (0);
if (ret < 0)
client->schedule_ms += 250;
if (ret > 0) if (ret > 0)
written += ret; written += ret;
return written == 0 ? -1 : written; return written == 0 ? -1 : written;
@ -521,8 +523,6 @@ static int complete_read (source_t *source)
{ {
int read_in = source_mp3->queue_block_size - source_mp3->read_count; int read_in = source_mp3->queue_block_size - source_mp3->read_count;
bytes = client_read_bytes (client, buf, read_in); bytes = client_read_bytes (client, buf, read_in);
if (bytes < read_in)
client->schedule_ms = client->worker->time_ms + source->skip_duration;
if (bytes < 0) if (bytes < 0)
return 0; return 0;
rate_add (format->in_bitrate, bytes, client->worker->current_time.tv_sec); rate_add (format->in_bitrate, bytes, client->worker->current_time.tv_sec);
@ -695,9 +695,8 @@ static int format_mp3_create_client_data(source_t *source, client_t *client)
mp3_client_data *client_mp3 = calloc(1,sizeof(mp3_client_data)); mp3_client_data *client_mp3 = calloc(1,sizeof(mp3_client_data));
mp3_state *source_mp3 = source->format->_state; mp3_state *source_mp3 = source->format->_state;
const char *metadata; const char *metadata;
/* the +-2 is for overwriting the last set of \r\n */ size_t remaining = 4096;
size_t remaining = 4096 - client->refbuf->len + 2; char *ptr = client->refbuf->data;
char *ptr = client->refbuf->data + client->refbuf->len - 2;
int bytes; int bytes;
const char *useragent; const char *useragent;
@ -721,6 +720,15 @@ static int format_mp3_create_client_data(source_t *source, client_t *client)
client->format_data = client_mp3; client->format_data = client_mp3;
client->free_client_data = free_mp3_client_data; client->free_client_data = free_mp3_client_data;
client->refbuf->len = 4096 - remaining;
if (format_general_headers (source, client) < 0)
return -1;
remaining = 4096 - client->refbuf->len + 2;
ptr = client->refbuf->data + client->refbuf->len - 2;
/* check for shoutcast style metadata inserts */
metadata = httpp_getvar(client->parser, "icy-metadata"); metadata = httpp_getvar(client->parser, "icy-metadata");
if (metadata && atoi(metadata)) if (metadata && atoi(metadata))
{ {

11
src/format_ogg.c
View File

@ -462,8 +462,6 @@ static refbuf_t *ogg_get_buffer (source_t *source)
data = ogg_sync_buffer (&ogg_info->oy, 4096); data = ogg_sync_buffer (&ogg_info->oy, 4096);
bytes = client_read_bytes (source->client, data, 4096); bytes = client_read_bytes (source->client, data, 4096);
if (bytes < 4096)
source->client->schedule_ms = source->client->worker->time_ms + source->skip_duration;
if (bytes <= 0) if (bytes <= 0)
{ {
ogg_sync_wrote (&ogg_info->oy, 0); ogg_sync_wrote (&ogg_info->oy, 0);
@ -486,7 +484,7 @@ static int create_ogg_client_data (source_t *source, client_t *client)
client_data->headers_sent = 1; client_data->headers_sent = 1;
client->format_data = client_data; client->format_data = client_data;
client->free_client_data = free_ogg_client_data; client->free_client_data = free_ogg_client_data;
ret = 0; ret = format_general_headers (source, client);
} }
return ret; return ret;
} }
@ -575,8 +573,11 @@ static int write_buf_to_client (client_t *client)
ret = 0; ret = 0;
} while (0); } while (0);
if (ret > 0) if (ret > 0) /* short write */
written += ret; {
client->schedule_ms += 250;
written += ret;
}
return written; return written;
} }

47
src/fserve.c
View File

@ -109,11 +109,11 @@ void fserve_shutdown(void)
avl_tree_free (mimetypes, _delete_mapping); avl_tree_free (mimetypes, _delete_mapping);
if (fh_cache) if (fh_cache)
{ {
int count = 100; int count = 20;
while (fh_cache->length && count) while (fh_cache->length && count)
{ {
DEBUG1 ("waiting for %u entries to clear", fh_cache->length); DEBUG1 ("waiting for %u entries to clear", fh_cache->length);
thread_sleep (20000); thread_sleep (100000);
count--; count--;
} }
avl_tree_free (fh_cache, _delete_fh); avl_tree_free (fh_cache, _delete_fh);
@ -189,7 +189,11 @@ static int _delete_fh (void *mapping)
fh_node *fh = mapping; fh_node *fh = mapping;
if (fh->refcount) if (fh->refcount)
WARN2 ("handle for %s has refcount %d", fh->finfo.mount, fh->refcount); WARN2 ("handle for %s has refcount %d", fh->finfo.mount, fh->refcount);
thread_mutex_destroy (&fh->lock); else
{
thread_mutex_unlock (&fh->lock);
thread_mutex_destroy (&fh->lock);
}
if (fh->fp) if (fh->fp)
fclose (fh->fp); fclose (fh->fp);
free (fh->finfo.mount); free (fh->finfo.mount);
@ -571,13 +575,13 @@ static void fserve_move_listener (client_t *client)
static int prefile_send (client_t *client) static int prefile_send (client_t *client)
{ {
refbuf_t *refbuf = client->refbuf; refbuf_t *refbuf = client->refbuf;
int loop = 3, bytes; int loop = 6, bytes, written = 0;
while (loop) while (loop)
{ {
fh_node *fh = client->shared_data; fh_node *fh = client->shared_data;
loop--; loop--;
if (fserve_running == 0 || client->con->error) if (fserve_running == 0 || client->connection.error)
return -1; return -1;
if (refbuf == NULL || client->pos == refbuf->len) if (refbuf == NULL || client->pos == refbuf->len)
{ {
@ -620,14 +624,15 @@ static int prefile_send (client_t *client)
bytes = format_generic_write_to_client (client); bytes = format_generic_write_to_client (client);
if (bytes < 0) if (bytes < 0)
{ {
client->schedule_ms = client->worker->time_ms + 150; client->schedule_ms = client->worker->time_ms + 300;
return 0; return 0;
} }
written += bytes;
global_add_bitrates (global.out_bitrate, bytes, client->worker->time_ms); global_add_bitrates (global.out_bitrate, bytes, client->worker->time_ms);
if (bytes < 4096) if (written > 30000)
break; break;
} }
client->schedule_ms = client->worker->time_ms + (loop ? 50 : 15); client->schedule_ms = client->worker->time_ms + 150;
return 0; return 0;
} }
@ -635,15 +640,16 @@ static int prefile_send (client_t *client)
static int file_send (client_t *client) static int file_send (client_t *client)
{ {
refbuf_t *refbuf = client->refbuf; refbuf_t *refbuf = client->refbuf;
int loop = 5, bytes; int loop = 6, bytes, written = 0;
fh_node *fh = client->shared_data; fh_node *fh = client->shared_data;
if (client->con->discon_time && client->worker->current_time.tv_sec >= client->con->discon_time) if (client->connection.discon_time &&
client->worker->current_time.tv_sec >= client->connection.discon_time)
return -1; return -1;
while (loop) while (loop)
{ {
loop--; loop--;
if (fserve_running == 0 || client->con->error) if (fserve_running == 0 || client->connection.error)
return -1; return -1;
if (fh->finfo.limit) if (fh->finfo.limit)
{ {
@ -681,12 +687,18 @@ static int file_send (client_t *client)
} }
bytes = client->check_buffer (client); bytes = client->check_buffer (client);
if (bytes < 0) if (bytes < 0)
{
client->schedule_ms = client->worker->time_ms + 300;
return 0; return 0;
}
written += bytes;
global_add_bitrates (global.out_bitrate, bytes, client->worker->time_ms); global_add_bitrates (global.out_bitrate, bytes, client->worker->time_ms);
if (fh->finfo.limit) if (fh->finfo.limit)
rate_add (client->out_bitrate, bytes, client->worker->time_ms); rate_add (client->out_bitrate, bytes, client->worker->time_ms);
if (written > 30000)
break;
} }
client->schedule_ms = client->worker->time_ms + 10; client->schedule_ms = client->worker->time_ms + 150;
return 1; return 1;
} }
@ -709,8 +721,17 @@ void fserve_setup_client_fb (client_t *client, fbinfo *finfo)
{ {
client->check_buffer = format_generic_write_to_client; client->check_buffer = format_generic_write_to_client;
} }
client->flags &= ~CLIENT_HAS_INTRO_CONTENT;
client->intro_offset = 0; client->intro_offset = 0;
client->flags |= CLIENT_ACTIVE; if (client->flags & CLIENT_ACTIVE)
client->schedule_ms = client->worker->time_ms;
else
{
client->flags |= CLIENT_ACTIVE;
thread_mutex_lock (&client->worker->lock);
thread_cond_signal (&client->worker->cond);
thread_mutex_unlock (&client->worker->lock);
}
} }

2
src/global.c
View File

@ -78,7 +78,7 @@ void global_add_bitrates (struct rate_calc *rate, unsigned long value, uint64_t
void global_reduce_bitrate_sampling (struct rate_calc *rate) void global_reduce_bitrate_sampling (struct rate_calc *rate)
{ {
thread_spin_lock (&global.spinlock); thread_spin_lock (&global.spinlock);
rate_reduce (rate, 0); rate_reduce (rate, 2);
thread_spin_unlock (&global.spinlock); thread_spin_unlock (&global.spinlock);
} }

6
src/logging.c
View File

@ -134,7 +134,7 @@ void logging_access(client_t *client)
httpp_getvar (client->parser, HTTPP_VAR_PROTOCOL), httpp_getvar (client->parser, HTTPP_VAR_PROTOCOL),
httpp_getvar (client->parser, HTTPP_VAR_VERSION)); httpp_getvar (client->parser, HTTPP_VAR_VERSION));
stayed = now - client->con->con_time; stayed = now - client->connection.con_time;
if (client->username == NULL) if (client->username == NULL)
username = "-"; username = "-";
@ -151,12 +151,12 @@ void logging_access(client_t *client)
config = config_get_config(); config = config_get_config();
if (config->access_log.log_ip) if (config->access_log.log_ip)
ip = client->con->ip; ip = client->connection.ip;
config_release_config (); config_release_config ();
log_write_direct (accesslog, log_write_direct (accesslog,
"%s - %s [%s] \"%s\" %d %" PRIu64 " \"%s\" \"%s\" %lu", "%s - %s [%s] \"%s\" %d %" PRIu64 " \"%s\" \"%s\" %lu",
ip, username, ip, username,
datebuf, reqbuf, client->respcode, client->con->sent_bytes, datebuf, reqbuf, client->respcode, client->connection.sent_bytes,
referrer, user_agent, (unsigned long)stayed); referrer, user_agent, (unsigned long)stayed);
} }

24
src/main.c
View File

@ -14,16 +14,18 @@
#ifdef HAVE_CONFIG_H #ifdef HAVE_CONFIG_H
#include <config.h> #include <config.h>
#endif #endif
#ifdef WIN32_SERVICE
#define _WIN32_WINNT 0x0400
#include <windows.h>
#endif
#include <stdio.h> #include <stdio.h>
#include <string.h> #include <string.h>
#include <errno.h> #include <errno.h>
#ifdef WIN32
#define _WIN32_WINNT 0x0400
/* For getpid() */
#include <process.h>
#include <windows.h>
#endif
#ifdef HAVE_UNISTD_H #ifdef HAVE_UNISTD_H
# include <unistd.h> # include <unistd.h>
#endif #endif
@ -61,11 +63,6 @@
#include <libxml/xmlmemory.h> #include <libxml/xmlmemory.h>
#ifdef _WIN32
/* For getpid() */
#include <process.h>
#endif
#undef CATMODULE #undef CATMODULE
#define CATMODULE "main" #define CATMODULE "main"
@ -113,7 +110,6 @@ void _initialize_subsystems(void)
refbuf_initialize(); refbuf_initialize();
stats_initialize(); stats_initialize();
fserve_initialize();
xslt_initialize(); xslt_initialize();
#ifdef HAVE_CURL_GLOBAL_INIT #ifdef HAVE_CURL_GLOBAL_INIT
curl_global_init (CURL_GLOBAL_ALL); curl_global_init (CURL_GLOBAL_ALL);
@ -122,15 +118,15 @@ void _initialize_subsystems(void)
void _shutdown_subsystems(void) void _shutdown_subsystems(void)
{ {
refbuf_shutdown(); fserve_shutdown();
slave_shutdown(); slave_shutdown();
auth_shutdown(); auth_shutdown();
yp_shutdown(); yp_shutdown();
stats_shutdown(); stats_shutdown();
fserve_shutdown();
connection_shutdown(); connection_shutdown();
config_shutdown(); config_shutdown();
refbuf_shutdown();
resolver_shutdown(); resolver_shutdown();
sock_shutdown(); sock_shutdown();
@ -316,7 +312,6 @@ static void _server_proc(void)
} }
slave_initialize(); slave_initialize();
connection_thread_shutdown();
connection_setup_sockets (NULL); connection_setup_sockets (NULL);
} }
@ -459,6 +454,7 @@ int main(int argc, char **argv)
} }
_ch_root_uid_setup(); /* Change user id and root if requested/possible */ _ch_root_uid_setup(); /* Change user id and root if requested/possible */
fserve_initialize();
#ifdef CHUID #ifdef CHUID
/* We'll only have getuid() if we also have setuid(), it's reasonable to /* We'll only have getuid() if we also have setuid(), it's reasonable to

23
src/slave.c
View File

@ -297,7 +297,7 @@ static int open_relay_connection (client_t *client, relay_server *relay, relay_s
char *server_id = NULL; char *server_id = NULL;
ice_config_t *config; ice_config_t *config;
http_parser_t *parser = NULL; http_parser_t *parser = NULL;
connection_t *con=NULL; connection_t *con = &client->connection;
char *server = strdup (master->ip); char *server = strdup (master->ip);
char *mount = strdup (master->mount); char *mount = strdup (master->mount);
int port = master->port; int port = master->port;
@ -346,7 +346,7 @@ static int open_relay_connection (client_t *client, relay_server *relay, relay_s
WARN3 ("Failed to connect to %s:%d for %s", server, port, relay->localmount); WARN3 ("Failed to connect to %s:%d for %s", server, port, relay->localmount);
break; break;
} }
con = connection_create (streamsock, SOCK_ERROR, strdup (server)); connection_init (con, streamsock);
/* At this point we may not know if we are relaying an mp3 or vorbis /* At this point we may not know if we are relaying an mp3 or vorbis
* stream, but only send the icy-metadata header if the relay details * stream, but only send the icy-metadata header if the relay details
@ -405,7 +405,6 @@ static int open_relay_connection (client_t *client, relay_server *relay, relay_s
strncpy (server, uri, len); strncpy (server, uri, len);
connection_close (con); connection_close (con);
httpp_destroy (parser); httpp_destroy (parser);
con = NULL;
parser = NULL; parser = NULL;
} }
else else
@ -417,7 +416,6 @@ static int open_relay_connection (client_t *client, relay_server *relay, relay_s
break; break;
} }
sock_set_blocking (streamsock, 0); sock_set_blocking (streamsock, 0);
client->con = con;
client->parser = parser; client->parser = parser;
client_set_queue (client, NULL); client_set_queue (client, NULL);
free (server); free (server);
@ -434,8 +432,7 @@ static int open_relay_connection (client_t *client, relay_server *relay, relay_s
free (mount); free (mount);
free (server_id); free (server_id);
free (auth_header); free (auth_header);
if (con) connection_close (con);
connection_close (con);
if (parser) if (parser)
httpp_destroy (parser); httpp_destroy (parser);
return -1; return -1;
@ -495,9 +492,7 @@ static void *start_relay_stream (void *arg)
else else
yp_remove (relay->localmount); yp_remove (relay->localmount);
if (client->con) connection_close (&client->connection);
connection_close (client->con);
client->con = NULL;
if (client->parser) if (client->parser)
httpp_destroy (client->parser); httpp_destroy (client->parser);
client->parser = NULL; client->parser = NULL;
@ -551,6 +546,7 @@ static void *start_relay_stream (void *arg)
INFO2 ("listener count still on %s is %d", src->mount, src->listeners); INFO2 ("listener count still on %s is %d", src->mount, src->listeners);
source_clear_listeners (src); source_clear_listeners (src);
source_clear_source (src); source_clear_source (src);
thread_mutex_unlock (&src->lock);
relay->start = client->worker->current_time.tv_sec + relay->interval; relay->start = client->worker->current_time.tv_sec + relay->interval;
client->schedule_ms = timing_get_time() + 1000; client->schedule_ms = timing_get_time() + 1000;
client->flags |= CLIENT_ACTIVE; client->flags |= CLIENT_ACTIVE;
@ -595,7 +591,7 @@ static void check_relay_stream (relay_server *relay)
{ {
client_t *client; client_t *client;
global_lock(); global_lock();
client = client_create (NULL, NULL); client = client_create (SOCK_ERROR);
global_unlock(); global_unlock();
source->client = client; source->client = client;
client->shared_data = relay; client->shared_data = relay;
@ -1254,13 +1250,12 @@ static int relay_read (client_t *client)
client->ops = &relay_startup_ops; client->ops = &relay_startup_ops;
relay->running = 0; relay->running = 0;
global_reduce_bitrate_sampling (global.out_bitrate); global_reduce_bitrate_sampling (global.out_bitrate);
if (client->con) connection_close (&client->connection);
connection_close (client->con);
client->con = NULL;
if (client->parser) if (client->parser)
httpp_destroy (client->parser); httpp_destroy (client->parser);
client->parser = NULL; client->parser = NULL;
source_clear_source (source); source_clear_source (source);
thread_mutex_unlock (&source->lock);
thread_rwlock_unlock (&global.shutdown_lock); thread_rwlock_unlock (&global.shutdown_lock);
slave_update_all_mounts(); slave_update_all_mounts();
return 0; return 0;
@ -1299,6 +1294,8 @@ static int relay_startup (client_t *client)
if (relay->cleanup) if (relay->cleanup)
return -1; return -1;
if (global.running != ICE_RUNNING)
return 0; /* wait for cleanup */
if (relay->enable == 0 || relay->start > client->worker->current_time.tv_sec) if (relay->enable == 0 || relay->start > client->worker->current_time.tv_sec)
{ {
client->schedule_ms = client->worker->time_ms + 1000; client->schedule_ms = client->worker->time_ms + 1000;

162
src/source.c
View File

@ -61,7 +61,6 @@
/* avl tree helper */ /* avl tree helper */
static int _compare_clients(void *compare_arg, void *a, void *b);
static void _parse_audio_info (source_t *source, const char *s); static void _parse_audio_info (source_t *source, const char *s);
static void source_client_release (client_t *client); static void source_client_release (client_t *client);
static void source_listener_release (client_t *client); static void source_listener_release (client_t *client);
@ -75,6 +74,7 @@ static int http_source_intro (client_t *client);
static int locate_start_on_queue (source_t *source, client_t *client); static int locate_start_on_queue (source_t *source, client_t *client);
static void listener_change_worker (client_t *client, source_t *source); static void listener_change_worker (client_t *client, source_t *source);
static void source_change_worker (source_t *source); static void source_change_worker (source_t *source);
static int source_client_callback (client_t *client);
#ifdef _WIN32 #ifdef _WIN32
#define source_run_script(x,y) WARN0("on [dis]connect scripts disabled"); #define source_run_script(x,y) WARN0("on [dis]connect scripts disabled");
@ -247,6 +247,8 @@ void source_clear_listeners (source_t *source)
client = source->client_list; client = source->client_list;
source->client_list = client->next; source->client_list = client->next;
client->next = NULL; client->next = NULL;
client->shared_data = NULL;
client_set_queue (client, NULL);
/* do not count listeners who have joined but haven't done any processing */ /* do not count listeners who have joined but haven't done any processing */
if (client->respcode == 200) if (client->respcode == 200)
i++; i++;
@ -256,7 +258,6 @@ void source_clear_listeners (source_t *source)
if (i) if (i)
{ {
stats_event_sub (NULL, "listeners", i); stats_event_sub (NULL, "listeners", i);
stats_event_sub (source->mount, "listeners", i);
} }
source->listeners = 0; source->listeners = 0;
source->prev_listeners = 0; source->prev_listeners = 0;
@ -323,7 +324,6 @@ void source_clear_source (source_t *source)
} }
source->flags &= ~SOURCE_ON_DEMAND_REQ; source->flags &= ~SOURCE_ON_DEMAND_REQ;
thread_mutex_unlock (&source->lock);
} }
@ -342,6 +342,7 @@ static int _free_source (void *p)
if (source->client_list) if (source->client_list)
WARN1("active listeners on mountpoint %s", source->mount); WARN1("active listeners on mountpoint %s", source->mount);
thread_mutex_unlock (&source->lock);
thread_mutex_destroy (&source->lock); thread_mutex_destroy (&source->lock);
INFO1 ("freeing source \"%s\"", source->mount); INFO1 ("freeing source \"%s\"", source->mount);
@ -363,16 +364,12 @@ void source_free_source (source_t *source)
client_t *source_find_client(source_t *source, int id) client_t *source_find_client(source_t *source, int id)
{ {
client_t fakeclient, *client = NULL; client_t *client = NULL;
connection_t fakecon;
fakeclient.con = &fakecon;
fakeclient.con->id = id;
client = source->client_list; client = source->client_list;
while (client) while (client)
{ {
if (_compare_clients (NULL, client, &fakeclient) == 0) if (client->connection.id == id)
break; break;
client = client->next; client = client->next;
} }
@ -403,7 +400,7 @@ static void update_source_stats (source_t *source)
{ {
worker_t *worker = source->client->worker; worker_t *worker = source->client->worker;
stats_event_args (source->mount, "connected", "%"PRIu64, stats_event_args (source->mount, "connected", "%"PRIu64,
(uint64_t)(worker->current_time.tv_sec - source->client->con->con_time)); (uint64_t)(worker->current_time.tv_sec - source->client->connection.con_time));
} }
stats_event_add (NULL, "stream_kbytes_sent", kbytes_sent); stats_event_add (NULL, "stream_kbytes_sent", kbytes_sent);
stats_event_add (NULL, "stream_kbytes_read", kbytes_read); stats_event_add (NULL, "stream_kbytes_read", kbytes_read);
@ -430,8 +427,7 @@ void source_read (source_t *source)
source->flags &= ~SOURCE_RUNNING; source->flags &= ~SOURCE_RUNNING;
do do
{ {
if (source->flags & SOURCE_TEMPORARY_FALLBACK && source->fallback.mount && if (source->fallback.mount && source->termination_count == 0)
source->termination_count == 0)
{ {
DEBUG1 ("listeners have now moved to %s", source->fallback.mount); DEBUG1 ("listeners have now moved to %s", source->fallback.mount);
free (source->fallback.mount); free (source->fallback.mount);
@ -467,7 +463,7 @@ void source_read (source_t *source)
break; break;
} }
} }
fds = util_timed_wait_for_fd (client->con->sock, 0); fds = util_timed_wait_for_fd (client->connection.sock, 0);
if (fds < 0) if (fds < 0)
{ {
if (! sock_recoverable (sock_error())) if (! sock_recoverable (sock_error()))
@ -491,10 +487,10 @@ void source_read (source_t *source)
if (source->skip_duration < 60) if (source->skip_duration < 60)
source->skip_duration = 80; source->skip_duration = 80;
else else
source->skip_duration = (long)(source->skip_duration *1.3); source->skip_duration = (long)(source->skip_duration *1.8);
break; break;
} }
source->skip_duration = (long)(source->skip_duration * 0.5); source->skip_duration = (long)(source->skip_duration * 0.9);
skip = 0; skip = 0;
source->last_read = current; source->last_read = current;
@ -544,21 +540,20 @@ void source_read (source_t *source)
/* save stream to file */ /* save stream to file */
if (source->dumpfile && source->format->write_buf_to_file) if (source->dumpfile && source->format->write_buf_to_file)
source->format->write_buf_to_file (source, refbuf); source->format->write_buf_to_file (source, refbuf);
client->schedule_ms = client->worker->time_ms + 5;
} }
else else
{ {
if (client->con->error) skip = 1;
if (client->connection.error)
{ {
INFO1 ("End of Stream %s", source->mount); INFO1 ("End of Stream %s", source->mount);
source->flags &= ~SOURCE_RUNNING; source->flags &= ~SOURCE_RUNNING;
skip = 1;
} }
break; break;
} }
loop--; loop--;
} while (loop); } while (0);
if (loop == 0)
client->schedule_ms += 20;
/* lets see if we have too much data in the queue */ /* lets see if we have too much data in the queue */
while (source->queue_size > source->queue_size_limit || while (source->queue_size > source->queue_size_limit ||
@ -618,14 +613,20 @@ static int source_queue_advance (client_t *client)
refbuf_t *refbuf; refbuf_t *refbuf;
if (client->refbuf == NULL && locate_start_on_queue (source, client) < 0) if (client->refbuf == NULL && locate_start_on_queue (source, client) < 0)
{
client->schedule_ms += 200;
return -1; return -1;
}
refbuf = client->refbuf; refbuf = client->refbuf;
/* move to the next buffer if we have finished with the current one */ /* move to the next buffer if we have finished with the current one */
if (client->pos == refbuf->len) if (client->pos == refbuf->len)
{ {
if (refbuf->next == NULL) if (refbuf->next == NULL)
{
client->schedule_ms = source->client->schedule_ms + 20;
return -1; return -1;
}
client_set_queue (client, refbuf->next); client_set_queue (client, refbuf->next);
} }
return source->format->write_buf_to_client (client); return source->format->write_buf_to_client (client);
@ -695,18 +696,19 @@ static int http_source_intro (client_t *client)
} }
if (format_file_read (client, source->intro_file) < 0) if (format_file_read (client, source->intro_file) < 0)
{ {
client->schedule_ms = client->worker->time_ms + 1000;
if (source->stream_data_tail) if (source->stream_data_tail)
{ {
/* better find the right place in queue for this client */ /* better find the right place in queue for this client */
client_set_queue (client, NULL); client_set_queue (client, NULL);
client->check_buffer = source_queue_advance; client->check_buffer = source_queue_advance;
if (client->connection.sent_bytes == 0) // no intro
client->schedule_ms = client->worker->time_ms;
return 0; return 0;
} }
client->intro_offset = 0; /* replay intro file */ client->intro_offset = 0; /* replay intro file */
client->schedule_ms = client->worker->time_ms + 100;
return -1; return -1;
} }
client->schedule_ms = client->worker->time_ms + 20;
return source->format->write_buf_to_client (client); return source->format->write_buf_to_client (client);
} }
@ -724,17 +726,22 @@ static int http_source_listener (client_t *client)
if (client->respcode == 0) if (client->respcode == 0)
{ {
int (*build_headers)(source_t *, client_t *) = format_general_headers;
if (source_running (source) == 0) if (source_running (source) == 0)
{ {
client->schedule_ms = client->worker->time_ms + 200; client->schedule_ms = client->worker->time_ms + 200;
return 0; return 0;
} }
if (format_prepare_headers (source, client) < 0) if (source->format->create_client_data)
build_headers = source->format->create_client_data;
refbuf->len = 0;
if (build_headers (source, client) < 0)
{ {
ERROR0 ("internal problem, dropping client"); ERROR0 ("internal problem, dropping client");
return -1; return -1;
} }
client->respcode = 200;
stats_event_inc (NULL, "listeners"); stats_event_inc (NULL, "listeners");
stats_event_inc (NULL, "listener_connections"); stats_event_inc (NULL, "listener_connections");
stats_event_inc (source->mount, "listener_connections"); stats_event_inc (source->mount, "listener_connections");
@ -743,11 +750,18 @@ static int http_source_listener (client_t *client)
{ {
client->check_buffer = http_source_intro; client->check_buffer = http_source_intro;
client->intro_offset = 0; client->intro_offset = 0;
client->pos = refbuf->len = 4096; if (client->flags & CLIENT_HAS_INTRO_CONTENT)
client->con->sent_bytes = 0; {
return -1; client->refbuf = refbuf->next;
refbuf->next = NULL;
refbuf_release (refbuf);
client->pos = 0;
}
else
client->pos = refbuf->len = 4096;
client->connection.sent_bytes = 0;
return 0;
} }
client->schedule_ms = client->worker->time_ms + 10;
return format_generic_write_to_client (client); return format_generic_write_to_client (client);
} }
@ -757,12 +771,12 @@ static int http_source_listener (client_t *client)
static int send_to_listener (client_t *client) static int send_to_listener (client_t *client)
{ {
int bytes; int bytes;
int loop = 8; /* max number of iterations in one go */ int loop = 6; /* max number of iterations in one go */
long total_written = 0; long total_written = 0;
int ret = 0; int ret = 0;
source_t *source = client->shared_data; source_t *source = client->shared_data;
if (client->con->error) if (client->connection.error || source == NULL)
return -1; return -1;
if (source->fallback.mount) if (source->fallback.mount)
{ {
@ -779,7 +793,8 @@ static int send_to_listener (client_t *client)
*pnext = client->next; *pnext = client->next;
if (client->check_buffer != http_source_listener) if (client->check_buffer != http_source_listener)
{ {
client_set_queue (client, NULL); if ((client->flags & CLIENT_HAS_INTRO_CONTENT) == 0)
client_set_queue (client, NULL);
client->check_buffer = source->format->write_buf_to_client; client->check_buffer = source->format->write_buf_to_client;
} }
thread_mutex_unlock (&source->lock); thread_mutex_unlock (&source->lock);
@ -802,10 +817,10 @@ static int send_to_listener (client_t *client)
return -1; return -1;
} }
/* check for limited listener time */ /* check for limited listener time */
if (client->con->discon_time && if (client->connection.discon_time &&
client->worker->current_time.tv_sec >= client->con->discon_time) client->worker->current_time.tv_sec >= client->connection.discon_time)
{ {
INFO1 ("time limit reached for client #%lu", client->con->id); INFO1 ("time limit reached for client #%lu", client->connection.id);
return -1; return -1;
} }
if (source_running (source) == 0) if (source_running (source) == 0)
@ -824,7 +839,7 @@ static int send_to_listener (client_t *client)
while (loop) while (loop)
{ {
/* jump out if client connection has died */ /* jump out if client connection has died */
if (client->con->error) if (client->connection.error)
{ {
ret = -1; ret = -1;
break; break;
@ -832,17 +847,20 @@ static int send_to_listener (client_t *client)
/* lets not send too much to one client in one go, but don't /* lets not send too much to one client in one go, but don't
sleep for too long if more data can be sent */ sleep for too long if more data can be sent */
if (total_written > source->listener_send_trigger) if (total_written > source->listener_send_trigger)
{
client->schedule_ms = client->worker->time_ms;
break; break;
}
bytes = client->check_buffer (client); bytes = client->check_buffer (client);
if (bytes < 0) if (bytes < 0)
break; /* can't write any more */ break; /* can't write any more */
client->schedule_ms += 5; client->schedule_ms += 100;
total_written += bytes; total_written += bytes;
loop--; loop--;
} }
if (loop == 0) if (loop == 0)
client->schedule_ms -= 20; client->schedule_ms -= 500;
if (total_written) if (total_written)
{ {
rate_add (source->format->out_bitrate, total_written, client->worker->time_ms); rate_add (source->format->out_bitrate, total_written, client->worker->time_ms);
@ -855,7 +873,7 @@ static int send_to_listener (client_t *client)
if (client->refbuf && (client->refbuf->flags & SOURCE_BLOCK_RELEASE)) if (client->refbuf && (client->refbuf->flags & SOURCE_BLOCK_RELEASE))
{ {
INFO2 ("Client %lu (%s) has fallen too far behind, removing", INFO2 ("Client %lu (%s) has fallen too far behind, removing",
client->con->id, client->con->ip); client->connection.id, client->connection.ip);
stats_event_inc (source->mount, "slow_listeners"); stats_event_inc (source->mount, "slow_listeners");
client_set_queue (client, NULL); client_set_queue (client, NULL);
ret = -1; ret = -1;
@ -893,7 +911,7 @@ void source_init (source_t *source)
stats_event_hidden (source->mount, "total_mbytes_sent", "0", STATS_COUNTERS); stats_event_hidden (source->mount, "total_mbytes_sent", "0", STATS_COUNTERS);
stats_event_hidden (source->mount, "total_bytes_sent", "0", STATS_COUNTERS); stats_event_hidden (source->mount, "total_bytes_sent", "0", STATS_COUNTERS);
stats_event_hidden (source->mount, "total_bytes_read", "0", STATS_COUNTERS); stats_event_hidden (source->mount, "total_bytes_read", "0", STATS_COUNTERS);
stats_event (source->mount, "source_ip", source->client->con->ip); stats_event (source->mount, "source_ip", source->client->connection.ip);
source->last_read = time(NULL); source->last_read = time(NULL);
source->prev_listeners = -1; source->prev_listeners = -1;
@ -955,8 +973,17 @@ void source_set_override (const char *mount, const char *dest)
source = source_find_mount (mount); source = source_find_mount (mount);
if (source) if (source)
{ {
source->fallback.limit = 0; if (strcmp (source->mount, dest) != 0)
source->fallback.mount = strdup (dest); {
thread_mutex_lock (&source->lock);
if (source->listeners && source->fallback.mount == NULL)
{
source->fallback.limit = 0;
source->fallback.mount = strdup (dest);
source->termination_count = source->listeners;
}
thread_mutex_unlock (&source->lock);
}
} }
else else
fserve_set_override (mount, dest); fserve_set_override (mount, dest);
@ -1002,21 +1029,6 @@ void source_shutdown (source_t *source, int with_fallback)
} }
static int _compare_clients(void *compare_arg, void *a, void *b)
{
client_t *clienta = (client_t *)a;
client_t *clientb = (client_t *)b;
connection_t *cona = clienta->con;
connection_t *conb = clientb->con;
if (cona->id < conb->id) return -1;
if (cona->id > conb->id) return 1;
return 0;
}
static void _parse_audio_info (source_t *source, const char *s) static void _parse_audio_info (source_t *source, const char *s)
{ {
const char *start = s; const char *start = s;
@ -1330,13 +1342,14 @@ void source_update_settings (ice_config_t *config, source_t *source, mount_proxy
} }
int source_client_callback (client_t *client, void *arg) static int source_client_callback (client_t *client)
{ {
const char *agent; const char *agent;
source_t *source = arg; source_t *source = client->shared_data;
if (client->con->error) /* did http response fail? */ if (client->connection.error) /* did http response fail? */
{ {
thread_mutex_unlock (&source->lock);
global_lock(); global_lock();
global.sources--; global.sources--;
global_unlock(); global_unlock();
@ -1352,7 +1365,6 @@ int source_client_callback (client_t *client, void *arg)
source_init (source); source_init (source);
client->ops = &source_client_ops; client->ops = &source_client_ops;
client->shared_data = source;
return 0; return 0;
} }
@ -1474,12 +1486,12 @@ static int check_duplicate_logins (source_t *source, client_t *client, auth_t *a
existing = source->client_list; existing = source->client_list;
while (existing) while (existing)
{ {
if (existing->con->error == 0 && existing->username && if (existing->connection.error == 0 && existing->username &&
strcmp (existing->username, client->username) == 0) strcmp (existing->username, client->username) == 0)
{ {
if (auth->drop_existing_listener) if (auth->drop_existing_listener)
{ {
existing->con->error = 1; existing->connection.error = 1;
return 1; return 1;
} }
else else
@ -1500,9 +1512,9 @@ static int source_client_shutdown (client_t *client)
int ret = -1; int ret = -1;
client->schedule_ms = client->worker->time_ms + 100; client->schedule_ms = client->worker->time_ms + 100;
if (client->con->discon_time) if (client->connection.discon_time)
{ {
if (client->con->discon_time >= client->worker->current_time.tv_sec) if (client->connection.discon_time >= client->worker->current_time.tv_sec)
return 0; return 0;
else else
return -1; return -1;
@ -1513,7 +1525,7 @@ static int source_client_shutdown (client_t *client)
if (source->wait_time) if (source->wait_time)
{ {
/* set a wait time for leaving the source reserved */ /* set a wait time for leaving the source reserved */
client->con->discon_time = client->worker->current_time.tv_sec + source->wait_time; client->connection.discon_time = client->worker->current_time.tv_sec + source->wait_time;
INFO2 ("keeping %s reserved for %d seconds", source->mount, source->wait_time); INFO2 ("keeping %s reserved for %d seconds", source->mount, source->wait_time);
ret = 0; ret = 0;
} }
@ -1532,7 +1544,7 @@ void source_client_release (client_t *client)
thread_mutex_lock (&source->lock); thread_mutex_lock (&source->lock);
/* log bytes read in access log */ /* log bytes read in access log */
if (source->format) if (source->format)
client->con->sent_bytes = source->format->read_bytes; client->connection.sent_bytes = source->format->read_bytes;
thread_mutex_unlock (&source->lock); thread_mutex_unlock (&source->lock);
client_destroy (client); client_destroy (client);
@ -1551,6 +1563,11 @@ static void source_listener_release (client_t *client)
client_t **pnext; client_t **pnext;
int value; int value;
if (source == NULL)
{
client_destroy (client);
return;
}
if (source_running (source) == 0) if (source_running (source) == 0)
return; return;
@ -1621,6 +1638,7 @@ int source_add_listener (const char *mount, mount_proxy *mountinfo, client_t *cl
{ {
if (source->client == NULL && (source->flags & SOURCE_ON_DEMAND) == 0) if (source->client == NULL && (source->flags & SOURCE_ON_DEMAND) == 0)
{ {
thread_mutex_unlock (&source->lock);
client_send_403 (client, "Slave relay reading from time unregulated stream"); client_send_403 (client, "Slave relay reading from time unregulated stream");
return -1; return -1;
} }
@ -1657,8 +1675,8 @@ int source_add_listener (const char *mount, mount_proxy *mountinfo, client_t *cl
} }
/* set a per-mount disconnect time if auth hasn't set one already */ /* set a per-mount disconnect time if auth hasn't set one already */
if (mountinfo->max_listener_duration && client->con->discon_time == 0) if (mountinfo->max_listener_duration && client->connection.discon_time == 0)
client->con->discon_time = time(NULL) + mountinfo->max_listener_duration; client->connection.discon_time = time(NULL) + mountinfo->max_listener_duration;
INFO3 ("max on %s is %ld (cur %lu)", source->mount, INFO3 ("max on %s is %ld (cur %lu)", source->mount,
mountinfo->max_listeners, source->listeners); mountinfo->max_listeners, source->listeners);
@ -1698,7 +1716,7 @@ int source_add_listener (const char *mount, mount_proxy *mountinfo, client_t *cl
return -1; return -1;
} while (1); } while (1);
client->con->sent_bytes = 0; client->connection.sent_bytes = 0;
client->refbuf->len = PER_CLIENT_REFBUF_SIZE; client->refbuf->len = PER_CLIENT_REFBUF_SIZE;
memset (client->refbuf->data, 0, PER_CLIENT_REFBUF_SIZE); memset (client->refbuf->data, 0, PER_CLIENT_REFBUF_SIZE);
@ -1731,6 +1749,7 @@ void source_setup_listener (source_t *source, client_t *client)
static int source_client_http_send (client_t *client) static int source_client_http_send (client_t *client)
{ {
refbuf_t *stream; refbuf_t *stream;
source_t *source = client->shared_data;
if (client->pos < client->refbuf->len) if (client->pos < client->refbuf->len)
{ {
@ -1744,7 +1763,8 @@ static int source_client_http_send (client_t *client)
client->refbuf = stream; client->refbuf = stream;
client->pos = client->intro_offset; client->pos = client->intro_offset;
client->intro_offset = 0; client->intro_offset = 0;
return source_client_callback (client, client->shared_data); thread_mutex_lock (&source->lock);
return source_client_callback (client);
} }
@ -1770,7 +1790,7 @@ int source_startup (client_t *client, const char *uri)
if (client->server_conn && client->server_conn->shoutcast_compat) if (client->server_conn && client->server_conn->shoutcast_compat)
{ {
source->flags |= SOURCE_SHOUTCAST_COMPAT; source->flags |= SOURCE_SHOUTCAST_COMPAT;
source_client_callback (client, source); source_client_callback (client);
} }
else else
{ {

3
src/source.h
View File

@ -86,13 +86,12 @@ typedef struct source_tag
#define SOURCE_TERMINATING 020 #define SOURCE_TERMINATING 020
#define SOURCE_TEMPORARY_FALLBACK 040 #define SOURCE_TEMPORARY_FALLBACK 040
#define source_available(x) ((x)->flags & (SOURCE_RUNNING|SOURCE_ON_DEMAND)) #define source_available(x) (((x)->flags & (SOURCE_RUNNING|SOURCE_ON_DEMAND)) && (x)->fallback.mount == NULL)
#define source_running(x) ((x)->flags & SOURCE_RUNNING) #define source_running(x) ((x)->flags & SOURCE_RUNNING)
source_t *source_reserve (const char *mount); source_t *source_reserve (const char *mount);
void *source_client_thread (void *arg); void *source_client_thread (void *arg);
int source_startup (client_t *client, const char *uri); int source_startup (client_t *client, const char *uri);
int source_client_callback (client_t *client, void *source);
void source_update_settings (ice_config_t *config, source_t *source, mount_proxy *mountinfo); void source_update_settings (ice_config_t *config, source_t *source, mount_proxy *mountinfo);
void source_clear_listeners (source_t *source); void source_clear_listeners (source_t *source);
void source_clear_source (source_t *source); void source_clear_source (source_t *source);

2
src/stats.c
View File

@ -585,7 +585,7 @@ static int stats_listeners_send (client_t *client)
int ret = 0; int ret = 0;
event_listener_t *listener = client->shared_data; event_listener_t *listener = client->shared_data;
if (client->con->error) if (client->connection.error)
return -1; return -1;
if (client->flags & STATS_LARGE) if (client->flags & STATS_LARGE)
loop = 4; loop = 4;