[tracker/class-signal: 3/39] tracker-store: Storage of signal events as they arrive

[Philip Van Hoof <pvanhoof src gnome org>]

commit a7776b030c6fb2ff61825fe151d23405c11a3a58
Author: Philip Van Hoof <philip codeminded be>
Date:   Tue Aug 17 15:32:28 2010 +0200

    tracker-store: Storage of signal events as they arrive

 src/tracker-store/tracker-events.c    |  268 +++++++++++++++++++++++----------
 src/tracker-store/tracker-events.h    |   24 +++-
 src/tracker-store/tracker-resources.c |   30 ++---
 3 files changed, 216 insertions(+), 106 deletions(-)
---
diff --git a/src/tracker-store/tracker-events.c b/src/tracker-store/tracker-events.c
index 583e2bf..e577469 100644
--- a/src/tracker-store/tracker-events.c
+++ b/src/tracker-store/tracker-events.c
@@ -26,108 +26,207 @@
 
 #include "tracker-events.h"
 
-#define MAX_EVENTS 1000000
-
 typedef struct {
+	GHashTable *allowances_id;
 	GHashTable *allowances;
-	GArray *events;
-	GStringChunk *chunk;
+	struct {
+		GArray *sub_pred_ids;
+		GArray *object_ids;
+	} deletes;
+	struct {
+		GArray *sub_pred_ids;
+		GArray *object_ids;
+	} inserts;
 	gboolean frozen;
 } EventsPrivate;
 
 static EventsPrivate *private;
 
 static void
