MyFavMirrors/owntone-server
1
0
Fork 0
mirror of https://github.com/owntone/owntone-server.git synced 2024-12-26 23:25:56 -05:00
Code Issues Packages Projects Releases Wiki Activity

Add a worker thread to support async tasks from the player thread

Browse Source 
(and maybe others later)
This commit is contained in:
ejurgensen 2015-03-31 23:05:24 +02:00
parent 96200eb808
commit e49c941a00
12 changed files with 511 additions and 171 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/Makefile.am
View File

@ -120,6 +120,7 @@ forked_daapd_SOURCES = main.c \
rsp_query.c rsp_query.h \
daap_query.c daap_query.h \
player.c player.h \
worker.c worker.h \
$(ALSA_SRC) $(OSS4_SRC) laudio.h \
raop.c raop.h \
$(RTSP_SRC) \

87
src/db.c
View File

@ -62,11 +62,6 @@ struct db_unlock {
pthread_mutex_t lck;
};
struct async_query {
char *query;
int delay;
};
#define DB_TYPE_CHAR 1
#define DB_TYPE_INT 2
#define DB_TYPE_INT64 3
@ -623,83 +618,6 @@ db_exec(const char *query, char **errmsg)
}
// This will run in its own shortlived, detached thread, created by db_exec_nonblock
static void *
db_exec_thread(void *arg)
{
struct async_query *async = arg;
char *errmsg;
time_t start, end;
int ret;
// When switching tracks we update playcount and select the next track's
// metadata. We want the update to run after the selects so it won't lock
// the database.
sleep(async->delay);
ret = db_perthread_init();
if (ret < 0)
{
DPRINTF(E_LOG, L_DB, "Error in db_exec_thread: Could not init thread\n");
return NULL;
}
DPRINTF(E_DBG, L_DB, "Running delayed query '%s'\n", async->query);
time(&start);
ret = db_exec(async->query, &errmsg);
if (ret != SQLITE_OK)
DPRINTF(E_LOG, L_DB, "Error running query '%s': %s\n", async->query, errmsg);
time(&end);
if (end - start > 1)
DPRINTF(E_LOG, L_DB, "Warning: Slow query detected '%s' - database performance problems?\n", async->query);
sqlite3_free(errmsg);
sqlite3_free(async->query);
free(async);
db_perthread_deinit();
return NULL;
}
// Creates a one-off thread to run a delayed, fire-and-forget, non-blocking query
static void
db_exec_nonblock(char *query, int delay)
{
struct async_query *async;
pthread_t tid;
pthread_attr_t attr;
int ret;
ret = pthread_attr_init(&attr);
if (ret != 0)
{
DPRINTF(E_LOG, L_DB, "Error in db_exec_nonblock: Could not init attributes\n");
return;
}
async = malloc(sizeof(struct async_query));
if (!async)
{
DPRINTF(E_LOG, L_DB, "Out of memory\n");
return;
}
async->query = query;
async->delay = delay;
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
ret = pthread_create(&tid, &attr, db_exec_thread, async);
if (ret != 0)
{
DPRINTF(E_LOG, L_DB, "Error in db_exec_nonblock: Could not create thread\n");
}
pthread_attr_destroy(&attr);
}
/* Maintenance and DB hygiene */
static void
db_analyze(void)
@ -2073,8 +1991,7 @@ db_file_inc_playcount(int id)
return;
}
// Run the query non-blocking so we don't block playback if the update is slow
db_exec_nonblock(query, 5);
db_query_run(query, 1, 0);
#undef Q_TMPL
}
@ -2737,7 +2654,7 @@ db_file_update_icy(int id, char *artist, char *album)
return;
}
db_exec_nonblock(query, 0);
db_query_run(query, 1, 0);
#undef Q_TMPL
}

2
src/filescanner_ffmpeg.c
View File

