Set an upper limit on how many piece messages to queue for

writing to a peer. If more requests arrive they will be ignored.
When all pieces have been sent to the peer, in order for it not to
wait on the ignored requests, its state will be reset by a choke
followed by an unchoke message. 

Without this limit there was no bound on how much memory btpd would
consume to satisfy a greedy peer.

btpd/net.c

@@ -129,6 +129,7 @@ net_write(struct peer *p, unsigned long wmax)
     while (bcount > 0) {
 	unsigned long bufdelta = nl->nb->len - p->outq_off;
 	if (bcount >= bufdelta) {
+	    peer_sent(p, nl->nb);
 	    if (nl->nb->type == NB_TORRENTDATA) {
 		p->tp->uploaded += bufdelta;
 		p->rate_from_me[btpd.seconds % RATEHISTORY] += bufdelta;

btpd/peer.c

@@ -86,6 +86,10 @@ peer_unsend(struct peer *p, struct nb_link *nl)
 {
     if (!(nl == BTPDQ_FIRST(&p->outq) && p->outq_off > 0)) {
 	BTPDQ_REMOVE(&p->outq, nl, entry);
+	if (nl->nb->type == NB_TORRENTDATA) {
+	    assert(p->npiece_msgs > 0);
+	    p->npiece_msgs--;
+	}
 	nb_drop(nl->nb);
 	free(nl);
 	if (BTPDQ_EMPTY(&p->outq)) {
@@ -100,6 +104,21 @@ peer_unsend(struct peer *p, struct nb_link *nl)
 	return 0;
 }
 
+void
+peer_sent(struct peer *p, struct net_buf *nb)
+{
+    switch (nb->type) {
+    case NB_TORRENTDATA:
+	assert(p->npiece_msgs > 0);
+	p->npiece_msgs--;
+	break;
+    case NB_UNCHOKE:
+	assert(p->npiece_msgs == 0);
+	p->flags &= ~PF_NO_REQUESTS;
+	break;
+    }
+}
+
 void
 peer_request(struct peer *p, uint32_t index, uint32_t begin, uint32_t len)
 {
@@ -345,10 +364,18 @@ void
 peer_on_request(struct peer *p, uint32_t index, uint32_t begin,
     uint32_t length)
 {
-    off_t cbegin = index * p->tp->meta.piece_length + begin;
-    char * content = torrent_get_bytes(p->tp, cbegin, length);
-    peer_send(p, nb_create_piece(index, begin, length));
-    peer_send(p, nb_create_torrentdata(content, length));
+    if ((p->flags & PF_NO_REQUESTS) == 0) {
+	off_t cbegin = index * p->tp->meta.piece_length + begin;
+	char * content = torrent_get_bytes(p->tp, cbegin, length);
+	peer_send(p, nb_create_piece(index, begin, length));
+	peer_send(p, nb_create_torrentdata(content, length));
+	p->npiece_msgs++;
+	if (p->npiece_msgs >= MAXPIECEMSGS) {
+	    peer_send(p, btpd.choke_msg);
+	    peer_send(p, btpd.unchoke_msg);
+	    p->flags |= PF_NO_REQUESTS;
+	}
+    }
 }
 
 void

btpd/peer.h

@@ -1,22 +1,23 @@
 #ifndef BTPD_PEER_H
 #define BTPD_PEER_H
 
-#define PF_I_WANT	0x01	/* We want to download from the peer */
-#define PF_I_CHOKE	0x02	/* We choke the peer */
-#define	PF_P_WANT	0x04	/* The peer wants to download from us */
-#define	PF_P_CHOKE	0x08	/* The peer is choking us */
-#define PF_ON_READQ	0x10
-#define PF_ON_WRITEQ	0x20
-#define PF_ATTACHED	0x40
-#define PF_WRITE_CLOSE	0x80	/* Close connection after writing all data */
+#define PF_I_WANT	  0x1	/* We want to download from the peer */
+#define PF_I_CHOKE	  0x2	/* We choke the peer */
+#define PF_P_WANT	  0x4	/* The peer wants to download from us */
+#define PF_P_CHOKE	  0x8	/* The peer is choking us */
+#define PF_ON_READQ	 0x10
+#define PF_ON_WRITEQ	 0x20
+#define PF_ATTACHED	 0x40
+#define PF_WRITE_CLOSE	 0x80	/* Close connection after writing all data */
+#define PF_NO_REQUESTS	0x100
 
 #define RATEHISTORY 20
-
+#define MAXPIECEMSGS 128
 #define MAXPIPEDREQUESTS 10
 
 struct peer {
     int sd;
-    uint8_t flags;
+    uint16_t flags;
     uint8_t *piece_field;
     uint32_t npieces;
     uint32_t nwant;
@@ -28,6 +29,7 @@ struct peer {
     struct nb_tq my_reqs;
 
     unsigned nreqs_out;
+    unsigned npiece_msgs;
 
     size_t outq_off;
     struct nb_tq outq;
@@ -50,6 +52,7 @@ BTPDQ_HEAD(peer_tq, peer);
 
 void peer_send(struct peer *p, struct net_buf *nb);
 int peer_unsend(struct peer *p, struct nb_link *nl);
+void peer_sent(struct peer *p, struct net_buf *nb);
 
 void peer_unchoke(struct peer *p);
 void peer_choke(struct peer *p);