-tracker_events_add_allow (const gchar *rdf_class)
+get_from_array (GArray *sub_pred_ids,
+                GArray *object_ids_in,
+                GArray *subject_ids,
+                GArray *pred_ids,
+                GArray *object_ids)
 {
-	TrackerClass *cl;
+	guint i;
 
-	g_return_if_fail (private != NULL);
+	g_array_set_size (subject_ids, 0);
+	g_array_set_size (pred_ids, 0);
+	g_array_set_size (object_ids, 0);
 
-	cl = tracker_ontologies_get_class_by_uri (rdf_class);
-	if (!cl) {
-		g_critical ("Unknown class %s", rdf_class);
-		return;
+	for (i = 0; i < sub_pred_ids->len; i++) {
+		gint subject_id, pred_id, object_id;
+		gint64 sub_pred_id;
+
+		sub_pred_id = g_array_index (sub_pred_ids, gint64, i);
+		subject_id = sub_pred_id >> 32;
+		pred_id = sub_pred_id << 32;
+		object_id = g_array_index (object_ids_in, gint, i);
+
+		g_array_append_val (subject_ids, subject_id);
+		g_array_append_val (pred_ids, pred_id);
+		g_array_append_val (object_ids, object_id);
 	}
+}
 
-	g_hash_table_insert (private->allowances, cl,
-	                     GINT_TO_POINTER (TRUE));
+void
+tracker_events_get_inserts (GArray *subject_ids,
+                            GArray *pred_ids,
+                            GArray *object_ids)
+{
+	g_return_if_fail (private != NULL);
+	g_return_if_fail (subject_ids != NULL);
+	g_return_if_fail (pred_ids != NULL);
+	g_return_if_fail (object_ids != NULL);
+
+	get_from_array (private->inserts.sub_pred_ids,
+	                private->inserts.object_ids,
+	                subject_ids,
+	                pred_ids,
+	                object_ids);
 }
 
-static gboolean
-is_allowed (EventsPrivate *private, TrackerClass *rdf_class)
+void
+tracker_events_get_deletes (GArray *subject_ids,
+                            GArray *pred_ids,
+                            GArray *object_ids)
 {
-	return (g_hash_table_lookup (private->allowances, rdf_class) != NULL) ? TRUE : FALSE;
+	g_return_if_fail (private != NULL);
+	g_return_if_fail (subject_ids != NULL);
+	g_return_if_fail (pred_ids != NULL);
+	g_return_if_fail (object_ids != NULL);
+
+	get_from_array (private->deletes.sub_pred_ids,
+	                private->deletes.object_ids,
+	                subject_ids,
+	                pred_ids,
+	                object_ids);
 }
 
-static void
-prepare_event_for_rdf_type (EventsPrivate *private,
-                            TrackerClass  *rdf_class ,
-                            const gchar *uri,
-                            TrackerDBusEventsType type,
-                            const gchar *predicate)
+static gboolean
+is_allowed (EventsPrivate *private, TrackerClass *rdf_class, gint class_id)
 {
-	TrackerEvent event;
+	gboolean ret;
 
-	if (!private->events) {
-		private->events = g_array_new (TRUE, FALSE, sizeof (TrackerEvent));
+	if (rdf_class != NULL) {
+		ret = (g_hash_table_lookup (private->allowances, rdf_class) != NULL) ? TRUE : FALSE;
+	} else {
+		ret = (g_hash_table_lookup (private->allowances_id, GINT_TO_POINTER (class_id)) != NULL) ? TRUE : FALSE;
 	}
+	return ret;
+}
 
-	if (!private->chunk) {
-		private->chunk = g_string_chunk_new (4096);
+static void
+insert_vals_into_arrays (GArray *sub_pred_ids,
+                         GArray *object_ids,
+                         gint subject_id,
+                         gint pred_id,
+                         gint object_id)
+{
+	guint i;
+	gboolean inserted = FALSE;
+	gint64 sub_pred_id = (gint64) subject_id;
+
+	sub_pred_id = sub_pred_id << 32 | pred_id;
+
+	for (i = 0; i < sub_pred_ids->len; i++) {
+		if (sub_pred_id < g_array_index (sub_pred_ids, gint64, i)) {
+			g_array_insert_val (sub_pred_ids, i, sub_pred_id);
+			g_array_insert_val (object_ids, i, object_id);
+			inserted = TRUE;
+			break;
+		}
 	}
 
-	event.type = type;
-	event.class = rdf_class;
-	event.predicate = tracker_ontologies_get_property_by_uri (predicate);
-	event.subject = g_string_chunk_insert_const (private->chunk, uri);
+	if (!inserted) {
+		g_array_append_val (sub_pred_ids, sub_pred_ids);
+		g_array_append_val (object_ids, object_id);
+	}
 
-	g_array_append_val (private->events, event);
 }
 
 void
-tracker_events_insert (const gchar *uri,
-                       const gchar *predicate,
-                       const gchar *object,
-                       GPtrArray *rdf_types,
-                       TrackerDBusEventsType type)
+tracker_events_add_insert (gint         graph_id,
+                           gint         subject_id,
+                           const gchar *subject,
+                           gint         pred_id,
+                           gint         object_id,
+                           const gchar *object,
+                           GPtrArray   *rdf_types)
 {
-	g_return_if_fail (rdf_types || type != TRACKER_DBUS_EVENTS_TYPE_UPDATE);
+	TrackerProperty *rdf_type;
+
+	g_return_if_fail (rdf_types != NULL);
 	g_return_if_fail (private != NULL);
 
 	if (private->frozen) {
 		return;
 	}
 
-	if (private->events && private->events->len > MAX_EVENTS) {
-		/* too many events would lead to high memory usage and to dbus signal messages
-		   larger than maximum dbus message size, which results in dbus disconnect and tracker-store crash */
-		g_warning ("Too many events pending, disabling events for current transaction");
-		tracker_events_reset ();
-		private->frozen = TRUE;
-	}
+	rdf_type = tracker_ontologies_get_rdf_type ();
 
-	if (rdf_types && type == TRACKER_DBUS_EVENTS_TYPE_UPDATE) {
+	if (object_id != 0 && pred_id == tracker_property_get_id (rdf_type)) {
+		/* Resource create
+		 * In case of create, object is the rdf:type */
+		if (is_allowed (private, NULL, object_id)) {
+			insert_vals_into_arrays (private->inserts.sub_pred_ids,
+			                         private->inserts.object_ids,
+			                         subject_id, pred_id, object_id);
+		}
+	} else {
 		guint i;
 
 		for (i = 0; i < rdf_types->len; i++) {
+			if (is_allowed (private, rdf_types->pdata[i], 0)) {
+				insert_vals_into_arrays (private->inserts.sub_pred_ids,
+				                         private->inserts.object_ids,
+				                         subject_id, pred_id, object_id);
+			}
+		}
+	}
+}
+
+void
+tracker_events_add_delete (gint         graph_id,
+                           gint         subject_id,
+                           const gchar *subject,
+                           gint         pred_id,
+                           gint         object_id,
+                           const gchar *object,
+                           GPtrArray   *rdf_types)
+{
+	TrackerProperty *rdf_type;
 
-			/* object is not very important for updates (we don't expose
-			 * the value being set to the user's DBus API in tracker-store) */
-			if (is_allowed (private, rdf_types->pdata[i])) {
+	g_return_if_fail (rdf_types != NULL);
+	g_return_if_fail (private != NULL);
 
-				prepare_event_for_rdf_type (private, rdf_types->pdata[i],
-				                            uri, type, predicate);
-			}
+	if (private->frozen) {
+		return;
+	}
+
+	rdf_type = tracker_ontologies_get_rdf_type ();
+
+	if (object_id != 0 && pred_id == tracker_property_get_id (rdf_type)) {
+		/* Resource delete
+		 * In case of delete, object is the rdf:type */
+		if (is_allowed (private, NULL, object_id)) {
+			insert_vals_into_arrays (private->deletes.sub_pred_ids,
+			                         private->deletes.object_ids,
+			                         subject_id, pred_id, object_id);
 		}
 	} else {
-		TrackerClass *class = tracker_ontologies_get_class_by_uri (object);
-		/* In case of delete and create, object is the rdf:type */
-		if (is_allowed (private, class)) {
-			prepare_event_for_rdf_type (private, class,
-			                            uri, type, predicate);
+		guint i;
+
+		for (i = 0; i < rdf_types->len; i++) {
+			if (is_allowed (private, rdf_types->pdata[i], 0)) {
+				insert_vals_into_arrays (private->deletes.sub_pred_ids,
+				                         private->deletes.object_ids,
+				                         subject_id, pred_id, object_id);
+			}
 		}
 	}
+
 }
 
 void
@@ -135,12 +234,11 @@ tracker_events_reset (void)
 {
 	g_return_if_fail (private != NULL);
 
-	if (private->events) {
-		g_array_free (private->events, TRUE);
-		g_string_chunk_free (private->chunk);
-		private->chunk = NULL;
-		private->events = NULL;
-	}
+	g_array_set_size (private->deletes.sub_pred_ids, 0);
+	g_array_set_size (private->deletes.object_ids, 0);
+	g_array_set_size (private->inserts.sub_pred_ids, 0);
+	g_array_set_size (private->inserts.object_ids, 0);
+
 	private->frozen = FALSE;
 }
 
@@ -152,37 +250,37 @@ tracker_events_freeze (void)
 	private->frozen = TRUE;
 }
 
-GArray *
-tracker_events_get_pending (void)
-{
-	g_return_val_if_fail (private != NULL, NULL);
-
-	return private->events;
-}
-
 static void
 free_private (EventsPrivate *private)
 {
 	g_hash_table_unref (private->allowances);
+	g_hash_table_unref (private->allowances_id);
+	g_array_free (private->deletes.sub_pred_ids, TRUE);
+	g_array_free (private->deletes.object_ids, TRUE);
+	g_array_free (private->inserts.sub_pred_ids, TRUE);
+	g_array_free (private->inserts.object_ids, TRUE);
 	g_free (private);
 }
 
 void
 tracker_events_init (TrackerNotifyClassGetter callback)
 {
-	GStrv          classes_to_signal;
-	gint           i, count;
-
-	private = g_new0 (EventsPrivate, 1);
-
-	private->allowances = g_hash_table_new (g_direct_hash, g_direct_equal);
-
-	private->events = NULL;
+	GStrv classes_to_signal;
+	gint  i, count;
 
 	if (!callback) {
 		return;
 	}
 
+	private = g_new0 (EventsPrivate, 1);
+
+	private->allowances = g_hash_table_new (g_direct_hash, g_direct_equal);
+	private->allowances_id = g_hash_table_new (g_direct_hash, g_direct_equal);
+	private->deletes.sub_pred_ids = g_array_new (FALSE, FALSE, sizeof (gint64));
+	private->deletes.object_ids = g_array_new (FALSE, FALSE, sizeof (gint));
+	private->inserts.sub_pred_ids = g_array_new (FALSE, FALSE, sizeof (gint64));
+	private->inserts.object_ids = g_array_new (FALSE, FALSE, sizeof (gint));
+
 	classes_to_signal = (*callback)();
 
 	if (!classes_to_signal)
@@ -190,9 +288,16 @@ tracker_events_init (TrackerNotifyClassGetter callback)
 
 	count = g_strv_length (classes_to_signal);
 	for (i = 0; i < count; i++) {
-		tracker_events_add_allow (classes_to_signal[i]);
+		TrackerClass *class = tracker_ontologies_get_class_by_uri (classes_to_signal[i]);
+		if (class != NULL) {
+			g_hash_table_insert (private->allowances,
+			                     class,
+			                     GINT_TO_POINTER (TRUE));
+			g_hash_table_insert (private->allowances_id,
+			                     GINT_TO_POINTER (tracker_class_get_id (class)),
+			                     GINT_TO_POINTER (TRUE));
+		}
 	}
-
 	g_strfreev (classes_to_signal);
 }
 
@@ -200,7 +305,6 @@ void
 tracker_events_shutdown (void)
 {
 	if (private != NULL) {
-		tracker_events_reset ();
 		free_private (private);
 		private = NULL;
 	} else {
diff --git a/src/tracker-store/tracker-events.h b/src/tracker-store/tracker-events.h
index 981233c..cf8ed49 100644
--- a/src/tracker-store/tracker-events.h
+++ b/src/tracker-store/tracker-events.h
@@ -41,12 +41,26 @@ typedef GStrv (*TrackerNotifyClassGetter) (void);
 
 void       tracker_events_init        (TrackerNotifyClassGetter  callback);
 void       tracker_events_shutdown    (void);
-void       tracker_events_insert      (const gchar              *uri,
-                                       const gchar              *predicate,
+void       tracker_events_add_insert  (gint                      graph_id,
+                                       gint                      subject_id,
+                                       const gchar              *subject,
+                                       gint                      pred_id,
+                                       gint                      object_id,
                                        const gchar              *object,
-                                       GPtrArray                *rdf_types,
-                                       TrackerDBusEventsType     type);
-GArray    *tracker_events_get_pending (void);
+                                       GPtrArray                *rdf_types);
+void       tracker_events_add_delete  (gint                      graph_id,
+                                       gint                      subject_id,
+                                       const gchar              *subject,
+                                       gint                      pred_id,
+                                       gint                      object_id,
+                                       const gchar              *object,
+                                       GPtrArray                *rdf_types);
+void       tracker_events_get_inserts (GArray                   *subject_ids,
+                                       GArray                   *pred_ids,
+                                       GArray                   *object_ids);
+void       tracker_events_get_deletes (GArray                   *subject_ids,
+                                       GArray                   *pred_ids,
+                                       GArray                   *object_ids);
 void       tracker_events_reset       (void);
 void       tracker_events_freeze      (void);
 
diff --git a/src/tracker-store/tracker-resources.c b/src/tracker-store/tracker-resources.c
index 8052bce..a3489d7 100644
--- a/src/tracker-store/tracker-resources.c
+++ b/src/tracker-store/tracker-resources.c
@@ -527,17 +527,17 @@ static void
 on_statements_committed (gpointer user_data)
 {
 	TrackerResources *resources = user_data;
-	GArray *events;
+	/* GArray *events; */
 	GHashTable *writebacks;
 	TrackerResourcesPrivate *priv;
 
 	priv = TRACKER_RESOURCES_GET_PRIVATE (resources);
 
-	/* Class signals feature */
+	/* Class signals feature *
 	events = tracker_events_get_pending ();
 
-	/* Do not call tracker_events_reset before calling tracker_resource_class_emit_events
-	   as we're reusing the same strings without copies */
+	* Do not call tracker_events_reset before calling tracker_resource_class_emit_events
+	   as we're reusing the same strings without copies *
 
 	if (events) {
 		GSList *event_sources, *l;
@@ -577,6 +577,7 @@ on_statements_committed (gpointer user_data)
 			g_hash_table_destroy (to_emit);
 		}
 	}
+	*/
 
 	tracker_events_reset ();
 
@@ -609,13 +610,8 @@ on_statement_inserted (gint         graph_id,
                        GPtrArray   *rdf_types,
                        gpointer     user_data)
 {
-	//if (g_strcmp0 (predicate, RDF_PREFIX "type") == 0) {
-		//tracker_events_insert (subject, predicate, object, rdf_types, TRACKER_DBUS_EVENTS_TYPE_ADD);
-	//} else {
-		//tracker_events_insert (subject, predicate, object, rdf_types, TRACKER_DBUS_EVENTS_TYPE_UPDATE);
-	//}
-
-	/* For predicates it's always update here */
+	tracker_events_add_insert (graph_id, subject_id, subject, pred_id,
+	                           object_id, object, rdf_types);
 	tracker_writeback_check (graph_id, subject_id, subject, pred_id, object_id, object, rdf_types);
 }
 
@@ -629,14 +625,10 @@ on_statement_deleted (gint         graph_id,
                       GPtrArray   *rdf_types,
                       gpointer     user_data)
 {
-	//if (g_strcmp0 (predicate, RDF_PREFIX "type") == 0) {
-		//tracker_events_insert (subject, predicate, object, rdf_types, TRACKER_DBUS_EVENTS_TYPE_DELETE);
-	//} else {
-		//tracker_events_insert (subject, predicate, object, rdf_types, TRACKER_DBUS_EVENTS_TYPE_UPDATE);
-	//}
-
-	/* For predicates it's always delete here */
-	tracker_writeback_check (graph_id, subject_id, subject, pred_id, object_id, object, rdf_types);
+	tracker_events_add_delete (graph_id, subject_id, subject, pred_id,
+	                           object_id, object, rdf_types);
+	tracker_writeback_check (graph_id, subject_id, subject, pred_id,
+	                         object_id, object, rdf_types);
 }
 
 void