[tracker/wip/removable-device-completed: 8/11] Emit DeviceCompleted when finished mining a removable device

[Sam Thursfield <sthursfield src gnome org>]

commit 2b4cfc7ca012064f19562ed732389889bb9d97a8
Author: Sam Thursfield <sam thursfield codethink co uk>
Date:   Wed May 16 19:31:41 2012 +0900

    Emit DeviceCompleted when finished mining a removable device

 src/libtracker-miner/tracker-file-notifier.c    |  138 ++++++++++++++++++++---
 src/libtracker-miner/tracker-miner-fs.c         |   60 ++++++++++
 src/libtracker-miner/tracker-miner-fs.h         |    4 +
 src/libtracker-miner/tracker-miner-object.c     |   25 ++++
 src/libtracker-miner/tracker-miner-object.h     |    1 +
 src/libtracker-miner/tracker-removable-device.c |   30 +++++
 src/libtracker-miner/tracker-removable-device.h |    3 +
 src/libtracker-miner/tracker-sparql-buffer.c    |   12 ++
 src/libtracker-miner/tracker-sparql-buffer.h    |    2 +
 src/miners/fs/tracker-miner-files.c             |   13 ++
 10 files changed, 271 insertions(+), 17 deletions(-)
---
diff --git a/src/libtracker-miner/tracker-file-notifier.c b/src/libtracker-miner/tracker-file-notifier.c
index f877ded..349f4f7 100644
--- a/src/libtracker-miner/tracker-file-notifier.c
+++ b/src/libtracker-miner/tracker-file-notifier.c
@@ -29,12 +29,15 @@
 #include "tracker-crawler.h"
 #include "tracker-monitor.h"
 #include "tracker-marshal.h"
+#include "tracker-removable-device.h"
+#include "tracker-storage.h"
 
 static GQuark quark_property_crawled = 0;
 static GQuark quark_property_queried = 0;
 static GQuark quark_property_iri = 0;
 static GQuark quark_property_store_mtime = 0;
 static GQuark quark_property_filesystem_mtime = 0;
