Split transactions from a block. #1

Transactions are ripped from incoming blocks and put in the send queue
with the height of the block they belong to. Transactions are
priorized over blocks of same height to ensure all transactions are
transferred before the block containing them. Blocks contain only
header and dhashes of transactions.

This is less space efficient than the previous version which leaves
unsent transactions inside the block but packs already sent TXs using
their hash. But this also allows validation of Merkle tree and easesq
combining messages from multiple transponders.

src/bitcoin.c

@@ -131,11 +131,16 @@ bool bitcoin_inv_insert(struct bitcoin_storage const *st, struct msg *const m)
 	g_hash_table_insert(st->inv,key,m);
 
 	// Put hash key to the send queue
-	g_sequence_insert_sorted(st->send_queue,key,comparator,st->inv);
+	bitcoin_enqueue(st,key);
 
 	return true;
 }
 
+void bitcoin_enqueue(struct bitcoin_storage const *st, guchar *key)
+{
+	g_sequence_insert_sorted(st->send_queue,key,comparator,st->inv);
+}
+
 struct msg *bitcoin_dequeue(struct bitcoin_storage const *st)
 {
 	// Fetch and dequeue key
@@ -206,22 +211,23 @@ static gint comparator(gconstpointer a, gconstpointer b, gpointer inv)
 		     (char *)a,(char *)b);
 	}
 
-	// Sort by type if possible
-	if (msg_a->type != msg_b->type) return msg_a->type-msg_b->type;
+	// Already sent items are pushed towards the small end to aid
+	// them getting out of the queue
+	if (msg_a->sent) return -1;
+	if (msg_b->sent) return 1;
 
-	// TODO Compare transaction priority using Satoshi's
-	// algoritm. Meanwhile transactions are considered to have
-	// equal priority
-	if (msg_a->type == TX) return 0; // tie
+	// Send lowest height first
+	if (msg_a->height != msg_b->height) return msg_a->height - msg_b->height; 
 
-	// Blocks are sorted by creation date. Earlier one has a priority
-	if (msg_a->type == BLOCK) {
-		return 
-			GUINT32_FROM_LE(msg_a->block.timestamp_le) - 
-			GUINT32_FROM_LE(msg_b->block.timestamp_le);
-	}
+	// Send transactions first, then block. The priority is defined
+	// by the enum in bitcoin.h
+	if (msg_a->type != msg_b->type) return msg_a->type - msg_b->type;
 
-	// If both are OTHER, consider it a tie
+	// Now we may have two unconfirmed transactions, two blocks of
+	// same height (fork) or two transactions belonging to the
+	// same block. More elegant sorting (Satoshi's algorithm) may
+	// be done especially for the unconfirmed transactions, but
+	// currently we are considering it as a tie.
 	return 0;
 }
 

src/bitcoin.h

