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

[Aleksander Morgado <aleksm src gnome org>]

commit 6968eb9a818b5f2341c0da5888cfcd399f5aa75d
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 |  148 ++++++++++++++++++++++++++-----
 1 files changed, 126 insertions(+), 22 deletions(-)
---
diff --git a/src/libtracker-miner/tracker-miner-fs.c b/src/libtracker-miner/tracker-miner-fs.c
index a00f583..0d3db28 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
@@ -142,6 +147,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;
@@ -690,6 +701,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);
 
@@ -1015,6 +1034,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;
@@ -1455,12 +1478,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) {
@@ -1497,32 +1555,66 @@ item_add_or_update_cb (TrackerMinerFS *fs,
 
 			item_queue_handlers_set_up (fs);
 		}
-	} else {
-		gchar *full_sparql;
 
-		if (data->urn) {
-			g_debug ("Updating item '%s' with urn '%s'", uri, data->urn);
+		g_free (uri);
+		return;
+	}
 
-			/* 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 {
-			g_debug ("Creating new item '%s'", uri);
+	/* Get parent of this file we're updating/creating */
+	parent = g_file_get_parent (data->file);
 
-			/* new file */
-			full_sparql = g_strdup (tracker_sparql_builder_get_result (data->builder));
-		}
+	/* Update count of items in buffer */
+	priv->sparql_buffer_n++;
 
-		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);
+	/* 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) {
+		/* update, delete all statements inserted by miner
+		   except for rdf:type statements as they could cause implicit deletion of user data */
+		g_debug ("(Sparql buffer) Appending item update: '%s' with urn '%s'", uri, data->urn);
+		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 {
+		/* 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));
+	}
+
+	/* 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);
 }
 
@@ -1627,6 +1719,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);
@@ -1659,7 +1754,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);
@@ -1912,6 +2008,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);
@@ -1975,7 +2074,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);
@@ -2377,6 +2477,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;