Gossip store optimizations II: the gossipd parts

cli/test/run-large-input.c

@@ -25,12 +25,6 @@ int test_printf(const char *format, ...);
 #undef main
 
 /* AUTOGENERATED MOCKS START */
-/* Generated stub for amount_sat_eq */
-bool amount_sat_eq(struct amount_sat a UNNEEDED, struct amount_sat b UNNEEDED)
-{ fprintf(stderr, "amount_sat_eq called!\n"); abort(); }
-/* Generated stub for amount_sat_less */
-bool amount_sat_less(struct amount_sat a UNNEEDED, struct amount_sat b UNNEEDED)
-{ fprintf(stderr, "amount_sat_less called!\n"); abort(); }
 /* Generated stub for version_and_exit */
 char *version_and_exit(const void *unused UNNEEDED)
 { fprintf(stderr, "version_and_exit called!\n"); abort(); }

gossipd/broadcast.c

@@ -8,88 +8,126 @@
 #include <common/pseudorand.h>
 #include <common/status.h>
 #include <common/type_to_string.h>
+#include <errno.h>
 #include <gossipd/broadcast.h>
+#include <gossipd/gossip_store.h>
 #include <wire/gen_peer_wire.h>
 
-struct queued_message {
-	/* Broadcast index. */
-	u64 index;
-
-	/* Timestamp, for filtering. */
-	u32 timestamp;
-
-	/* Serialized payload */
-	const u8 *payload;
-};
+static void destroy_broadcast_state(struct broadcast_state *bstate)
+{
+	uintmap_clear(&bstate->broadcasts);
+}
 
-struct broadcast_state *new_broadcast_state(tal_t *ctx)
+struct broadcast_state *new_broadcast_state(struct routing_state *rstate,
+					    struct gossip_store *gs,
+					    struct list_head *peers)
 {
-	struct broadcast_state *bstate = tal(ctx, struct broadcast_state);
+	struct broadcast_state *bstate = tal(rstate, struct broadcast_state);
 	uintmap_init(&bstate->broadcasts);
-	/* Skip 0 because we initialize peers with 0 */
-	bstate->next_index = 1;
 	bstate->count = 0;
+	bstate->gs = gs;
+	bstate->peers = peers;
+	tal_add_destructor(bstate, destroy_broadcast_state);
 	return bstate;
 }
 