@ -530,7 +530,7 @@ scan_metadata_ffmpeg(char *file, struct media_file_info *mfi)
mfi->genre = strdup(icy_metadata->genre);
}
if (icy_metadata)
http_icy_metadata_free(icy_metadata);
http_icy_metadata_free(icy_metadata, 0);
}
/* Get some more information on the audio stream */

5
src/http.c
View File

@ -621,7 +621,7 @@ http_icy_metadata_get(AVFormatContext *fmtctx, int packet_only)
#endif
void
http_icy_metadata_free(struct http_icy_metadata *metadata)
http_icy_metadata_free(struct http_icy_metadata *metadata, int content_only)
{
if (metadata->name)
free(metadata->name);
@ -641,5 +641,6 @@ http_icy_metadata_free(struct http_icy_metadata *metadata)
if (metadata->artwork_url)
free(metadata->artwork_url);
free(metadata);
if (!content_only)
free(metadata);
}

5
src/http.h
View File

@ -35,6 +35,8 @@ struct http_client_ctx
struct http_icy_metadata
{
uint32_t id;
/* Static stream metadata from icy_metadata_headers */
char *name;
char *description;
@ -91,8 +93,9 @@ http_icy_metadata_get(AVFormatContext *fmtctx, int packet_only);
/* Frees an ICY metadata struct
*
* @param metadata struct to free
* @param content_only just free content, not the struct
*/
void
http_icy_metadata_free(struct http_icy_metadata *metadata);
http_icy_metadata_free(struct http_icy_metadata *metadata, int content_only);
#endif /* !__HTTP_H__ */

15
src/main.c
View File

@ -65,6 +65,7 @@ GCRY_THREAD_OPTION_PTHREAD_IMPL;
#include "mdns.h"
#include "remote_pairing.h"
#include "player.h"
#include "worker.h"
#if LIBAVFORMAT_VERSION_MAJOR < 53
# include "ffmpeg_url_evbuffer.h"
#endif
@ -676,6 +677,16 @@ main(int argc, char **argv)
goto db_fail;
}
/* Spawn worker thread */
ret = worker_init();
if (ret != 0)
{
DPRINTF(E_FATAL, L_MAIN, "Worker thread failed to start\n");
ret = EXIT_FAILURE;
goto worker_fail;
}
/* Spawn cache thread */
ret = cache_init();
if (ret != 0)
@ -848,6 +859,10 @@ main(int argc, char **argv)
cache_deinit();
cache_fail:
DPRINTF(E_LOG, L_MAIN, "Worker deinit\n");
worker_deinit();
worker_fail:
DPRINTF(E_LOG, L_MAIN, "Database deinit\n");
db_perthread_deinit();
db_deinit();

69
src/player.c
View File

