[tracker/extraction-improvements: 7/8] tracker-extract: Implement thread awareness for modules
- From: Carlos Garnacho <carlosg src gnome org>
- To: commits-list gnome org
- Cc:
- Subject: [tracker/extraction-improvements: 7/8] tracker-extract: Implement thread awareness for modules
- Date: Tue, 10 May 2011 16:04:37 +0000 (UTC)
commit 82e6cac6fcac58864efa70a8c1c1803a23b5ed04
Author: Carlos Garnacho <carlosg gnome org>
Date: Tue May 10 17:38:10 2011 +0200
tracker-extract: Implement thread awareness for modules
Tracker extract modules may provide now an init() function, which
also reports the thread awareness of the given module, allowing
these to run sequentially in the main thread, on a dedicated thread
for the extract module, or in a thread pool.
At the moment no extractor modules do this, so they run by default
in the main thread as before.
src/libtracker-extract/tracker-module-manager.c | 102 ++++---
src/libtracker-extract/tracker-module-manager.h | 17 +-
src/tracker-extract/tracker-extract.c | 347 +++++++++++++++++------
3 files changed, 331 insertions(+), 135 deletions(-)
---
diff --git a/src/libtracker-extract/tracker-module-manager.c b/src/libtracker-extract/tracker-module-manager.c
index 3b5fbdb..cdbb790 100644
--- a/src/libtracker-extract/tracker-module-manager.c
+++ b/src/libtracker-extract/tracker-module-manager.c
@@ -24,6 +24,8 @@
#include "tracker-module-manager.h"
#define EXTRACTOR_FUNCTION "tracker_extract_get_metadata"
+#define INIT_FUNCTION "tracker_extract_init"
+#define SHUTDOWN_FUNCTION "tracker_extract_shutdown"
typedef struct {
const gchar *module_path; /* intern string */
@@ -32,7 +34,10 @@ typedef struct {
typedef struct {
GModule *module;
- TrackerExtractMetadataFunc func;
+ TrackerModuleThreadAwareness thread_awareness;
+ TrackerExtractMetadataFunc extract_func;
+ TrackerExtractInitFunc init_func;
+ TrackerExtractShutdownFunc shutdown_func;
} ModuleInfo;
static GHashTable *modules = NULL;
@@ -205,16 +210,24 @@ lookup_rule (const gchar *mimetype)
}
GModule *
-tracker_extract_module_manager_get_for_mimetype (const gchar *mimetype,
- TrackerExtractMetadataFunc *func)
+tracker_extract_module_manager_get_for_mimetype (const gchar *mimetype,
+ TrackerExtractInitFunc *init_func,
+ TrackerExtractShutdownFunc *shutdown_func,
+ TrackerExtractMetadataFunc *extract_func)
{
- TrackerExtractMetadataFunc extract_func;
- ModuleInfo *module_info;
- GModule *module;
+ ModuleInfo *module_info = NULL;
RuleInfo *info;
- if (*func) {
- *func = NULL;
+ if (init_func) {
+ *init_func = NULL;
+ }
+
+ if (shutdown_func) {
+ *shutdown_func = NULL;
+ }
+
+ if (extract_func) {
+ *extract_func = NULL;
}
if (!initialized &&
@@ -230,53 +243,60 @@ tracker_extract_module_manager_get_for_mimetype (const gchar *mim
if (modules) {
module_info = g_hash_table_lookup (modules, info->module_path);
+ }
- if (module_info) {
- if (func) {
- *func = module_info->func;
- }
+ if (!module_info) {
+ GModule *module;
+
+ /* Load the module */
+ module = g_module_open (info->module_path, G_MODULE_BIND_LOCAL);
- return module_info->module;
+ if (!module) {
+ g_warning ("Could not load module '%s': %s",
+ info->module_path,
+ g_module_error ());
+ return NULL;
}
- }
- /* Load the module */
- module = g_module_open (info->module_path, G_MODULE_BIND_LOCAL);
+ g_module_make_resident (module);
- if (!module) {
- g_warning ("Could not load module '%s': %s",
- info->module_path,
- g_module_error ());
- return NULL;
- }
+ module_info = g_slice_new0 (ModuleInfo);
+ module_info->module = module;
- g_module_make_resident (module);
+ if (!g_module_symbol (module, EXTRACTOR_FUNCTION, (gpointer *) &module_info->extract_func)) {
+ g_warning ("Could not load module '%s': Function %s() was not found, is it exported?",
+ g_module_name (module), EXTRACTOR_FUNCTION);
+ g_slice_free (ModuleInfo, module_info);
+ return NULL;
+ }
- if (!g_module_symbol (module, EXTRACTOR_FUNCTION, (gpointer *) &extract_func)) {
- g_warning ("Could not load module '%s': Function %s() was not found, is it exported?",
- g_module_name (module), EXTRACTOR_FUNCTION);
- return NULL;
- }
+ g_module_symbol (module, INIT_FUNCTION, (gpointer *) &module_info->init_func);
+ g_module_symbol (module, SHUTDOWN_FUNCTION, (gpointer *) &module_info->shutdown_func);
+
+ /* Add it to the cache */
+ if (G_UNLIKELY (!modules)) {
+ /* Key is an intern string, so
+ * pointer comparison suffices
+ */
+ modules = g_hash_table_new (NULL, NULL);
+ }
- /* Add it to the cache */
- if (G_UNLIKELY (!modules)) {
- /* Key is an intern string, so
- * pointer comparison suffices
- */
- modules = g_hash_table_new (NULL, NULL);
+ g_hash_table_insert (modules, (gpointer) info->module_path, module_info);
}
- module_info = g_slice_new0 (ModuleInfo);
- module_info->module = module;
- module_info->func = extract_func;
+ if (extract_func) {
+ *extract_func = module_info->extract_func;
+ }
- g_hash_table_insert (modules, (gpointer) info->module_path, module_info);
+ if (init_func) {
+ *init_func = module_info->init_func;
+ }
- if (func) {
- *func = extract_func;
+ if (shutdown_func) {
+ *shutdown_func = module_info->shutdown_func;
}
- return module;
+ return module_info->module;
}
gboolean
diff --git a/src/libtracker-extract/tracker-module-manager.h b/src/libtracker-extract/tracker-module-manager.h
index 596e233..2839548 100644
--- a/src/libtracker-extract/tracker-module-manager.h
+++ b/src/libtracker-extract/tracker-module-manager.h
@@ -31,6 +31,17 @@
G_BEGIN_DECLS
+typedef enum {
+ TRACKER_MODULE_NONE,
+ TRACKER_MODULE_MAIN_THREAD,
+ TRACKER_MODULE_SINGLE_THREAD,
+ TRACKER_MODULE_MULTI_THREAD
+} TrackerModuleThreadAwareness;
+
+typedef gboolean (* TrackerExtractInitFunc) (TrackerModuleThreadAwareness *thread_awareness_ret,
+ GError **error);
+typedef void (* TrackerExtractShutdownFunc) (void);
+
typedef gboolean (* TrackerExtractMetadataFunc) (const gchar *uri,
const gchar *mime_type,
TrackerSparqlBuilder *preupdate,
@@ -39,8 +50,10 @@ typedef gboolean (* TrackerExtractMetadataFunc) (const gchar *uri,
gboolean tracker_extract_module_manager_init (void) G_GNUC_CONST;
-GModule * tracker_extract_module_manager_get_for_mimetype (const gchar *mimetype,
- TrackerExtractMetadataFunc *func);
+GModule * tracker_extract_module_manager_get_for_mimetype (const gchar *mimetype,
+ TrackerExtractInitFunc *init_func,
+ TrackerExtractShutdownFunc *shutdown_func,
+ TrackerExtractMetadataFunc *extract_func);
gboolean tracker_extract_module_manager_mimetype_is_handled (const gchar *mimetype);
G_END_DECLS
diff --git a/src/tracker-extract/tracker-extract.c b/src/tracker-extract/tracker-extract.c
index f4aa61e..e18b746 100644
--- a/src/tracker-extract/tracker-extract.c
+++ b/src/tracker-extract/tracker-extract.c
@@ -81,8 +81,25 @@ typedef struct {
} StatisticsData;
typedef struct {
+ TrackerExtract *extract;
+ GModule *module;
+ guint success : 1;
+} StatsReportData;
+
+typedef struct {
GHashTable *statistics_data;
+ /* module -> thread awareness enum for initialized modules */
+ GHashTable *modules;
+
+ /* Thread pool for multi-threaded extractors */
+ GThreadPool *thread_pool;
+
+ /* module -> async queue hashtable
+ * for single-threaded extractors
+ */
+ GHashTable *single_thread_extractors;
+
gboolean disable_shutdown;
gboolean force_internal_extractors;
gboolean disable_summary_on_finalize;
@@ -96,10 +113,14 @@ typedef struct {
GAsyncResult *res;
gchar *file;
gchar *mimetype;
+ TrackerExtractMetadataFunc func;
+ GModule *module;
} TrackerExtractTask;
static void tracker_extract_finalize (GObject *object);
static void report_statistics (GObject *object);
+static gboolean get_metadata (TrackerExtractTask *task);
+
G_DEFINE_TYPE(TrackerExtract, tracker_extract, G_TYPE_OBJECT)
@@ -126,6 +147,10 @@ tracker_extract_init (TrackerExtract *object)
priv = TRACKER_EXTRACT_GET_PRIVATE (object);
priv->statistics_data = g_hash_table_new (NULL, NULL);
+ priv->modules = g_hash_table_new (NULL, NULL);
+ priv->single_thread_extractors = g_hash_table_new (NULL, NULL);
+ priv->thread_pool = g_thread_pool_new ((GFunc) get_metadata,
+ NULL, 10, TRUE, NULL);
}
static void
@@ -135,6 +160,12 @@ tracker_extract_finalize (GObject *object)
priv = TRACKER_EXTRACT_GET_PRIVATE (object);
+ /* FIXME: Shutdown modules? */
+
+ g_hash_table_destroy (priv->modules);
+ g_hash_table_destroy (priv->single_thread_extractors);
+ g_thread_pool_free (priv->thread_pool, TRUE, FALSE);
+
if (!priv->disable_summary_on_finalize) {
report_statistics (object);
}
@@ -213,14 +244,49 @@ tracker_extract_new (gboolean disable_shutdown,
}
static gboolean
-get_file_metadata (TrackerExtract *extract,
- const gchar *uri,
- const gchar *mime,
+report_stats_cb (StatsReportData *report_data)
+{
+ TrackerExtractPrivate *priv;
+ StatisticsData *stats_data;
+
+ priv = TRACKER_EXTRACT_GET_PRIVATE (report_data->extract);
+ stats_data = g_hash_table_lookup (priv->statistics_data, report_data->module);
+
+ if (!stats_data) {
+ stats_data = g_slice_new0 (StatisticsData);
+ g_hash_table_insert (priv->statistics_data, report_data->module, stats_data);
+ }
+
+ stats_data->extracted_count++;
+
+ if (!report_data->success) {
+ stats_data->failed_count++;
+ }
+
+ return FALSE;
+}
+
+static void
+report_stats (TrackerExtractTask *task,
+ gboolean success)
+{
+ StatsReportData *data;
+
+ data = g_slice_new0 (StatsReportData);
+ data->extract = task->extract;
+ data->module = task->module;
+ data->success = success;
+
+ /* Send to main thread, where stats hashtable is maintained */
+ g_idle_add ((GSourceFunc) report_stats_cb, data);
+}
+
+static gboolean
+get_file_metadata (TrackerExtractTask *task,
TrackerSparqlBuilder **preupdate_out,
TrackerSparqlBuilder **statements_out,
gchar **where_out)
{
- TrackerExtractPrivate *priv;
TrackerSparqlBuilder *statements, *preupdate;
GString *where;
gchar *mime_used = NULL;
@@ -228,8 +294,6 @@ get_file_metadata (TrackerExtract *extract,
gchar *content_type = NULL;
#endif
- priv = TRACKER_EXTRACT_GET_PRIVATE (extract);
-
*preupdate_out = NULL;
*statements_out = NULL;
*where_out = NULL;
@@ -261,10 +325,9 @@ get_file_metadata (TrackerExtract *extract,
}
#endif /* HAVE_LIBSTREAMANALYZER */
- if (mime && *mime) {
+ if (task->mimetype && *task->mimetype) {
/* We know the mime */
- mime_used = g_strdup (mime);
- g_strstrip (mime_used);
+ mime_used = g_strdup (task->mimetype);
}
#ifdef HAVE_LIBSTREAMANALYZER
else if (content_type && *content_type) {
@@ -274,87 +337,20 @@ get_file_metadata (TrackerExtract *extract,
}
#endif /* HAVE_LIBSTREAMANALYZER */
else {
- GFile *file;
- GFileInfo *info;
- GError *error = NULL;
-
- file = g_file_new_for_uri (uri);
- if (!file) {
- g_warning ("Could not create GFile for uri:'%s'",
- uri);
- g_object_unref (statements);
- g_object_unref (preupdate);
- g_string_free (where, TRUE);
- return FALSE;
- }
-
- info = g_file_query_info (file,
- G_FILE_ATTRIBUTE_STANDARD_CONTENT_TYPE,
- G_FILE_QUERY_INFO_NONE,
- NULL,
- &error);
-
- if (error || !info) {
- /* FIXME: Propagate error */
- g_error_free (error);
-
- if (info) {
- g_object_unref (info);
- }
-
- g_object_unref (file);
- g_object_unref (statements);
- g_object_unref (preupdate);
- g_string_free (where, TRUE);
-
- return FALSE;
- }
-
- mime_used = g_strdup (g_file_info_get_content_type (info));
-
- g_object_unref (info);
- g_object_unref (file);
+ return FALSE;
}
/* Now we have sanity checked everything, actually get the
* data we need from the extractors.
*/
if (mime_used) {
- TrackerExtractMetadataFunc func;
- GModule *module;
-
- module = tracker_extract_module_manager_get_for_mimetype (mime_used, &func);
-
- if (module) {
- StatisticsData *data;
+ if (task->func) {
gint items;
- (func) (uri, mime_used, preupdate, statements, where);
+ (task->func) (task->file, mime_used, preupdate, statements, where);
items = tracker_sparql_builder_get_length (statements);
-
- data = g_hash_table_lookup (priv->statistics_data, module);
-
- if (!data) {
- data = g_slice_new0 (StatisticsData);
- g_hash_table_insert (priv->statistics_data, module, data);
- }
-
- data->extracted_count++;
-
- if (items > 0) {
- tracker_sparql_builder_insert_close (statements);
-
- *preupdate_out = preupdate;
- *statements_out = statements;
- *where_out = g_string_free (where, FALSE);
-
- return TRUE;
- } else {
- data->failed_count++;
- }
- } else {
- priv->unhandled_count++;
+ report_stats (task, items > 0);
}
}
@@ -386,7 +382,7 @@ tracker_extract_info_free (TrackerExtractInfo *info)
static TrackerExtractTask *
extract_task_new (TrackerExtract *extract,
- const gchar *file,
+ const gchar *uri,
const gchar *mimetype,
GCancellable *cancellable,
GAsyncResult *res)
@@ -395,31 +391,55 @@ extract_task_new (TrackerExtract *extract,
task = g_slice_new0 (TrackerExtractTask);
task->cancellable = cancellable;
- task->res = g_object_ref (res);
- task->file = g_strdup (file);
+ task->res = (res) ? g_object_ref (res) : NULL;
+ task->file = g_strdup (uri);
task->mimetype = g_strdup (mimetype);
task->extract = extract;
+ if (mimetype) {
+ task->mimetype = g_strdup (mimetype);
+ } else {
+ GFile *file;
+ GFileInfo *info;
+
+ file = g_file_new_for_uri (uri);
+ info = g_file_query_info (file,
+ G_FILE_ATTRIBUTE_STANDARD_CONTENT_TYPE,
+ G_FILE_QUERY_INFO_NONE,
+ NULL, NULL);
+
+ if (info) {
+ task->mimetype = g_strdup (g_file_info_get_content_type (info));
+ } else {
+ g_warning ("Could not get mimetype for '%s'", uri);
+ }
+
+ g_object_unref (info);
+ g_object_unref (file);
+ }
+
return task;
}
static void
extract_task_free (TrackerExtractTask *task)
{
- g_object_unref (task->res);
+ if (task->res) {
+ g_object_unref (task->res);
+ }
+
g_free (task->file);
g_free (task->mimetype);
g_slice_free (TrackerExtractTask, task);
}
static gboolean
-get_metadata_cb (gpointer user_data)
+get_metadata (TrackerExtractTask *task)
{
- TrackerExtractTask *task = user_data;
TrackerExtractInfo *info;
#ifdef THREAD_ENABLE_TRACE
- g_debug ("Thread:%p (Main) --> File:'%s' - Extracted",
+ g_debug ("Thread:%p --> File:'%s' - Extracted",
g_thread_self (),
task->file);
#endif /* THREAD_ENABLE_TRACE */
@@ -436,8 +456,7 @@ get_metadata_cb (gpointer user_data)
info = g_slice_new (TrackerExtractInfo);
- if (get_file_metadata (task->extract,
- task->file, task->mimetype,
+ if (get_file_metadata (task,
&info->preupdate,
&info->statements,
&info->where)) {
@@ -458,6 +477,137 @@ get_metadata_cb (gpointer user_data)
return FALSE;
}
+static void
+single_thread_get_metadata (GAsyncQueue *queue)
+{
+ while (TRUE) {
+ TrackerExtractTask *task;
+
+ task = g_async_queue_pop (queue);
+ g_message ("Dispatching '%s' in dedicated thread", task->file);
+ get_metadata (task);
+ }
+}
+
+/* This function is executed in the main thread, decides the
+ * module that's going to be run for a given task, and dispatches
+ * the task according to the threading strategy of that module.
+ */
+static gboolean
+dispatch_task_cb (TrackerExtractTask *task)
+{
+ TrackerModuleThreadAwareness thread_awareness;
+ TrackerExtractInitFunc init_func;
+ TrackerExtractPrivate *priv;
+ GError *error = NULL;
+ GModule *module;
+ gpointer value;
+
+#ifdef THREAD_ENABLE_TRACE
+ g_debug ("Thread:%p (Main) <-- File:'%s' - Dispatching\n",
+ g_thread_self (),
+ task->file);
+#endif /* THREAD_ENABLE_TRACE */
+
+ if (!task->mimetype) {
+ g_warning ("Discarding task with no mimetype for '%s'", task->file);
+ return FALSE;
+ }
+
+ priv = TRACKER_EXTRACT_GET_PRIVATE (task->extract);
+ task->module = module = tracker_extract_module_manager_get_for_mimetype (task->mimetype, &init_func, NULL, &task->func);
+
+ if (!module || !task->func) {
+ g_warning ("Discarding task with no module '%s'", task->file);
+ priv->unhandled_count++;
+ return FALSE;
+ }
+
+ if (g_hash_table_lookup_extended (priv->modules, module, NULL, &value)) {
+ thread_awareness = GPOINTER_TO_UINT (value);
+ } else {
+ /* Module not initialized */
+ if (init_func) {
+ if (! (init_func) (&thread_awareness, &error)) {
+ g_warning ("Could not initialize module '%s': %s", g_module_name (module),
+ (error) ? error->message : "No error given");
+ thread_awareness = TRACKER_MODULE_NONE;
+ g_clear_error (&error);
+ }
+ } else {
+ /* Backwards compatibility for modules without init() */
+ thread_awareness = TRACKER_MODULE_MAIN_THREAD;
+ }
+
+ g_hash_table_insert (priv->modules, module, GUINT_TO_POINTER (thread_awareness));
+ }
+
+ switch (thread_awareness) {
+ case TRACKER_MODULE_NONE:
+ /* Error out */
+ g_simple_async_result_set_error ((GSimpleAsyncResult *) task->res,
+ TRACKER_DBUS_ERROR, 0,
+ "Module '%s' initialization failed",
+ g_module_name (module));
+ g_simple_async_result_complete_in_idle ((GSimpleAsyncResult *) task->res);
+ extract_task_free (task);
+ break;
+ case TRACKER_MODULE_MAIN_THREAD:
+ /* Dispatch the task right away in this thread */
+ g_message ("Dispatching '%s' in main thread", task->file);
+ get_metadata (task);
+ break;
+ case TRACKER_MODULE_SINGLE_THREAD:
+ {
+ GAsyncQueue *async_queue;
+
+ async_queue = g_hash_table_lookup (priv->single_thread_extractors, module);
+
+ if (!async_queue) {
+ /* No thread created yet for this module, create it
+ * together with the async queue used to pass data to it
+ */
+ async_queue = g_async_queue_new ();
+
+ g_thread_create ((GThreadFunc) single_thread_get_metadata,
+ g_async_queue_ref (async_queue),
+ FALSE, &error);
+
+ if (error) {
+ g_simple_async_result_set_from_error ((GSimpleAsyncResult *) task->res, error);
+ g_simple_async_result_complete_in_idle ((GSimpleAsyncResult *) task->res);
+ extract_task_free (task);
+ g_error_free (error);
+
+ return FALSE;
+ }
+
+ g_hash_table_insert (priv->single_thread_extractors, module, async_queue);
+ }
+
+ g_async_queue_push (async_queue, task);
+ }
+ break;
+ case TRACKER_MODULE_MULTI_THREAD:
+ /* Put task in thread pool */
+ g_message ("Dispatching '%s' in thread pool", task->file);
+ g_thread_pool_push (priv->thread_pool, task, &error);
+
+ if (error) {
+ g_simple_async_result_set_from_error ((GSimpleAsyncResult *) task->res, error);
+ g_simple_async_result_complete_in_idle ((GSimpleAsyncResult *) task->res);
+ extract_task_free (task);
+ g_error_free (error);
+
+ return FALSE;
+ }
+
+ break;
+ }
+
+ return FALSE;
+}
+
/* This function can be called in any thread */
void
tracker_extract_file (TrackerExtract *extract,
@@ -475,7 +625,7 @@ tracker_extract_file (TrackerExtract *extract,
g_return_if_fail (cb != NULL);
#ifdef THREAD_ENABLE_TRACE
- g_debug ("Thread:%p (Main) <-- File:'%s' - Extracting\n",
+ g_debug ("Thread:%p <-- File:'%s' - Extracting\n",
g_thread_self (),
file);
#endif /* THREAD_ENABLE_TRACE */
@@ -483,7 +633,7 @@ tracker_extract_file (TrackerExtract *extract,
res = g_simple_async_result_new (G_OBJECT (extract), cb, user_data, NULL);
task = extract_task_new (extract, file, mimetype, cancellable, G_ASYNC_RESULT (res));
- g_idle_add (get_metadata_cb, task);
+ g_idle_add ((GSourceFunc) dispatch_task_cb, task);
/* task takes a ref */
g_object_unref (res);
@@ -497,6 +647,8 @@ tracker_extract_get_metadata_by_cmdline (TrackerExtract *object,
TrackerSparqlBuilder *statements, *preupdate;
gchar *where;
TrackerExtractPrivate *priv;
+ TrackerExtractTask *task;
+ TrackerExtractInitFunc init_func;
priv = TRACKER_EXTRACT_GET_PRIVATE (object);
priv->disable_summary_on_finalize = TRUE;
@@ -505,7 +657,16 @@ tracker_extract_get_metadata_by_cmdline (TrackerExtract *object,
g_message ("Extracting...");
- if (get_file_metadata (object, uri, mime, &preupdate, &statements, &where)) {
+ task = extract_task_new (object, uri, mime, NULL, NULL);
+
+ tracker_extract_module_manager_get_for_mimetype (task->mimetype, &init_func, NULL, &task->func);
+
+ if (init_func) {
+ /* Initialize module for this single run */
+ (init_func) (NULL, NULL);
+ }
+
+ if (get_file_metadata (task, &preupdate, &statements, &where)) {
const gchar *preupdate_str, *statements_str;
preupdate_str = statements_str = NULL;
@@ -529,4 +690,6 @@ tracker_extract_get_metadata_by_cmdline (TrackerExtract *object,
g_object_unref (preupdate);
g_free (where);
}
+
+ extract_task_free (task);
}
[
Date Prev][
Date Next] [
Thread Prev][
Thread Next]
[
Thread Index]
[
Date Index]
[
Author Index]