-void broadcast_del(struct broadcast_state *bstate, u64 index, const u8 *payload)
+void broadcast_del(struct broadcast_state *bstate,
+		   struct broadcastable *bcast)
 {
-	const struct queued_message *q = uintmap_del(&bstate->broadcasts, index);
-	if (q != NULL) {
-		assert(q->payload == payload);
-		tal_free(q);
+	const struct broadcastable *b
+		= uintmap_del(&bstate->broadcasts, bcast->index);
+	if (b != NULL) {
+		assert(b == bcast);
+		bstate->count--;
 		broadcast_state_check(bstate, "broadcast_del");
+		bcast->index = 0;
 	}
 }
 
-static void destroy_queued_message(struct queued_message *msg,
-				   struct broadcast_state *bstate)
+static void add_broadcast(struct broadcast_state *bstate,
+			  struct broadcastable *bcast)
 {
-	broadcast_del(bstate, msg->index, msg->payload);
-	bstate->count--;
+	assert(bcast);
+	assert(bcast->index);
+	if (!uintmap_add(&bstate->broadcasts, bcast->index, bcast))
+		abort();
+	bstate->count++;
 }
 
-static struct queued_message *new_queued_message(const tal_t *ctx,
-						 struct broadcast_state *bstate,
-						 const u8 *payload,
-						 u32 timestamp,
-						 u64 index)
+void insert_broadcast_nostore(struct broadcast_state *bstate,
+			      struct broadcastable *bcast)
 {
-	struct queued_message *msg = tal(ctx, struct queued_message);
-	assert(payload);
-	msg->payload = payload;
-	msg->index = index;
-	msg->timestamp = timestamp;
-	uintmap_add(&bstate->broadcasts, index, msg);
-	tal_add_destructor2(msg, destroy_queued_message, bstate);
-	bstate->count++;
-	return msg;
+	add_broadcast(bstate, bcast);
+	broadcast_state_check(bstate, "insert_broadcast");
 }
 
-u64 insert_broadcast(struct broadcast_state *bstate,
-		      const u8 *payload, u32 timestamp)
+void insert_broadcast(struct broadcast_state **bstate,
+		      const u8 *msg,
+		      struct broadcastable *bcast)
 {
-	/* Free payload, free index. */
-	new_queued_message(payload, bstate, payload, timestamp,
-			   bstate->next_index);
-	broadcast_state_check(bstate, "insert_broadcast");
-	return bstate->next_index++;
+	u32 offset;
+
+	/* If we're loading from the store, we already have index */
+	if (!bcast->index) {
+		u64 idx;
+
+		bcast->index = idx = gossip_store_add((*bstate)->gs, msg);
+		if (!idx)
+			status_failed(STATUS_FAIL_INTERNAL_ERROR,
+				      "Could not add to gossip store: %s",
+				      strerror(errno));
+		/* We assume we can fit in 32 bits for now! */
+		assert(idx == bcast->index);
+	}
+
+	insert_broadcast_nostore(*bstate, bcast);
+
+	/* If it compacts, it replaces *bstate */
+	gossip_store_maybe_compact((*bstate)->gs, bstate, &offset);
+	if (offset)
+		update_peers_broadcast_index((*bstate)->peers, offset);
+}
+
+struct broadcastable *next_broadcast_raw(struct broadcast_state *bstate,
+					 u32 *last_index)
+{
+	struct broadcastable *b;
+	u64 idx = *last_index;
+
+	b = uintmap_after(&bstate->broadcasts, &idx);
+	if (!b)
+		return NULL;
+	/* Assert no overflow */
+	*last_index = idx;
+	assert(*last_index == idx);
+	return b;
 }
 
-const u8 *next_broadcast(struct broadcast_state *bstate,
+const u8 *next_broadcast(const tal_t *ctx,
+			 struct broadcast_state *bstate,
 			 u32 timestamp_min, u32 timestamp_max,
-			 u64 *last_index)
+			 u32 *last_index)
 {
-	struct queued_message *m;
+	struct broadcastable *b;
 
-	while ((m = uintmap_after(&bstate->broadcasts, last_index)) != NULL) {
-		if (m->timestamp >= timestamp_min
-		    && m->timestamp <= timestamp_max)
-			return m->payload;
+	while ((b = next_broadcast_raw(bstate, last_index)) != NULL) {
+		if (b->timestamp >= timestamp_min
+		    && b->timestamp <= timestamp_max) {
+			return gossip_store_get(ctx, bstate->gs, b->index);
+		}
 	}
 	return NULL;
 }
 
+u64 broadcast_final_index(const struct broadcast_state *bstate)
+{
+	u64 idx;
+
+	if (!uintmap_last(&bstate->broadcasts, &idx))
+		return 0;
+	return idx;
+}
+
 #ifdef PEDANTIC
 static const struct pubkey *
 pubkey_keyof(const struct pubkey *pk)
@@ -134,7 +172,8 @@ struct broadcast_state *broadcast_state_check(struct broadcast_state *b,
 	struct pubkey node_id_1,  node_id_2, bitcoin_key;
 	u32 timestamp, fees;
 	u16 flags, expiry;
-	u64 index = 0, htlc_minimum_msat;
+	u32 index = 0;
+	u64 htlc_minimum_msat;
 	struct pubkey_set pubkeys;
 	/* We actually only need a set, not a map. */
 	UINTMAP(u64 *) channels;

gossipd/broadcast.h

@@ -9,31 +9,67 @@
 
 /* Common functionality to implement staggered broadcasts with replacement. */
 
+struct routing_state;
+
+/* This is nested inside a node, chan or half_chan; rewriting the store can
+ * cause it to change! */
+struct broadcastable {
+	/* This is also the offset within the gossip_store; even with 1M
+	 * channels we still have a factor of 8 before this wraps. */
+	u32 index;
+	u32 timestamp;
+};
+
 struct broadcast_state {
-	u64 next_index;
-	UINTMAP(struct queued_message *) broadcasts;
+	UINTMAP(struct broadcastable *) broadcasts;
 	size_t count;
+	struct gossip_store *gs;
+	struct list_head *peers;
 };
 
-struct broadcast_state *new_broadcast_state(tal_t *ctx);
+static inline void broadcastable_init(struct broadcastable *bcast)
+{
+	bcast->index = 0;
+}
+
+struct broadcast_state *new_broadcast_state(struct routing_state *rstate,
+					    struct gossip_store *gs,
+					    struct list_head *peers);
 
-/* Append a queued message for broadcast.  Freeing the msg will remove it. */
-u64 insert_broadcast(struct broadcast_state *bstate, const u8 *msg,
-		     u32 timestamp);
+/* Append a queued message for broadcast.  Must be explicitly deleted.
+ * Also adds it to the gossip store. */
+void insert_broadcast(struct broadcast_state **bstate,
+		      const u8 *msg,
+		      struct broadcastable *bcast);
 
-/* Manually delete a broadcast: not usually needed, since destructor does it */
-void broadcast_del(struct broadcast_state *bstate, u64 index, const u8 *payload);
+/* Add to broadcast, but not store: for gossip store compaction. */
+void insert_broadcast_nostore(struct broadcast_state *bstate,
+			      struct broadcastable *bcast);
+
+/* Delete a broadcast: not usually needed, since destructor does it */
+void broadcast_del(struct broadcast_state *bstate,
+		   struct broadcastable *bcast);
+
+/* Return the next broadcastable entry; doesn't load it. */
+struct broadcastable *next_broadcast_raw(struct broadcast_state *bstate,
+					 u32 *last_index);
 
 /* Return the broadcast with index >= *last_index, timestamp >= min and <= max
  * and update *last_index.
  * There's no broadcast with index 0. */
-const u8 *next_broadcast(struct broadcast_state *bstate,
+const u8 *next_broadcast(const tal_t *ctx,
+			 struct broadcast_state *bstate,
 			 u32 timestamp_min, u32 timestamp_max,
-			 u64 *last_index);
+			 u32 *last_index);
+
+/* index of last entry. */
+u64 broadcast_final_index(const struct broadcast_state *bstate);
 
 /* Returns b if all OK, otherwise aborts if abortstr non-NULL, otherwise returns
  * NULL. */
 struct broadcast_state *broadcast_state_check(struct broadcast_state *b,
 					      const char *abortstr);
 
+/* Callback for after we compacted the store */
+void update_peers_broadcast_index(struct list_head *peers, u32 offset);
 #endif /* LIGHTNING_GOSSIPD_BROADCAST_H */