@ -57,6 +57,7 @@
#include "player.h"
#include "raop.h"
#include "laudio.h"
#include "worker.h"
#ifdef LASTFM
# include "lastfm.h"
@ -625,6 +626,32 @@ player_laudio_status_cb(enum laudio_state status)
}
}
/* Callbacks from the worker thread */
static void
playcount_inc_cb(void *arg)
{
int *id = arg;
db_file_inc_playcount(*id);
}
static void
metadata_send_cb(void *arg)
{
struct raop_metadata_arg *rma = arg;
raop_metadata_send(rma);
}
static void
update_icy_cb(void *arg)
{
struct http_icy_metadata *metadata = arg;
db_file_update_icy(metadata->id, metadata->artist, metadata->title);
http_icy_metadata_free(metadata, 1);
}
/* Metadata */
static void
@ -642,35 +669,37 @@ metadata_purge(void)
static void
metadata_send(struct player_source *ps, int startup)
{
uint64_t offset;
uint64_t rtptime;
struct raop_metadata_arg rma;
offset = 0;
rma.id = ps->id;
rma.offset = 0;
rma.startup = startup;
/* Determine song boundaries, dependent on context */
/* Restart after pause/seek */
if (ps->stream_start)
{
offset = ps->output_start - ps->stream_start;
rtptime = ps->stream_start;
rma.offset = ps->output_start - ps->stream_start;
rma.rtptime = ps->stream_start;
}
else if (startup)
{
rtptime = last_rtptime + AIRTUNES_V2_PACKET_SAMPLES;
rma.rtptime = last_rtptime + AIRTUNES_V2_PACKET_SAMPLES;
}
/* Generic case */
else if (cur_streaming && (cur_streaming->end))
{
rtptime = cur_streaming->end + 1;
rma.rtptime = cur_streaming->end + 1;
}
else
{
rtptime = 0;
rma.rtptime = 0;
DPRINTF(E_LOG, L_PLAYER, "PTOH! Unhandled song boundary case in metadata_send()\n");
}
raop_metadata_send(ps->id, rtptime, offset, startup);
/* Defer the actual work of sending the metadata to the worker thread */
worker_execute(metadata_send_cb, &rma, sizeof(struct raop_metadata_arg), 0);
}
static void
@ -697,13 +726,23 @@ metadata_icy_poll_cb(int fd, short what, void *arg)
if (!changed)
goto no_update;
/* Update db (async) and send status update to clients */
db_file_update_icy(cur_streaming->id, metadata->artist, metadata->title);
metadata->id = cur_streaming->id;
/* Defer the database update to the worker thread */
worker_execute(update_icy_cb, metadata, sizeof(struct http_icy_metadata), 0);
status_update(player_state);
metadata_send(cur_streaming, 0);
/* Only free the struct, the content must be preserved for update_icy_cb */
free(metadata);
evtimer_add(metaev, &tv);
return;
no_update:
http_icy_metadata_free(metadata);
http_icy_metadata_free(metadata, 0);
no_metadata:
evtimer_add(metaev, &tv);
@ -1623,6 +1662,7 @@ source_check(void)
uint64_t pos;
enum repeat_mode r_mode;
int i;
int id;
int ret;
if (!cur_streaming)
@ -1699,9 +1739,10 @@ source_check(void)
{
i++;
db_file_inc_playcount((int)cur_playing->id);
id = (int)cur_playing->id;
worker_execute(playcount_inc_cb, &id, sizeof(int), 5);
#ifdef LASTFM
lastfm_scrobble((int)cur_playing->id);
lastfm_scrobble(id);
#endif
/* Stop playback if:

72
src/raop.c
View File

@ -61,7 +61,6 @@
#include "evrtsp/evrtsp.h"
#include <gcrypt.h>
#include <pthread.h>
#include "conffile.h"
#include "logger.h"
@ -158,14 +157,6 @@ struct raop_metadata
struct raop_metadata *next;
};
struct raop_metadata_send_ctx
{
int id;
uint64_t rtptime;
uint64_t offset;
int startup;
};
struct raop_service
{
int fd;
@ -2174,25 +2165,17 @@ raop_metadata_startup_send(struct raop_session *rs)
}
}
static void *
raop_metadata_send_thread(void *arg)
void
raop_metadata_send(struct raop_metadata_arg *rma)
{
struct raop_metadata_send_ctx *ctx = arg;
struct raop_session *rs;
struct raop_metadata *rmd;
uint32_t delay;
int ret;
ret = db_perthread_init();
if (ret < 0)
{
DPRINTF(E_LOG, L_RAOP, "Error in raop_metadata_send_thread: Could not init thread\n");
return NULL;
}
rmd = raop_metadata_prepare(ctx->id, ctx->rtptime);
rmd = raop_metadata_prepare(rma->id, rma->rtptime);
if (!rmd)
goto no_metadata;
return;
for (rs = sessions; rs; rs = rs->next)
{
@ -2202,58 +2185,15 @@ raop_metadata_send_thread(void *arg)
if (!rs->wants_metadata)
continue;
delay = (ctx->startup) ? RAOP_MD_DELAY_STARTUP : RAOP_MD_DELAY_SWITCH;
delay = (rma->startup) ? RAOP_MD_DELAY_STARTUP : RAOP_MD_DELAY_SWITCH;
ret = raop_metadata_send_internal(rs, rmd, ctx->offset, delay);
ret = raop_metadata_send_internal(rs, rmd, rma->offset, delay);
if (ret < 0)
{
raop_session_failure(rs);
continue;
}
}
no_metadata:
db_perthread_deinit();
free(ctx);
return NULL;
}
void
raop_metadata_send(int id, uint64_t rtptime, uint64_t offset, int startup)
{
struct raop_metadata_send_ctx *ctx;
pthread_t tid;
pthread_attr_t attr;
int ret;
ctx = malloc(sizeof(struct raop_metadata_send_ctx));
if (!ctx)
{
DPRINTF(E_LOG, L_RAOP, "Out of memory\n");
return;
}
ctx->id = id;
ctx->rtptime = rtptime;
ctx->offset = offset;
ctx->startup = startup;
ret = pthread_attr_init(&attr);
if (ret != 0)
{
DPRINTF(E_LOG, L_RAOP, "Error in raop_metadata_send: Could not init attributes\n");
return;
}
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
ret = pthread_create(&tid, &attr, raop_metadata_send_thread, ctx);
if (ret != 0)
{
DPRINTF(E_LOG, L_RAOP, "Error in raop_metadata_send: Could not create thread\n");
}
pthread_attr_destroy(&attr);
}
/* Volume handling */

10
src/raop.h
View File

@ -54,6 +54,14 @@ struct raop_device
struct raop_device *next;
};
struct raop_metadata_arg
{
int id;
uint64_t rtptime;
uint64_t offset;
int startup;
};
/* RAOP session state */
/* Session is starting up */
@ -115,7 +123,7 @@ raop_playback_stop(void);
void
raop_metadata_send(int id, uint64_t rtptime, uint64_t offset, int startup);
raop_metadata_send(struct raop_metadata_arg *rma);
int
raop_set_volume_one(struct raop_session *rs, int volume, raop_status_cb cb);

