[tracker/miner-fs-merge-updates] libtracker-miner: Merge several SPARQL updates while crawling

[Aleksander Morgado <aleksm src gnome org>]

commit 5d907210e6a982f4ac4db2176b9be411fd528708
Author: Aleksander Morgado <aleksander lanedo com>
Date:   Thu Sep 9 15:46:32 2010 +0200

    libtracker-miner: Merge several SPARQL updates while crawling

 src/libtracker-miner/tracker-miner-fs.c |  190 ++++++++++++++++++++++++-------
 1 files changed, 148 insertions(+), 42 deletions(-)
---
diff --git a/src/libtracker-miner/tracker-miner-fs.c b/src/libtracker-miner/tracker-miner-fs.c
index e9c0fe2..e9e7e68 100644
--- a/src/libtracker-miner/tracker-miner-fs.c
+++ b/src/libtracker-miner/tracker-miner-fs.c
@@ -38,6 +38,11 @@
 /* If defined will print contents of populated mtime cache while running */
 #undef PRINT_MTIME_CACHE_CONTENTS
 
+/* Maximum number of items of the same directory to pile up in the SPARQL buffer */
+#define MAX_SPARQL_BUFFER_ITEMS 100
+/* Maximum time (seconds) before forcing a sparql buffer flush */
+#define MAX_SPARQL_BUFFER_TIME  15
+
 /**
  * SECTION:tracker-miner-fs
  * @short_description: Abstract base class for filesystem miners
@@ -143,6 +148,12 @@ struct _TrackerMinerFSPrivate {
 	/* Files to check if no longer exist */
 	GHashTable     *check_removed;
 
+	/* SPARQL buffer to pile up several UPDATEs */
+	GString        *sparql_buffer;
+	guint           sparql_buffer_n;
+	GFile          *sparql_buffer_current_parent;
+	time_t          sparql_buffer_start_time;
+
 	/* Status */
 	guint           been_started : 1;
 	guint           been_crawled : 1;
@@ -734,6 +745,14 @@ fs_finalize (GObject *object)
 		g_list_free (priv->config_directories);
 	}
 
+	if (priv->sparql_buffer_current_parent) {
+		g_object_unref (priv->sparql_buffer_current_parent);
+	}
+
+	if (priv->sparql_buffer) {
+		g_string_free (priv->sparql_buffer, TRUE);
+	}
+
 	g_queue_foreach (priv->crawled_directories, (GFunc) crawled_directory_data_free, NULL);
 	g_queue_free (priv->crawled_directories);
 