+static GQuark quark_property_removable_device = 0;
 
 enum {
 	PROP_0,
@@ -76,6 +79,8 @@ typedef struct {
 
 typedef struct {
 	TrackerFileNotifier *notifier;
+	GSList *mounts;
+
 	GNode *cur_parent_node;
 
 	/* Canonical copy from priv->file_system */
@@ -215,6 +220,7 @@ file_notifier_traverse_tree_foreach (GFile    *file,
 	TrackerFileNotifier *notifier;
 	TrackerFileNotifierPrivate *priv;
 	guint64 *store_mtime, *disk_mtime;
+	TrackerRemovableDevice *removable_device;
 
 	notifier = user_data;
 	priv = notifier->priv;
@@ -236,19 +242,28 @@ file_notifier_traverse_tree_foreach (GFile    *file,
 	           abs (*disk_mtime - *store_mtime) > 2) {
 		/* Mtime changed, update */
 		g_signal_emit (notifier, signals[FILE_UPDATED], 0, file, FALSE);
-	} else if (!store_mtime && !disk_mtime) {
-		/* what are we doing with such file? should happen rarely,
-		 * only with files that we've queried, but we decided not
-		 * to crawl (i.e. embedded root directories, that would
-		 * be processed when that root is being crawled).
-		 */
-		if (!tracker_indexing_tree_file_is_root (priv->indexing_tree, file)) {
-			gchar *uri;
+	} else {
+		/* No change, update tally if on a removable device */
+		removable_device = g_object_get_data (G_OBJECT (file),
+		                                      "tracker-removable-device");
 
-			uri = g_file_get_uri (file);
-			g_critical ("File '%s' has no disk nor store mtime",
-			            uri);
-			g_free (uri);
+		if (removable_device != NULL) {
+			tracker_removable_device_file_notify (removable_device, file);
+		}
+
+		if (!store_mtime && !disk_mtime) {
+			/* what are we doing with such file? should happen rarely,
+			 * only with files that we've queried, but we decided not
+			 * to crawl (i.e. embedded root directories, that would
+			 * be processed when that root is being crawled).
+			 */
+			if (!tracker_indexing_tree_file_is_root (priv->indexing_tree, file)) {
+				gchar *uri;
+
+				uri = g_file_get_uri (file);
+				g_critical ("File '%s' has no disk nor store mtime", uri);
+				g_free (uri);
+			}
 		}
 	}
 
@@ -309,15 +324,39 @@ file_notifier_add_node_foreach (GNode    *node,
 	TrackerFileNotifierPrivate *priv;
 	GFileInfo *file_info;
 	GFile *canonical, *file;
+	TrackerRemovableDevice *removable_device = NULL;
+	GSList *l;
 
 	priv = data->notifier->priv;
 	file = node->data;
 
-	if (node->parent &&
-	    node->parent != data->cur_parent_node) {
-		data->cur_parent_node = node->parent;
-		data->cur_parent = tracker_file_system_peek_file (priv->file_system,
-		                                                  node->parent->data);
+	if (data->cur_parent != NULL) {
+		removable_device = tracker_file_system_get_property (priv->file_system,
+		                                                     data->cur_parent,
+		                                                     quark_property_removable_device);
+	}
+
+	if (node->parent != data->cur_parent_node || node->parent == NULL) {
+		/* We are at a directory */
+
+		if (removable_device == NULL) {
+			/* Is this directory a mount point for a removable device? */
+			for (l = data->mounts; l != NULL; l = l->next) {
+				TrackerRemovableDevice *possible_removable_device;
+				possible_removable_device = TRACKER_REMOVABLE_DEVICE (l->data);
+				/* FIXME: memory leak */
+				if (g_file_equal (file, g_mount_get_root (tracker_removable_device_get_mount (possible_removable_device)))) {
+					removable_device = possible_removable_device;
+					break;
+				}
+			}
+		}
+
+		if (node->parent != NULL) {
+			data->cur_parent_node = node->parent;
+			data->cur_parent = tracker_file_system_peek_file (priv->file_system,
+			                                                  node->parent->data);
+		}
 	}
 
 	file_info = tracker_crawler_get_file_info (priv->crawler, file);
@@ -325,6 +364,7 @@ file_notifier_add_node_foreach (GNode    *node,
 	if (file_info) {
 		GFileType file_type;
 		guint64 time, *time_ptr;
+		gint removable_device_file_count;
 
 		file_type = g_file_info_get_file_type (file_info);
 
@@ -342,6 +382,27 @@ file_notifier_add_node_foreach (GNode    *node,
 		tracker_file_system_set_property (priv->file_system, canonical,
 		                                  quark_property_filesystem_mtime,
 		                                  time_ptr);
+
+		if (removable_device) {
+			tracker_file_system_set_property (priv->file_system, canonical,
+			                                  quark_property_removable_device,
+			                                  removable_device);
+			/* TrackerMinerFS can't access the above properties easily */
+			g_object_set_data_full (G_OBJECT (canonical),
+			                        "tracker-removable-device",
+			                        g_object_ref (removable_device),
+			                        g_object_unref);
+
+			removable_device_file_count = GPOINTER_TO_INT
+			  (g_object_get_data (G_OBJECT (removable_device), "tracker-files-found"));
+
+			g_object_set_data (G_OBJECT (removable_device),
+			                   "tracker-files-found",
+			                   GINT_TO_POINTER (++ removable_device_file_count));
+		}
+	} else {
+		/* FIXME: an error if this doesn't happen? */
+		g_warn_if_reached ();
 	}
 
 	return FALSE;
@@ -359,8 +420,45 @@ crawler_directory_crawled_cb (TrackerCrawler *crawler,
 {
 	TrackerFileNotifier *notifier;
 	DirectoryCrawledData data = { 0 };
+	GSList *all_mounts;
+	GSList *child_mounts;
+	GSList *l;
+
+	/* Don't need to detect here (could be passed by miner-files, because
+	 * there's no actual reason to index a mount point except by its mount
+	 * notification)  */
+
+	/* Find all the mounts beneath this directory. While mounts can't be
+	 * nested (this is ony supposed to be for removable devices), we could have
+	 * just crawled /media, for example, so there may be more than one.
+	 */
+
+	all_mounts = tracker_storage_get_devices (tracker_storage_get (),
+	                                          TRACKER_STORAGE_REMOVABLE,
+	                                          FALSE);
+	child_mounts = NULL;
+
+	for (l = all_mounts; l != NULL; l = l->next) {
+		TrackerRemovableDevice *device;
+		GMount *mount;
+		GFile *mount_point;
+
+		device = TRACKER_REMOVABLE_DEVICE (l->data);
+		mount = tracker_removable_device_get_mount (device);
+		mount_point = g_mount_get_root (mount);
+
+		if (g_file_equal (mount_point, directory) ||
+		    g_file_has_prefix (mount_point, directory)) {
+			child_mounts = g_slist_prepend (child_mounts, device);
+		}
+
+		g_object_unref (mount_point);
+	}
+
+	g_slist_free (all_mounts);
 
 	notifier = data.notifier = user_data;
+	data.mounts = child_mounts;
 	g_node_traverse (tree,
 	                 G_PRE_ORDER,
 	                 G_TRAVERSE_ALL,
@@ -379,6 +477,8 @@ crawler_directory_crawled_cb (TrackerCrawler *crawler,
 	tracker_info ("  Found %d files, ignored %d files",
 	              files_found,
 	              files_ignored);
+
+	g_slist_free (data.mounts);
 }
 
 static void
@@ -1238,6 +1338,10 @@ tracker_file_notifier_class_init (TrackerFileNotifierClass *klass)
 	quark_property_filesystem_mtime = g_quark_from_static_string ("tracker-property-filesystem-mtime");
 	tracker_file_system_register_property (quark_property_filesystem_mtime,
 	                                       g_free);
+
+	quark_property_removable_device = g_quark_from_static_string ("tracker-property-removable-device");
+	tracker_file_system_register_property (quark_property_removable_device,
+	                                       NULL);
 }
 
 static void
diff --git a/src/libtracker-miner/tracker-miner-fs.c b/src/libtracker-miner/tracker-miner-fs.c
index a443895..3cbca21 100644
--- a/src/libtracker-miner/tracker-miner-fs.c
+++ b/src/libtracker-miner/tracker-miner-fs.c
@@ -36,6 +36,7 @@
 #include "tracker-task-pool.h"
 #include "tracker-sparql-buffer.h"
 #include "tracker-file-notifier.h"
+#include "tracker-storage.h"
 
 /* If defined will print the tree from GNode while running */
 #ifdef CRAWLED_TREE_ENABLE_TRACE
@@ -182,6 +183,9 @@ struct _TrackerMinerFSPrivate {
 
 	TrackerIndexingTree *indexing_tree;
 
+	/* Signals waiting for SPARQL buffer to empty */
+	GSList *completed_devices;
+
 	/* Status */
 	guint           been_started : 1;     /* TRUE if miner has been started */
 	guint           been_crawled : 1;     /* TRUE if initial crawling has been
@@ -1003,6 +1007,18 @@ sparql_buffer_task_finished_cb (GObject      *object,
 		g_error_free (error);
 	}
 
+	if (tracker_sparql_buffer_get_n_outstanding_updates (TRACKER_SPARQL_BUFFER (object)) == 0) {
+		/* Device completed notifications are deferred until all the data is flushed */
+		while (priv->completed_devices != NULL) {
+			tracker_miner_device_completed (TRACKER_MINER (fs));
+
+			priv->completed_devices = priv->completed_devices->next;
+		}
+	}
+
+	/* FIXME: can be moved into the block above (noop really since we don't
+	 * do simultaneous updates
+	 */
 	item_queue_handlers_set_up (fs);
 }
 
@@ -1116,6 +1132,7 @@ item_add_or_update_cb (TrackerMinerFS *fs,
 {
 	UpdateProcessingTaskContext *ctxt;
 	TrackerTask *sparql_task = NULL;
+	TrackerRemovableDevice *device;
 	GFile *task_file;
 	gchar *uri;
 
@@ -1203,6 +1220,16 @@ item_add_or_update_cb (TrackerMinerFS *fs,
 		}
 	}
 
+	/* Notify device after the SPARQL task is pushed to the buffer. That way,
+	 * if this is the last file, we can be certain all of the device's data is
+	 * in the store once the SPARQL buffer has emptied.
+	 */
+	device = g_object_get_data (G_OBJECT (task_file), "tracker-removable-device");
+
+	if (device != NULL) {
+		tracker_removable_device_file_notify (TRACKER_REMOVABLE_DEVICE (device), task_file);
+	}
+
 	if (tracker_miner_fs_has_items_to_process (fs) == FALSE &&
 	    tracker_task_pool_get_size (TRACKER_TASK_POOL (fs->priv->task_pool)) == 0) {
 		/* We need to run this one more time to trigger process_stop() */
@@ -3690,6 +3717,39 @@ tracker_miner_fs_get_indexing_tree (TrackerMinerFS *fs)
 	return fs->priv->indexing_tree;
 }
 
+/**
+ * tracker_miner_fs_device_completed:
+ * @fs: a #TrackerMinerFS
+ * @device: a #TrackerRemovableDevice
+ *
+ * A version of tracker_miner_device_completed() which avoids emitting the
+ * signal until the SPARQL buffer has been flushed. This prevents a race
+ * condition where an application receives the notification and queries the
+ * device contents before it is completely commited into the store.
+ **/
+/* This problem actually goes a lot deeper. On closed systems where contents of
+ * a removable device can't change, problem is solved, but extra steps are
+ * needed if you wish to connect to GraphUpdated because you need to be able
+ * to receive updates from the moment device-completed was emitted, but you
+ * don't know to connect to that signal until you receive it, which is a race
+ * condition. If you connect to GraphUpdated straight away. .. actually, you
+ * can just CONNECT to GraphUpdated but ignore the signals until your receive
+ * DeviceCompleted, and THEN go ... which can be atomic. So, yay.
+ */
+void
+tracker_miner_fs_device_completed (TrackerMinerFS         *fs,
+                                   TrackerRemovableDevice *device)
+{
+	g_return_if_fail (TRACKER_IS_MINER_FS (fs));
+
+	if (tracker_task_pool_get_size (TRACKER_TASK_POOL (fs->priv->sparql_buffer)) == 0 &&
+	    tracker_sparql_buffer_get_n_outstanding_updates (fs->priv->sparql_buffer) == 0) {
+		tracker_miner_device_completed (TRACKER_MINER (fs));
+	} else {
+		fs->priv->completed_devices = g_slist_prepend (fs->priv->completed_devices, device);
+	}
+}
+
 #ifdef EVENT_QUEUE_ENABLE_TRACE
 
 static void
diff --git a/src/libtracker-miner/tracker-miner-fs.h b/src/libtracker-miner/tracker-miner-fs.h
index f4642a1..887eeb0 100644
--- a/src/libtracker-miner/tracker-miner-fs.h
+++ b/src/libtracker-miner/tracker-miner-fs.h
@@ -31,6 +31,7 @@
 
 #include "tracker-miner-object.h"
 #include "tracker-indexing-tree.h"
+#include "tracker-removable-device.h"
 
 #include "tracker-miner-common.h"
 
@@ -152,6 +153,9 @@ void                  tracker_miner_fs_force_mtime_checking (TrackerMinerFS *fs,
 
 TrackerIndexingTree * tracker_miner_fs_get_indexing_tree    (TrackerMinerFS *fs);
 
+void                  tracker_miner_fs_device_completed     (TrackerMinerFS         *fs,
+                                                             TrackerRemovableDevice *device);
+
 G_END_DECLS
 
 #endif /* __LIBTRACKER_MINER_MINER_FS_H__ */
diff --git a/src/libtracker-miner/tracker-miner-object.c b/src/libtracker-miner/tracker-miner-object.c
index 09a2b80..d055c60 100644
--- a/src/libtracker-miner/tracker-miner-object.c
+++ b/src/libtracker-miner/tracker-miner-object.c
@@ -109,6 +109,7 @@ static const gchar introspection_xml[] =
   "      <arg type='s' name='status' />"
   "      <arg type='d' name='progress' />"
   "    </signal>"
+  "    <signal name='DeviceCompleted' />"
   "  </interface>"
   "</node>";
 
@@ -1075,6 +1076,30 @@ tracker_miner_resume (TrackerMiner  *miner,
 }
 
 /**
+ * tracker_miner_device_completed:
+ * @miner: a #TrackerMiner
+ *
+ * Emits the DeviceCompleted signal. FIXME: document the signal.
+ *
+ * Since: 0.14.2
+ **/
+void
+tracker_miner_device_completed (TrackerMiner *miner)
+{
+	g_return_if_fail (TRACKER_IS_MINER (miner));
+
+	if (miner->priv->d_connection) {
+		g_dbus_connection_emit_signal (miner->priv->d_connection,
+		                               NULL,
+		                               miner->priv->full_path,
+		                               TRACKER_MINER_DBUS_INTERFACE,
+		                               "DeviceCompleted",
+		                               NULL,
+		                               NULL);
+	}
+}
+
+/**
  * tracker_miner_get_connection:
  * @miner: a #TrackerMiner
  *
diff --git a/src/libtracker-miner/tracker-miner-object.h b/src/libtracker-miner/tracker-miner-object.h
index 611917d..d58a5b8 100644
--- a/src/libtracker-miner/tracker-miner-object.h
+++ b/src/libtracker-miner/tracker-miner-object.h
@@ -102,6 +102,7 @@ gint                     tracker_miner_pause               (TrackerMiner
 gboolean                 tracker_miner_resume              (TrackerMiner         *miner,
                                                             gint                  cookie,
                                                             GError              **error);
+void                     tracker_miner_device_completed    (TrackerMiner         *miner);
 
 TrackerSparqlConnection *tracker_miner_get_connection      (TrackerMiner         *miner);
 GDBusConnection         *tracker_miner_get_dbus_connection (TrackerMiner         *miner);
diff --git a/src/libtracker-miner/tracker-removable-device.c b/src/libtracker-miner/tracker-removable-device.c
index aad568a..8c1a4dc 100644
--- a/src/libtracker-miner/tracker-removable-device.c
+++ b/src/libtracker-miner/tracker-removable-device.c
@@ -157,3 +157,33 @@ tracker_removable_device_get_mount_point (TrackerRemovableDevice *device)
 
 	return g_mount_get_root (device->priv->mount);
 }
+/**
+ * tracker_removable_device_file_notify:
+ * @device: a #TrackerRemovableDevice
+ * @file: a #GFile
+ *
+ * Increases the files completed count of @device. If all files have been
+ * processed, this function will cause the ::mining-complete signal to be
+ * emitted.
+ */
+void
+tracker_removable_device_file_notify (TrackerRemovableDevice *device,
+                                      GFile                  *file)
+{
+	gint files_found;
+	gint files_processed;
+
+	files_found = GPOINTER_TO_INT (g_object_get_data (G_OBJECT (device), "tracker-files-found"));
+	files_processed = GPOINTER_TO_INT (g_object_get_data (G_OBJECT (device), "tracker-files-processed"));
+
+	g_object_set_data (G_OBJECT (device), "tracker-files-processed", GINT_TO_POINTER (++ files_processed));
+
+	g_warn_if_fail (files_found >= files_processed);
+
+	if (files_found == files_processed) {
+		g_signal_emit (device, signals[MINING_COMPLETE], 0);
+
+		g_object_set_data (G_OBJECT (device), "tracker-files-found", NULL);
+		g_object_set_data (G_OBJECT (device), "tracker-files-processed", NULL);
+	}
+}
diff --git a/src/libtracker-miner/tracker-removable-device.h b/src/libtracker-miner/tracker-removable-device.h
index b3a727b..fea1a8b 100644
--- a/src/libtracker-miner/tracker-removable-device.h
+++ b/src/libtracker-miner/tracker-removable-device.h
@@ -68,6 +68,9 @@ TrackerRemovableDevice *tracker_removable_device_new                      (GMoun
 GMount                 *tracker_removable_device_get_mount                (TrackerRemovableDevice *device);
 GFile                  *tracker_removable_device_get_mount_point          (TrackerRemovableDevice *device);
 
+void                    tracker_removable_device_file_notify              (TrackerRemovableDevice *device,
+                                                                           GFile                  *file);
+
 G_END_DECLS
 
 #endif /* __TRACKER_REMOVABLE_DEVICE_H__ */
diff --git a/src/libtracker-miner/tracker-sparql-buffer.c b/src/libtracker-miner/tracker-sparql-buffer.c
index 33e8601..e98f30d 100644
--- a/src/libtracker-miner/tracker-sparql-buffer.c
+++ b/src/libtracker-miner/tracker-sparql-buffer.c
@@ -714,6 +714,18 @@ tracker_sparql_buffer_push (TrackerSparqlBuffer *buffer,
 	}
 }
 
+gint
+tracker_sparql_buffer_get_n_outstanding_updates (TrackerSparqlBuffer *buffer)
+{
+	TrackerSparqlBufferPrivate *priv;
+
+	g_return_val_if_fail (TRACKER_IS_SPARQL_BUFFER (buffer), 0);
+
+	priv = buffer->priv;
+
+	return priv->n_updates;
+}
+
 static SparqlTaskData *
 sparql_task_data_new (guint    type,
                       gpointer data,
diff --git a/src/libtracker-miner/tracker-sparql-buffer.h b/src/libtracker-miner/tracker-sparql-buffer.h
index 7cb5f0b..c2b9421 100644
--- a/src/libtracker-miner/tracker-sparql-buffer.h
+++ b/src/libtracker-miner/tracker-sparql-buffer.h
@@ -73,6 +73,8 @@ void                 tracker_sparql_buffer_push  (TrackerSparqlBuffer *buffer,
                                                   GAsyncReadyCallback  cb,
                                                   gpointer             user_data);
 
+gint                 tracker_sparql_buffer_get_n_outstanding_updates (TrackerSparqlBuffer *buffer);
+
 TrackerTask *        tracker_sparql_task_new_take_sparql_str (GFile                *file,
                                                               gchar                *sparql_str);
 TrackerTask *        tracker_sparql_task_new_with_sparql_str (GFile                *file,
diff --git a/src/miners/fs/tracker-miner-files.c b/src/miners/fs/tracker-miner-files.c
index b5322b6..92864e6 100644
--- a/src/miners/fs/tracker-miner-files.c
+++ b/src/miners/fs/tracker-miner-files.c
@@ -1440,6 +1440,14 @@ miner_finished_cb (TrackerMinerFS *fs,
 #endif /* defined(HAVE_UPOWER) || defined(HAVE_HAL) */
 }
 
+/* Called when all files on a certain device have been crawled. */
+static void
+device_mining_complete_cb (TrackerRemovableDevice *device,
+                           TrackerMinerFiles      *mf)
+{
+	tracker_miner_fs_device_completed (TRACKER_MINER_FS (mf), device);
+}
+
 static void
 mount_pre_unmount_cb (GVolumeMonitor    *volume_monitor,
                       GMount            *mount,
@@ -2991,6 +2999,11 @@ miner_files_add_removable_or_optical_device (TrackerMinerFiles      *mf,
 	                         g_strdup (uuid),
 	                         (GDestroyNotify) g_free);
 
+	g_signal_connect (device,
+	                  "mining-complete",
+	                  G_CALLBACK (device_mining_complete_cb),
+	                  mf);
+
 	g_message ("  Adding removable/optical: '%s'", g_file_get_path (mount_point_file));
 	tracker_indexing_tree_add (indexing_tree,
 				   mount_point_file,