2
src/transcode.c
View File

@ -937,5 +937,5 @@ transcode_metadata_artwork_url(struct transcode_ctx *ctx, char **artwork_url, ch
if (m->artwork_url)
*artwork_url = strdup(m->artwork_url);
http_icy_metadata_free(m);
http_icy_metadata_free(m, 0);
}

386
src/worker.c Normal file
View File

@ -0,0 +1,386 @@
/*
* Copyright (C) 2014 Espen Jürgensen <espenjurgensen@gmail.com>
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 2 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*/
#ifdef HAVE_CONFIG_H
# include <config.h>
#endif
#include <stdio.h>
#include <inttypes.h>
#include <stdlib.h>
#include <stdint.h>
#include <fcntl.h>
#include <unistd.h>
#include <time.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <event2/event.h>
#include "db.h"
#include "logger.h"
#include "worker.h"
struct worker_command;
typedef int (*cmd_func)(struct worker_command *cmd);
struct worker_command
{
pthread_mutex_t lck;
pthread_cond_t cond;
cmd_func func;
int nonblock;
struct {
void (*cb)(void *);
void *cb_arg;
int delay;
struct event *timer;
} arg;
int ret;
};
/* --- Globals --- */
// worker thread
static pthread_t tid_worker;
// Event base, pipes and events
struct event_base *evbase_worker;
static int g_initialized;
static int g_exit_pipe[2];
static int g_cmd_pipe[2];
static struct event *g_exitev;
static struct event *g_cmdev;
/* --------------------------------- HELPERS ------------------------------- */
static void
execute_cb(int fd, short what, void *arg)
{
struct worker_command *cmd = arg;
cmd->arg.cb(cmd->arg.cb_arg);
event_free(cmd->arg.timer);
free(cmd->arg.cb_arg);
}
static int
execute(struct worker_command *cmd)
{
struct timeval tv = { cmd->arg.delay, 0 };
if (cmd->arg.delay)
{
cmd->arg.timer = evtimer_new(evbase_worker, execute_cb, cmd);
evtimer_add(cmd->arg.timer, &tv);
return 1; // Not done yet, ask caller not to free cmd
}
cmd->arg.cb(cmd->arg.cb_arg);
free(cmd->arg.cb_arg);
return 0;
}
/* ---------------------------- COMMAND EXECUTION -------------------------- */
static int
send_command(struct worker_command *cmd)
{
int ret;
if (!cmd->func)
{
DPRINTF(E_LOG, L_MISC, "BUG: cmd->func is NULL!\n");
return -1;
}
ret = write(g_cmd_pipe[1], &cmd, sizeof(cmd));
if (ret != sizeof(cmd))
{
DPRINTF(E_LOG, L_MISC, "Could not send command: %s\n", strerror(errno));
return -1;
}
return 0;
}
static int
nonblock_command(struct worker_command *cmd)
{
int ret;
ret = send_command(cmd);
if (ret < 0)
return -1;
return 0;
}
/* Thread: main */
static void
thread_exit(void)
{
int dummy = 42;
DPRINTF(E_DBG, L_MISC, "Killing worker thread\n");
if (write(g_exit_pipe[1], &dummy, sizeof(dummy)) != sizeof(dummy))
DPRINTF(E_LOG, L_MISC, "Could not write to exit fd: %s\n", strerror(errno));
}
/* --------------------------------- MAIN --------------------------------- */
/* Thread: worker */
static void *
worker(void *arg)
{
int ret;
DPRINTF(E_DBG, L_MISC, "Worker loop initiating\n");
ret = db_perthread_init();
if (ret < 0)
{
DPRINTF(E_LOG, L_MISC, "Error: DB init failed\n");
pthread_exit(NULL);
}
g_initialized = 1;
event_base_dispatch(evbase_worker);
if (g_initialized)
{
DPRINTF(E_LOG, L_MISC, "Worker event loop terminated ahead of time!\n");
g_initialized = 0;
}
db_perthread_deinit();
DPRINTF(E_DBG, L_MISC, "Worker loop terminating\n");
pthread_exit(NULL);
}
static void
exit_cb(int fd, short what, void *arg)
{
int dummy;
int ret;
ret = read(g_exit_pipe[0], &dummy, sizeof(dummy));
if (ret != sizeof(dummy))
DPRINTF(E_LOG, L_MISC, "Error reading from exit pipe\n");
event_base_loopbreak(evbase_worker);
g_initialized = 0;
event_add(g_exitev, NULL);
}
static void
command_cb(int fd, short what, void *arg)
{
struct worker_command *cmd;
int ret;
ret = read(g_cmd_pipe[0], &cmd, sizeof(cmd));
if (ret != sizeof(cmd))
{
DPRINTF(E_LOG, L_MISC, "Could not read command! (read %d): %s\n", ret, (ret < 0) ? strerror(errno) : "-no error-");
goto readd;
}
if (cmd->nonblock)
{
ret = cmd->func(cmd);
if (ret == 0)
free(cmd);
goto readd;
}
pthread_mutex_lock(&cmd->lck);
ret = cmd->func(cmd);
cmd->ret = ret;
pthread_cond_signal(&cmd->cond);
pthread_mutex_unlock(&cmd->lck);
readd:
event_add(g_cmdev, NULL);
}
/* ---------------------------- Our worker API --------------------------- */
/* Thread: player */
void
worker_execute(void (*cb)(void *), void *cb_arg, size_t arg_size, int delay)
{
struct worker_command *cmd;
void *argcpy;
DPRINTF(E_DBG, L_MISC, "Got worker execute request\n");
cmd = (struct worker_command *)malloc(sizeof(struct worker_command));
if (!cmd)
{
DPRINTF(E_LOG, L_MISC, "Could not allocate worker_command\n");
return;
}
memset(cmd, 0, sizeof(struct worker_command));
argcpy = malloc(arg_size);
if (!argcpy)
{
DPRINTF(E_LOG, L_MISC, "Out of memory\n");
return;
}
memcpy(argcpy, cb_arg, arg_size);
cmd->nonblock = 1;
cmd->func = execute;
cmd->arg.cb = cb;
cmd->arg.cb_arg = argcpy;
cmd->arg.delay = delay;
nonblock_command(cmd);
return;
}
int
worker_init(void)
{
int ret;
# if defined(__linux__)
ret = pipe2(g_exit_pipe, O_CLOEXEC);
# else
ret = pipe(g_exit_pipe);
# endif
if (ret < 0)
{
DPRINTF(E_LOG, L_MISC, "Could not create pipe: %s\n", strerror(errno));
goto exit_fail;
}
# if defined(__linux__)
ret = pipe2(g_cmd_pipe, O_CLOEXEC);
# else
ret = pipe(g_cmd_pipe);
# endif
if (ret < 0)
{
DPRINTF(E_LOG, L_MISC, "Could not create command pipe: %s\n", strerror(errno));
goto cmd_fail;
}
evbase_worker = event_base_new();
if (!evbase_worker)
{
DPRINTF(E_LOG, L_MISC, "Could not create an event base\n");
goto evbase_fail;
}
g_exitev = event_new(evbase_worker, g_exit_pipe[0], EV_READ, exit_cb, NULL);
if (!g_exitev)
{
DPRINTF(E_LOG, L_MISC, "Could not create exit event\n");
goto evnew_fail;
}
g_cmdev = event_new(evbase_worker, g_cmd_pipe[0], EV_READ, command_cb, NULL);
if (!g_cmdev)
{
DPRINTF(E_LOG, L_MISC, "Could not create cmd event\n");
goto evnew_fail;
}
event_add(g_exitev, NULL);
event_add(g_cmdev, NULL);
ret = pthread_create(&tid_worker, NULL, worker, NULL);
if (ret < 0)
{
DPRINTF(E_LOG, L_MISC, "Could not spawn worker thread: %s\n", strerror(errno));
goto thread_fail;
}
return 0;
thread_fail:
evnew_fail:
event_base_free(evbase_worker);
evbase_worker = NULL;
evbase_fail:
close(g_cmd_pipe[0]);
close(g_cmd_pipe[1]);
cmd_fail:
close(g_exit_pipe[0]);
close(g_exit_pipe[1]);
exit_fail:
return -1;
}
void
worker_deinit(void)
{
int ret;
thread_exit();
ret = pthread_join(tid_worker, NULL);
if (ret != 0)
{
DPRINTF(E_FATAL, L_MISC, "Could not join worker thread: %s\n", strerror(errno));
return;
}
// Free event base (should free events too)
event_base_free(evbase_worker);
// Close pipes
close(g_cmd_pipe[0]);
close(g_cmd_pipe[1]);
close(g_exit_pipe[0]);
close(g_exit_pipe[1]);
}

28
src/worker.h Normal file
View File

@ -0,0 +1,28 @@
#ifndef __WORKER_H__
#define __WORKER_H__
/* The worker thread is made for running asyncronous tasks from a real time
* thread, mainly the player thread.
* The worker_execute() function will trigger a callback from the worker thread.
* Before returning the function will copy the argument given, so the caller
* does not need to preserve them. However, if the argument contains pointers to
* data, the caller must either make sure that the data remains valid until the
* callback (which can free it), or make sure the callback does not refer to it.
*
* @param cb the function to call from the worker thread
* @param cb_arg arguments for callback
* @param arg_size size of the arguments given
* @param delay how much in seconds to delay the execution
*/
void
worker_execute(void (*cb)(void *), void *cb_arg, size_t arg_size, int delay);
int
worker_init(void);
void
worker_deinit(void);
#endif /* !__WORKER_H__ */