@@ -1059,6 +1078,10 @@ sparql_update_cb (GObject      *object,
 
 	tracker_sparql_connection_update_finish (TRACKER_SPARQL_CONNECTION (object), result, &error);
 
+	/* If update was done when crawling finished, no data will be given */
+	if (!user_data)
+		return;
+
 	data = user_data;
 	fs = TRACKER_MINER_FS (data->miner);
 	priv = fs->private;
@@ -1509,12 +1532,47 @@ do_process_file (TrackerMinerFS *fs,
 }
 
 static void
+sparql_buffer_flush (TrackerMinerFS *fs,
+                     ProcessData    *data)
+{
+	TrackerMinerFSPrivate *priv;
+
+	priv = fs->private;
+
+	if (priv->sparql_buffer) {
+		g_debug ("(Sparql buffer) Flushing buffer with '%u' items", priv->sparql_buffer_n);
+		tracker_sparql_connection_update_async (tracker_miner_get_connection (TRACKER_MINER (fs)),
+		                                        priv->sparql_buffer->str,
+		                                        G_PRIORITY_DEFAULT,
+		                                        NULL,
+		                                        sparql_update_cb,
+		                                        data);
+
+		/* Clear current parent */
+		if (priv->sparql_buffer_current_parent) {
+			g_object_unref (priv->sparql_buffer_current_parent);
+			priv->sparql_buffer_current_parent = NULL;
+		}
+		/* Clear buffer */
+		g_string_free (priv->sparql_buffer, TRUE);
+		priv->sparql_buffer = NULL;
+		priv->sparql_buffer_n = 0;
+		priv->sparql_buffer_start_time = 0;
+	} else {
+		g_debug ("(Sparql buffer) Nothing to flush");
+	}
+}
+
+static void
 item_add_or_update_cb (TrackerMinerFS *fs,
                        ProcessData    *data,
                        const GError   *error)
 {
+	TrackerMinerFSPrivate *priv;
 	gchar *uri;
+	GFile *parent;
 
+	priv = fs->private;
 	uri = g_file_get_uri (data->file);
 
 	if (error) {
@@ -1534,7 +1592,8 @@ item_add_or_update_cb (TrackerMinerFS *fs,
 		    (error->code == DBUS_GERROR_NO_REPLY ||
 		     error->code == DBUS_GERROR_TIMEOUT ||
 		     error->code == DBUS_GERROR_TIMED_OUT)) {
-			g_debug ("  Got DBus timeout error on '%s', but it could not be caused by it. Retrying file.", uri);
+			g_debug ("  Got DBus timeout error on '%s', but it could not "
+			         "be caused by it. Retrying file.", uri);
 
 			/* Reset the TrackerSparqlBuilder */
 			g_object_unref (data->builder);
@@ -1551,50 +1610,85 @@ item_add_or_update_cb (TrackerMinerFS *fs,
 
 			item_queue_handlers_set_up (fs);
 		}
-	} else {
-		gchar *full_sparql;
-
-		if (data->urn) {
-			gboolean attribute_update_only;
-
-			attribute_update_only = GPOINTER_TO_INT (g_object_steal_qdata (G_OBJECT (data->file),
-			                                                               fs->private->quark_attribute_updated));
-			g_debug ("Updating item '%s' with urn '%s'%s",
-			         uri,
-			         data->urn,
-			         attribute_update_only ? " (attributes only)" : "");
-
-			if (!attribute_update_only) {
-				/* update, delete all statements inserted by miner
-				 * except for rdf:type statements as they could cause implicit deletion of user data */
-				full_sparql = g_strdup_printf ("DELETE { GRAPH <%s> { <%s> ?p ?o } } "
-				                               "WHERE { GRAPH <%s> { <%s> ?p ?o FILTER (?p != rdf:type) } } %s",
-				                               TRACKER_MINER_FS_GRAPH_URN,
-				                               data->urn,
-				                               TRACKER_MINER_FS_GRAPH_URN,
-				                               data->urn,
-				                               tracker_sparql_builder_get_result (data->builder));
-			} else {
-				/* Do not drop graph if only updating attributes, the SPARQL builder
-				 * will already contain the necessary DELETE statements for the properties
-				 * being updated */
-				full_sparql = g_strdup (tracker_sparql_builder_get_result (data->builder));
-			}
-		} else {
-			g_debug ("Creating new item '%s'", uri);
+		g_free (uri);
+		return;
+	}
 
-			/* new file */
-			full_sparql = g_strdup (tracker_sparql_builder_get_result (data->builder));
+	/* Get parent of this file we're updating/creating */
+	parent = g_file_get_parent (data->file);
+
+	/* Update count of items in buffer */
+	priv->sparql_buffer_n++;
+
+	/* Start buffer if it not already done */
+	if (!priv->sparql_buffer) {
+		priv->sparql_buffer = g_string_new ("");
+		priv->sparql_buffer_start_time = time (NULL);
+	}
+
+	/* Append new SPARQL to buffer */
+	if (data->urn) {
+		gboolean attribute_update_only;
+
+		attribute_update_only = GPOINTER_TO_INT (g_object_steal_qdata (G_OBJECT (data->file),
+		                                                               fs->private->quark_attribute_updated));
+		g_debug ("(Sparql buffer) Appending item update: '%s' with urn '%s'%s",
+		         uri,
+		         data->urn,
+		         attribute_update_only ? " (attributes only)" : "");
+
+		if (!attribute_update_only) {
+			/* update, delete all statements inserted by miner
+			 * except for rdf:type statements as they could cause implicit deletion of user data */
+			g_string_append_printf (priv->sparql_buffer,
+			                        "DELETE { GRAPH <%s> { <%s> ?p ?o } } "
+			                        "WHERE { GRAPH <%s> { <%s> ?p ?o FILTER (?p != rdf:type) } } %s",
+			                        TRACKER_MINER_FS_GRAPH_URN,
+			                        data->urn,
+			                        TRACKER_MINER_FS_GRAPH_URN,
+			                        data->urn,
+			                        tracker_sparql_builder_get_result (data->builder));
+		} else {
+			/* Do not drop graph if only updating attributes, the SPARQL builder
+			 * will already contain the necessary DELETE statements for the properties
+			 * being updated */
+			g_string_append_printf (priv->sparql_buffer,
+			                        tracker_sparql_builder_get_result (data->builder));
 		}
+	} else {
+		/* new file */
+		g_debug ("(Sparql buffer) Appending item creation: '%s'", uri);
+		g_string_append_printf (priv->sparql_buffer,
+		                        " %s",
+		                        tracker_sparql_builder_get_result (data->builder));
+	}
 
-		tracker_sparql_connection_update_async (tracker_miner_get_connection (TRACKER_MINER (fs)),
-		                                        full_sparql,
-		                                        G_PRIORITY_DEFAULT,
-		                                        NULL,
-		                                        sparql_update_cb, data);
-		g_free (full_sparql);
+	/* Set current parent if not set already */
+	if (!priv->sparql_buffer_current_parent && parent) {
+		priv->sparql_buffer_current_parent = g_object_ref (parent);
+	}
+
+	/* Flush buffer if:
+	 *  - Last item has no parent
+	 *  - Parent change was detected
+	 *  - MAX_SPARQL_BUFFER_ITEMS items reached
+	 *  - Not flushed in the last MAX_SPARQL_BUFFER_TIME seconds
+	 */
+	if (!parent ||
+	    !g_file_equal (parent, priv->sparql_buffer_current_parent) ||
+	    priv->sparql_buffer_n >= MAX_SPARQL_BUFFER_ITEMS ||
+	    (time (NULL) - priv->sparql_buffer_start_time > MAX_SPARQL_BUFFER_TIME)) {
+		/* Flush! */
+		sparql_buffer_flush (fs, data);
+	} else {
+		/* If we didn't do the flush after the append, remove from processing data */
+		fs->private->processing_pool = g_list_remove (fs->private->processing_pool, data);
+		process_data_free (data);
+		item_queue_handlers_set_up (fs);
 	}
 
+	if (parent)
+		g_object_unref (parent);
 	g_free (uri);
 }
 
@@ -1699,6 +1793,9 @@ item_remove (TrackerMinerFS *fs,
 		return TRUE;
 	}
 
+	/* Before dumping the DELETE event, flush any possible CREATED/UPDATED event */
+	sparql_buffer_flush (fs, NULL);
+
 	tracker_thumbnailer_remove_add (uri, mime);
 
 	g_free (mime);
@@ -1731,7 +1828,8 @@ item_remove (TrackerMinerFS *fs,
 	                                        sparql->str,
 	                                        G_PRIORITY_DEFAULT,
 	                                        NULL,
-	                                        sparql_update_cb, data);
+	                                        sparql_update_cb,
+	                                        data);
 
 	g_string_free (sparql, TRUE);
 	g_free (uri);
@@ -1984,6 +2082,9 @@ item_move (TrackerMinerFS *fs,
 		return retval;
 	}
 
+	/* Before dumping the MOVE event, flush any possible CREATED/UPDATED event */
+	sparql_buffer_flush (fs, NULL);
+
 	g_debug ("Moving item from '%s' to '%s'",
 	         source_uri,
 	         uri);
@@ -2047,7 +2148,8 @@ item_move (TrackerMinerFS *fs,
 	                                        sparql->str,
 	                                        G_PRIORITY_DEFAULT,
 	                                        NULL,
-	                                        sparql_update_cb, data);
+	                                        sparql_update_cb,
+	                                        data);
 
 	g_free (uri);
 	g_free (source_uri);
@@ -2449,6 +2551,10 @@ item_queue_handlers_cb (gpointer user_data)
 			process_stop (fs);
 		}
 
+		/* Flush any possible pending update here */
+		g_debug ("(Sparql buffer) No more items in queue, flushing...");
+		sparql_buffer_flush (fs, NULL);
+
 		tracker_thumbnailer_send ();
 		/* No more files left to process */
 		keep_processing = FALSE;