@@ -10,15 +10,15 @@
 
 /**
  * Message types. Order also defines priority in transmission. The
- * priority is descending, meaning that BLOCK has priority over TX.
+ * priority is descending, meaning that TX has priority over BLOCK.
  * UNDEFINED doesn't match anything in Bitcoin Specification and
  * should never be stored (used internally in incoming_node_data()
  * when type is not known yet).
  */
 enum msg_type {
 	UNDEFINED,
-	BLOCK,
 	TX,
+	BLOCK,
 	ADDR,
 	VERSION,
 	VERACK,
@@ -145,6 +145,12 @@ struct bitcoin_storage bitcoin_new_storage();
  */
 bool bitcoin_inv_insert(struct bitcoin_storage const *st, struct msg *const m);
 
+/**
+ * Inserts given inventory hash to send queue. This doesn't add the
+ * object to inventory. You must do it manually.
+ */
+void bitcoin_enqueue(struct bitcoin_storage const *st, guchar *key);
+
 /**
  * Fetches unsent message with highest sending priority.
  */

src/incoming_node.c

@@ -8,12 +8,16 @@
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
+#include <openssl/sha.h>
 #include "bitcoin.h"
 #include "log.h"
 
 // Share memory with compact struct
 #define COMPACT ((struct msg *)buf)
 
+// Height of an unconfirmed transaction is high (priority low)
+#define UNCONFIRMED INT_MAX 
+
 // Processes messages having this format:
 // https://en.bitcoin.it/wiki/Protocol_specification#Message_structure
 void incoming_node_data(const int fd, struct bitcoin_storage *const st)
@@ -56,7 +60,6 @@ void incoming_node_data(const int fd, struct bitcoin_storage *const st)
 		if (buf_type != INV) {
 			COMPACT->type = buf_type;
 			COMPACT->length = payload_len;
-			COMPACT->height = ~0; // meaning: not important
 			COMPACT->sent = false;
 			// Rewind to compacted payload starting pos
 			buf_pos = offsetof(struct msg,payload);
@@ -97,22 +100,102 @@ void incoming_node_data(const int fd, struct bitcoin_storage *const st)
 
 		break;
 	case TX:
-	case BLOCK:
 		// Free some memory if the buffer is larger than contents
 		if (buf_allocated != buf_pos) {
 			buf = g_realloc(buf,buf_pos);
 		}
 
-		// Upadate height
-		if (buf_type == BLOCK) {
-			const struct msg *parent = g_hash_table_lookup(st->inv,COMPACT->block.prev_block);
-			if (parent == NULL) {
-				COMPACT->height = 0; // Start orphaned block from 0
+		COMPACT->height = UNCONFIRMED;
+
+		if (bitcoin_inv_insert(st,COMPACT)) {
+			// Do not reuse buffer memory because
+			// it is stored to the inventory
+			buf = NULL;
+			buf_allocated = 0;
+		} else {
+			// If already received, do not do anything
+			warnx("Protocol quirk: duplicate %s %s",
+			      bitcoin_type_str(COMPACT),
+			      hex256(bitcoin_inv_hash(COMPACT)));
+		}
+		break;
+	case BLOCK:
+		;
+		// Set block height
+		const struct msg *parent = g_hash_table_lookup(st->inv,COMPACT->block.prev_block);
+		if (parent == NULL) {
+			COMPACT->height = 0; // Start orphaned block from 0
+		} else {
+			COMPACT->height = parent->height+1;
+		}
+
+		// Traverse all transactions in the block and allocate
+		// memory separately for each transaction.
+		guint8 *hash_p = COMPACT->block.txs;
+		const guint64 txs = get_var_int((const guint8**)&hash_p); // Argh, typecasts
+		const guint8 *p = hash_p;
+		guchar* key = NULL;
+
+		for (guint64 tx_i=0; tx_i<txs; tx_i++) {
+			// If key buffer is not NULL it means the
+			// buffer was not used in last iteration (the
+			// transaction was sent already).
+			if (key == NULL) key = g_malloc(SHA256_DIGEST_LENGTH);
+
+			// Find out tx length and position and check
+			// if tx already exists
+			int length = bitcoin_tx_len(p);
+			const guchar *hash = dhash(p,length,key);
+			struct msg *tx = g_hash_table_lookup(st->inv,hash);
+
+			if (tx == NULL) {
+				// Transaction has not seen yet.
+				// Allocate storage for new tx.
+				tx = g_malloc(offsetof(struct msg,payload)+length);
+				// Set headers
+				tx->length = length;
+				tx->sent = false;
+				tx->type = TX;
+				tx->height = COMPACT->height;
+
+				// Fill with tx data
+				memcpy(tx->payload, p, length);
+
+				// Insert
+				g_hash_table_insert(st->inv,key,tx);
 			} else {
-				COMPACT->height = parent->height+1;
+				// Tune existing tx height
+				tx->height = COMPACT->height;
+			}
+
+			// Debugging
+			printf("Block tx %s, net bytes %d, sent %d\n",
+			       hex256(key), length-SHA256_DIGEST_LENGTH, tx->sent);
+
+			// Overwrite transaction data by its hash and
+			// advance pointers. Do not touch p after this.
+			memcpy(hash_p, key, SHA256_DIGEST_LENGTH);
+			hash_p += SHA256_DIGEST_LENGTH;
+			p += length;
+
+			// Requeue transaction if it is not sent, even
+			// if it is enqueued already.
+			if (!tx->sent) {
+				bitcoin_enqueue(st,key);
+				key = NULL; // Force reallocation for key
 			}
 		}
 
+		// Sanity checks
+		if (p != COMPACT->payload + COMPACT->length) {
+			errx(9,"Block length doesn't match to contents");
+		}
+
+		// Free unused memory freed by using hashes
+		COMPACT->length = hash_p - COMPACT->payload;
+		buf = g_realloc(buf, hash_p - (guint8*)buf);
+
+		// Finally, put the block in send queue
 		if (bitcoin_inv_insert(st,COMPACT)) {
 			// Do not reuse buffer memory because
 			// it is stored to the inventory

src/main.c

@@ -161,58 +161,7 @@ void serial(const int devfd, struct bitcoin_storage *const st)
 		encode(&s,&siglen,1); // FIXME doesn't work on big endian archs
 		encode(&s,&sig,siglen);
 		encode(&s,&m->type,1); // FIXME doesn't work on big endian archs
-		if (m->type == BLOCK) {
-			encode(&s,&m->block,sizeof(struct block));
-			const guint8 *p = m->block.txs;
-			const guint64 txs = get_var_int(&p);
-			for (guint64 tx=0; tx<txs; tx++) {
-				int length = bitcoin_tx_len(p);
-				const guchar *hash = dhash(p,length,NULL);
-
-				// Some debugging tools
-				bool dequeued = false;
-				int net_bytes;
-
-				// Look for the hash from inventory
-				struct msg *block_tx = g_hash_table_lookup(st->inv,hash);
-
-				// TODO we should add tx to inventory
-				// to allow supporting better queue
-				// position algorithm
-
-				if (block_tx == NULL || !block_tx->sent) {
-					// Mark it sent if it is also in the queue.
-					if (block_tx != NULL) {
-						block_tx->sent = true;
-						dequeued = true;
-					}
-
-					// Fresh meat, sending everything
-					const guint8 is_full = 1;
-					encode(&s,&is_full,sizeof(is_full));
-					encode(&s,p,length);
-
-					net_bytes = 1;
-				} else {
-					// Already seen. Transmit only the hash
-					const guint8 is_full = 0;
-					encode(&s,&is_full,sizeof(is_full));
-					encode(&s,hash,SHA256_DIGEST_LENGTH);
-
-					net_bytes = SHA256_DIGEST_LENGTH-length+1;
-				}
-				// Debugging
-				printf("Block tx %s, net bytes %d, dequeued %d\n",
-				       hex256(hash), net_bytes, dequeued);
-
-				p += length;
-			}
-			if (p != m->payload + m->length) {
-				errx(9,"Block length doesn't match to contents");
-			}
-		} else {
-			encode(&s,m->payload,m->length);
-		}
+		encode(&s,m->payload,m->length);
 
 		// Finishing encoding and updating buffer
 		buf_left = encode_end(&s);