* Implemented the full cm_ life cycle.

* Added fast resume support. A resume file is loaded when a torrent is started and saved when it's stopped or done. If no resume file is found or the file information doesn't match whats on disk, the content is tested for existing pieces. * cm_prealloc now can allocate several adjacent pieces to the given piece. This further reduces fragmentation. How many pieces are allocated at a time is controlled by cm_alloc_size which can be set by the user with the --prealloc option. * Some changes were also made to the torrent api.
18 lat temu · 7068f34a51
--- a/btpd/btpd.c
+++ b/btpd/btpd.c
@@ -115,7 +115,7 @@ load_library(void)
    for (int i = 0; i < ne; i++) {
        struct torrent *tp;
        struct dirent *e = entries[i];
        if (torrent_create(&tp, e->d_name) == 0)
        if (torrent_load(&tp, e->d_name) == 0)
            btpd_add_torrent(tp);
        free(e);
    }
--- a/btpd/content.c
+++ b/btpd/content.c
@@ -1,6 +1,10 @@
 #include <sys/types.h>
 #include <sys/stat.h>

 #include <fcntl.h>
 #include <math.h>
 #include <pthread.h>
 #include <stdio.h>
 #include <string.h>
 #include <unistd.h>

@@ -9,30 +13,36 @@
 #include "btpd.h"
 #include "stream.h"

 struct data {
 struct cm_write_data {
    uint32_t begin;
    uint8_t *buf;
    size_t len;
    BTPDQ_ENTRY(data) entry;
    BTPDQ_ENTRY(cm_write_data) entry;
 };

 BTPDQ_HEAD(data_tq, data);
 BTPDQ_HEAD(cm_write_data_tq, cm_write_data);

 enum cm_op_type {
    CM_ALLOC,
    CM_SAVE,
    CM_START,
    CM_TEST,
    CM_WRITE
 };

 struct cm_op {
    struct torrent *tp;

    int error;
    int received;
    enum cm_op_type type;
    union {
        struct {
            uint32_t piece;
            uint32_t pos;
        } alloc;
        struct {
            volatile sig_atomic_t cancel;
        } start;
        struct {
            uint32_t piece;
            uint32_t pos;
@@ -41,13 +51,10 @@ struct cm_op {
        struct {
            uint32_t piece;
            uint32_t pos;
            struct data_tq q;
            struct cm_write_data_tq q;
        } write;
    } u;

    int error;
    char *errmsg;

    BTPDQ_ENTRY(cm_op) cm_entry;
    BTPDQ_ENTRY(cm_op) td_entry;
 };
@@ -55,6 +62,8 @@ struct cm_op {
 BTPDQ_HEAD(cm_op_tq, cm_op);

 struct content {
    int active;

    uint32_t npieces_got;

    off_t ncontent_bytes;
@@ -74,10 +83,14 @@ struct content {

 #define ZEROBUFLEN (1 << 14)

 struct cm_comm {
    struct cm_op_tq q;
    pthread_mutex_t lock;
    pthread_cond_t cond;
 };

 static struct cm_comm m_long_comm, m_short_comm;
 static const uint8_t m_zerobuf[ZEROBUFLEN];
 static struct cm_op_tq m_tdq = BTPDQ_HEAD_INITIALIZER(m_tdq);
 static pthread_mutex_t m_tdq_lock;
 static pthread_cond_t m_tdq_cond;

 static int
 fd_cb_rd(const char *path, int *fd, void *arg)
@@ -94,6 +107,27 @@ fd_cb_wr(const char *path, int *fd, void *arg)
        path);
 }

 static void
 cm_td_post_common(struct cm_comm *comm, struct cm_op *op)
 {
    pthread_mutex_lock(&comm->lock);
    BTPDQ_INSERT_TAIL(&comm->q, op, td_entry);
    pthread_mutex_unlock(&comm->lock);
    pthread_cond_signal(&comm->cond);
 }

 static void
 cm_td_post_long(struct cm_op *op)
 {
    cm_td_post_common(&m_long_comm, op);
 }

 static void
 cm_td_post_short(struct cm_op *op)
 {
    cm_td_post_common(&m_short_comm, op);
 }

 static void
 run_todo(struct content *cm)
 {
@@ -107,10 +141,65 @@ run_todo(struct content *cm)
        return;
    }

    pthread_mutex_lock(&m_tdq_lock);
    BTPDQ_INSERT_TAIL(&m_tdq, op, td_entry);
    pthread_mutex_unlock(&m_tdq_lock);
    pthread_cond_signal(&m_tdq_cond);
    if (op->type != CM_START)
        cm_td_post_short(op);
    else
        cm_td_post_long(op);
 }

 static void
 add_todo(struct content *cm, struct cm_op *op)
 {
    int was_empty = BTPDQ_EMPTY(&cm->todoq);
    BTPDQ_INSERT_TAIL(&cm->todoq, op, cm_entry);
    if (was_empty)
        run_todo(cm);
 }

 void
 cm_destroy(struct torrent *tp)
 {
    struct content *cm = tp->cm;
    bts_close(cm->rds);
    free(cm->piece_field);
    free(cm->block_field);
    free(cm->hold_field);
    free(cm->pos_field);
    tp->cm = NULL;
    torrent_on_cm_stopped(tp);
 }

 void
 cm_save(struct torrent *tp)
 {
    struct content *cm = tp->cm;
    struct cm_op *op = btpd_calloc(1, sizeof(*op));
    op->tp = tp;
    op->type = CM_SAVE;
    add_todo(cm, op);
 }

 void
 cm_stop(struct torrent *tp)
 {
    struct content *cm = tp->cm;

    struct cm_op *op = BTPDQ_FIRST(&cm->todoq);
    if (op != NULL && op->type == CM_START) {
        pthread_mutex_lock(&m_long_comm.lock);
        if (op->received)
            op->u.start.cancel = 1;
        else {
            BTPDQ_REMOVE(&m_long_comm.q, op, td_entry);
            BTPDQ_REMOVE(&cm->todoq, op, cm_entry);
            free(op);
        }
        pthread_mutex_unlock(&m_long_comm.lock);
    } else if (cm->npieces_got < tp->meta.npieces)
        cm_save(tp);

    if (BTPDQ_EMPTY(&cm->todoq))
        cm_destroy(tp);
 }

 static void
@@ -121,19 +210,28 @@ cm_td_cb(void *arg)
    struct content *cm = tp->cm;

    if (op->error)
        btpd_err("%s", op->errmsg);
        btpd_err("IO error for %s.\n", tp->relpath);

    switch (op->type) {
    case CM_ALLOC:
        set_bit(cm->pos_field, op->u.alloc.pos);
        clear_bit(cm->hold_field, op->u.alloc.piece);
        break;
    case CM_START:
        if (cm->active) {
            assert(!op->u.start.cancel);
            torrent_on_cm_started(tp);
        }
        break;
    case CM_TEST:
        if (op->u.test.ok) {
            assert(cm->npieces_got < tp->meta.npieces);
            cm->npieces_got++;
            set_bit(cm->piece_field, op->u.test.piece);
            if (tp->net != NULL)
                dl_on_ok_piece(op->tp->net, op->u.test.piece);
            if (cm_full(tp))
                cm_save(tp);
        } else {
            cm->ncontent_bytes -= torrent_piece_size(tp, op->u.test.piece);
            bzero(cm->block_field + op->u.test.piece * cm->bppbf, cm->bppbf);
@@ -141,116 +239,16 @@ cm_td_cb(void *arg)
                dl_on_bad_piece(tp->net, op->u.test.piece);
        }
        break;
    default:
    case CM_SAVE:
    case CM_WRITE:
        break;
    }
    BTPDQ_REMOVE(&cm->todoq, op, cm_entry);
    free(op);
    if (!BTPDQ_EMPTY(&cm->todoq))
        run_todo(cm);
 }

 static int
 test_hash(struct torrent *tp, uint8_t *hash, uint32_t index)
 {
    if (tp->meta.piece_hash != NULL)
        return bcmp(hash, tp->meta.piece_hash[index], SHA_DIGEST_LENGTH);
    else {
        char piece_hash[SHA_DIGEST_LENGTH];
        int fd;
        int err;

        err = vopen(&fd, O_RDONLY, "library/%s/torrent", tp->relpath);
        if (err != 0)
            btpd_err("test_hash: %s\n", strerror(err));

        lseek(fd, tp->meta.pieces_off + index * SHA_DIGEST_LENGTH, SEEK_SET);
        read(fd, piece_hash, SHA_DIGEST_LENGTH);
        close(fd);

        return bcmp(hash, piece_hash, SHA_DIGEST_LENGTH);
    }
 }

 static void
 cm_td_alloc(struct cm_op *op)
 {
    struct bt_stream *bts;
    off_t len = torrent_piece_size(op->tp, op->u.alloc.pos);
    off_t off = op->tp->meta.piece_length * op->u.alloc.pos;
    bts_open(&bts, &op->tp->meta, fd_cb_wr, op->tp);
    while (len > 0) {
        size_t wlen = min(ZEROBUFLEN, len);
        bts_put(bts, off, m_zerobuf, wlen);
        len -= wlen;
        off += wlen;
    }
    bts_close(bts);
 }

 static void
 cm_td_test(struct cm_op *op)
 {
    uint8_t hash[SHA_DIGEST_LENGTH];
    struct bt_stream *bts;
    bts_open(&bts, &op->tp->meta, fd_cb_rd, op->tp);
    bts_sha(bts, op->u.test.pos * op->tp->meta.piece_length,
        torrent_piece_size(op->tp, op->u.test.piece), hash);
    bts_close(bts);
    op->u.test.ok = test_hash(op->tp, hash, op->u.test.piece) == 0;
 }

 static void
 cm_td_write(struct cm_op *op)
 {
    struct data *d, *next;
    off_t base = op->tp->meta.piece_length * op->u.write.pos;
    struct bt_stream *bts;
    bts_open(&bts, &op->tp->meta, fd_cb_wr, op->tp);
    BTPDQ_FOREACH(d, &op->u.write.q, entry)
        bts_put(bts, base + d->begin, d->buf, d->len);
    bts_close(bts);
    BTPDQ_FOREACH_MUTABLE(d, &op->u.write.q, entry, next)
        free(d);
 }

 static void
 cm_td(void *arg)
 {
    for (;;) {
        pthread_mutex_lock(&m_tdq_lock);
        while (BTPDQ_EMPTY(&m_tdq))
            pthread_cond_wait(&m_tdq_cond, &m_tdq_lock);
        struct cm_op *op = BTPDQ_FIRST(&m_tdq);
        BTPDQ_REMOVE(&m_tdq, op, td_entry);
        pthread_mutex_unlock(&m_tdq_lock);

        switch (op->type) {
        case CM_ALLOC:
            cm_td_alloc(op);
            break;
        case CM_TEST:
            cm_td_test(op);
            break;
        case CM_WRITE:
            cm_td_write(op);
            break;
        default:
            abort();
        }
        td_post_begin();
        td_post(cm_td_cb, op);
        td_post_end();
    }
 }

 void
 cm_init(void)
 {
    pthread_t td;
    pthread_mutex_init(&m_tdq_lock, NULL);
    pthread_cond_init(&m_tdq_cond, NULL);
    pthread_create(&td, NULL, (void *(*)(void *))cm_td, NULL);
    else if (!cm->active)
        cm_destroy(tp);
 }

 int
@@ -259,6 +257,7 @@ cm_start(struct torrent *tp)
    int err;
    struct content *cm = btpd_calloc(1, sizeof(*cm));
    size_t pfield_size = ceil(tp->meta.npieces / 8.0);
    cm->active = 1;
    cm->bppbf = ceil((double)tp->meta.piece_length / (1 << 17));
    cm->piece_field = btpd_calloc(pfield_size, 1);
    cm->hold_field = btpd_calloc(pfield_size, 1);
@@ -269,11 +268,12 @@ cm_start(struct torrent *tp)

    if ((err = bts_open(&cm->rds, &tp->meta, fd_cb_rd, tp)) != 0)
        btpd_err("Error opening stream (%s).\n", strerror(err));
    if ((err = bts_open(&cm->wrs, &tp->meta, fd_cb_wr, tp)) != 0)
        btpd_err("Error opening stream (%s).\n", strerror(err));

    tp->cm = cm;
    torrent_cm_cb(tp, CM_STARTED);

    struct cm_op *op = btpd_calloc(1, sizeof(*op));
    op->tp = tp;
    op->type = CM_START;
    add_todo(cm, op);
    return 0;
 }

@@ -289,22 +289,18 @@ cm_get_bytes(struct torrent *tp, uint32_t piece, uint32_t begin, size_t len,
    return 0;
 }

 void
 cm_prealloc(struct torrent *tp, uint32_t piece)
 static void
 cm_post_alloc(struct torrent *tp, uint32_t piece)
 {
    struct content *cm = tp->cm;
    if (has_bit(cm->pos_field, piece))
        return;

    set_bit(cm->hold_field, piece);

    int was_empty = BTPDQ_EMPTY(&cm->todoq);
    struct cm_op *op = btpd_calloc(1, sizeof(*op));
    op->tp = tp;
    op->type = CM_ALLOC;
    op->u.alloc.piece = piece;
    op->u.alloc.pos = piece;
    BTPDQ_INSERT_TAIL(&cm->todoq, op, cm_entry);
    add_todo(cm, op);

    op = btpd_calloc(1, sizeof(*op));
    op->tp = tp;
@@ -312,25 +308,40 @@ cm_prealloc(struct torrent *tp, uint32_t piece)
    op->u.write.piece = piece;
    op->u.write.pos = piece;
    BTPDQ_INIT(&op->u.write.q);
    BTPDQ_INSERT_TAIL(&cm->todoq, op, cm_entry);
    add_todo(cm, op);
 }

    if (was_empty)
        run_todo(cm);
 void
 cm_prealloc(struct torrent *tp, uint32_t piece)
 {
    struct content *cm = tp->cm;

    if (cm_alloc_size == 0)
        set_bit(cm->pos_field, piece);
    else {
        unsigned npieces = ceil((double)cm_alloc_size / tp->meta.piece_length);
        uint32_t start = piece - piece % npieces;
        uint32_t end = min(start + npieces, tp->meta.npieces);

        while (start < end) {
            if ((!has_bit(cm->pos_field, start)
                    && !has_bit(cm->hold_field, start)))
                cm_post_alloc(tp, start);
            start++;
        }
    }
 }

 void
 cm_test_piece(struct torrent *tp, uint32_t piece)
 {
    struct content *cm = tp->cm;
    int was_empty = BTPDQ_EMPTY(&cm->todoq);
    struct cm_op *op = btpd_calloc(1, sizeof(*op));
    op->tp = tp;
    op->type = CM_TEST;
    op->u.test.piece = piece;
    op->u.test.pos = piece;
    BTPDQ_INSERT_TAIL(&cm->todoq, op, cm_entry);
    if (was_empty)
        run_todo(cm);
    add_todo(cm, op);
 }

 int
@@ -341,7 +352,7 @@ cm_put_bytes(struct torrent *tp, uint32_t piece, uint32_t begin,
    struct content *cm = tp->cm;

    if (has_bit(cm->hold_field, piece)) {
        struct data *d = btpd_calloc(1, sizeof(*d) + len);
        struct cm_write_data *d = btpd_calloc(1, sizeof(*d) + len);
        d->begin = begin;
        d->len = len;
        d->buf = (uint8_t *)(d + 1);
@@ -350,7 +361,7 @@ cm_put_bytes(struct torrent *tp, uint32_t piece, uint32_t begin,
        BTPDQ_FOREACH(op, &cm->todoq, cm_entry)
            if (op->type == CM_WRITE && op->u.write.piece == piece)
                break;
        struct data *it;
        struct cm_write_data *it;
        BTPDQ_FOREACH(it, &op->u.write.q, entry)
            if (it->begin > begin) {
                BTPDQ_INSERT_BEFORE(it, d, entry);
@@ -407,3 +418,383 @@ cm_has_piece(struct torrent *tp, uint32_t piece)
 {
    return has_bit(tp->cm->piece_field, piece);
 }

 static int
 test_hash(struct torrent *tp, uint8_t *hash, uint32_t piece)
 {
    if (tp->meta.piece_hash != NULL)
        return bcmp(hash, tp->meta.piece_hash[piece], SHA_DIGEST_LENGTH);
    else {
        char piece_hash[SHA_DIGEST_LENGTH];
        int fd;
        int err;

        err = vopen(&fd, O_RDONLY, "library/%s/torrent", tp->relpath);
        if (err != 0)
            btpd_err("test_hash: %s\n", strerror(err));

        lseek(fd, tp->meta.pieces_off + piece * SHA_DIGEST_LENGTH, SEEK_SET);
        read(fd, piece_hash, SHA_DIGEST_LENGTH);
        close(fd);

        return bcmp(hash, piece_hash, SHA_DIGEST_LENGTH);
    }
 }

 static int
 test_piece(struct torrent *tp, uint32_t pos, uint32_t piece, int *ok)
 {
    int err;
    uint8_t hash[SHA_DIGEST_LENGTH];
    struct bt_stream *bts;
    if ((err = bts_open(&bts, &tp->meta, fd_cb_rd, tp)) != 0)
        return err;
    if ((err = bts_sha(bts, pos * tp->meta.piece_length,
             torrent_piece_size(tp, piece), hash)) != 0)
        return err;;
    bts_close(bts);
    *ok = test_hash(tp, hash, piece) == 0;
    return 0;
 }

 static void
 cm_td_alloc(struct cm_op *op)
 {
    struct torrent *tp = op->tp;
    struct content *cm = tp->cm;
    uint32_t pos = op->u.alloc.pos;
    struct bt_stream *bts;
    int err;

    assert(!has_bit(cm->pos_field, pos));

    if ((err = bts_open(&bts, &tp->meta, fd_cb_wr, tp)) != 0)
        goto out;

    off_t len = torrent_piece_size(tp, pos);
    off_t off = tp->meta.piece_length * pos;
    while (len > 0) {
        size_t wlen = min(ZEROBUFLEN, len);
        if ((err = bts_put(bts, off, m_zerobuf, wlen)) != 0) {
            bts_close(bts);
            goto out;
        }
        len -= wlen;
        off += wlen;
    }
    err = bts_close(bts);
 out:
    if (err != 0)
        op->error = 1;
 }

 static int
 test_torrent(struct torrent *tp, volatile sig_atomic_t *cancel)
 {
    int err;
    FILE *fp;
    uint8_t (*hashes)[SHA_DIGEST_LENGTH];
    uint8_t hash[SHA_DIGEST_LENGTH];

    if ((err = vfopen(&fp, "r", "library/%s/torrent", tp->relpath)) != 0)
        return err;

    hashes = btpd_malloc(tp->meta.npieces * SHA_DIGEST_LENGTH);
    fseek(fp, tp->meta.pieces_off, SEEK_SET);
    fread(hashes, SHA_DIGEST_LENGTH, tp->meta.npieces, fp);
    fclose(fp);

    tp->meta.piece_hash = hashes;

    struct content *cm = tp->cm;
    for (uint32_t piece = 0; piece < tp->meta.npieces; piece++) {
        if (!has_bit(cm->pos_field, piece))
            continue;
        err = bts_sha(cm->rds, piece * tp->meta.piece_length,
            torrent_piece_size(tp, piece), hash);
        if (err != 0)
            break;
        if (test_hash(tp, hash, piece) == 0)
            set_bit(tp->cm->piece_field, piece);
        if (*cancel) {
            err = EINTR;
            break;
        }
    }

    tp->meta.piece_hash = NULL;
    free(hashes);
    return err;
 }

 int
 stat_and_adjust(struct torrent *tp, struct stat ret[])
 {
    char path[PATH_MAX];
    for (int i = 0; i < tp->meta.nfiles; i++) {
        snprintf(path, PATH_MAX, "library/%s/content/%s", tp->relpath,
            tp->meta.files[i].path);
 again:
        if (stat(path, &ret[i]) == -1) {
            if (errno == ENOENT) {
                ret[i].st_mtime = -1;
                ret[i].st_size = -1;
            } else
                return errno;
        }
        if (ret[i].st_size > tp->meta.files[i].length) {
            if (truncate(path, tp->meta.files[i].length) != 0)
                return errno;
            goto again;
        }
    }
    return 0;
 }

 static int
 load_resume(struct torrent *tp, struct stat sbs[])
 {
    int err, ver;
    FILE *fp;
    size_t pfsiz = ceil(tp->meta.npieces / 8.0);
    size_t bfsiz = tp->meta.npieces * tp->cm->bppbf;

    if ((err = vfopen(&fp, "r" , "library/%s/resume", tp->relpath)) != 0)
        return err;

    if (fscanf(fp, "%d\n", &ver) != 1)
        goto invalid;
    if (ver != 1)
        goto invalid;
    for (int i = 0; i < tp->meta.nfiles; i++) {
        long long size;
        time_t time;
        if (fscanf(fp, "%qd %ld\n", &size, &time) != 2)
            goto invalid;
        if (sbs[i].st_size != size || sbs[i].st_mtime != time)
            goto invalid;
    }
    if (fread(tp->cm->piece_field, 1, pfsiz, fp) != pfsiz)
        goto invalid;
    if (fread(tp->cm->block_field, 1, bfsiz, fp) != bfsiz)
        goto invalid;
    fclose(fp);
    return 0;
 invalid:
    fclose(fp);
    bzero(tp->cm->piece_field, pfsiz);
    bzero(tp->cm->block_field, bfsiz);
    return EINVAL;
 }

 static int
 save_resume(struct torrent *tp)
 {
    int err;
    FILE *fp;
    struct stat sbs[tp->meta.nfiles];
    if ((err = stat_and_adjust(tp, sbs)) != 0)
        return err;
    if ((err = vfopen(&fp, "wb", "library/%s/resume", tp->relpath)) != 0)
        return err;
    fprintf(fp, "%d\n", 1);
    for (int i = 0; i < tp->meta.nfiles; i++)
        fprintf(fp, "%qd %ld\n", (long long)sbs[i].st_size, sbs[i].st_mtime);
    fwrite(tp->cm->piece_field, 1, ceil(tp->meta.npieces / 8.0), fp);
    fwrite(tp->cm->block_field, 1, tp->meta.npieces * tp->cm->bppbf, fp);
    if (fclose(fp) != 0)
        err = errno;
    return err;
 }

 static void
 cm_td_save(struct cm_op *op)
 {
    int err;
    struct torrent *tp = op->tp;
    struct content *cm = tp->cm;
    struct bt_stream *bts = cm->wrs;

    cm->wrs = NULL;
    if ((err = bts_close(bts)) != 0)
        goto out;

    for (int i = 0; i < tp->meta.nfiles; i++) {
        int lerr;
        if ((lerr = vfsync("library/%s/content/%s", tp->relpath,
                 tp->meta.files[i])) != 0 && lerr != ENOENT)
            err = lerr;
    }
    if (err != 0)
        goto out;
    save_resume(tp);
 out:
    if (err != 0)
        op->error = 1;
 }

 static void
 cm_td_start(struct cm_op *op)
 {
    int err;
    struct stat sbs[op->tp->meta.nfiles];
    struct torrent *tp = op->tp;
    struct content *cm = tp->cm;

    if ((err = stat_and_adjust(op->tp, sbs)) != 0)
        goto out;

    if (load_resume(tp, sbs) != 0) {
        memset(cm->pos_field, 0xff, ceil(tp->meta.npieces / 8.0));
        off_t off = 0;
        for (int i = 0; i < tp->meta.nfiles; i++) {
            if (sbs[i].st_size == -1 || sbs[i].st_size == 0) {
                uint32_t start = off / tp->meta.piece_length;
                uint32_t end = (off + tp->meta.files[i].length - 1) /
                    tp->meta.piece_length;
                while (start <= end) {
                    clear_bit(cm->pos_field, start);
                    start++;
                }
            } else if (sbs[i].st_size < tp->meta.files[i].length) {
                uint32_t start = (off + sbs[i].st_size) /
                    tp->meta.piece_length;
                uint32_t end = (off + tp->meta.files[i].length - 1) /
                    tp->meta.piece_length;
                while (start <= end) {
                    clear_bit(cm->pos_field, start);
                    start++;
                }
            }
            off += tp->meta.files[i].length;
        }
        if (op->u.start.cancel)
            goto out;
        if ((err = test_torrent(tp, &op->u.start.cancel)) != 0)
            goto out;
        save_resume(tp);
    }

    bzero(cm->pos_field, ceil(tp->meta.npieces / 8.0));
    for (uint32_t piece = 0; piece < tp->meta.npieces; piece++) {
        if (cm_has_piece(tp, piece)) {
            cm->ncontent_bytes += torrent_piece_size(tp, piece);
            cm->npieces_got++;
            set_bit(cm->pos_field, piece);
            continue;
        }
        uint8_t *bf = cm->block_field + cm->bppbf * piece;
        uint32_t nblocks = torrent_piece_blocks(tp, piece);
        uint32_t nblocks_got = 0;
        for (uint32_t i = 0; i < nblocks; i++) {
            if (has_bit(bf, i)) {
                nblocks_got++;
                cm->ncontent_bytes +=
                    torrent_block_size(tp, piece, nblocks, i);
            }
        }
        if (nblocks_got == nblocks) {
            int ok;
            if (((err = test_piece(tp, piece, piece, &ok)) != 0
                    || op->u.start.cancel))
                goto out;
            if (ok) {
                set_bit(cm->pos_field, piece);
                set_bit(cm->piece_field, piece);
            } else
                bzero(bf, cm->bppbf);
        } else if (nblocks_got > 0)
            set_bit(cm->pos_field, piece);
    }

    if (cm->npieces_got < tp->meta.npieces)
        if ((err = bts_open(&cm->wrs, &tp->meta, fd_cb_wr, tp)) != 0)
            goto out;
 out:
    if (!op->u.start.cancel && err != 0)
        op->error = 1;
 }

 static void
 cm_td_test(struct cm_op *op)
 {
    if (test_piece(op->tp, op->u.test.pos, op->u.test.piece,
            &op->u.test.ok) != 0)
        op->error = 1;
 }

 static void
 cm_td_write(struct cm_op *op)
 {
    int err;
    struct cm_write_data *d, *next;
    off_t base = op->tp->meta.piece_length * op->u.write.pos;
    struct bt_stream *bts;
    if ((err = bts_open(&bts, &op->tp->meta, fd_cb_wr, op->tp)) != 0)
        goto out;
    BTPDQ_FOREACH(d, &op->u.write.q, entry)
        if ((err = bts_put(bts, base + d->begin, d->buf, d->len)) != 0) {
            bts_close(bts);
            goto out;
        }
    err = bts_close(bts);
 out:
    BTPDQ_FOREACH_MUTABLE(d, &op->u.write.q, entry, next)
        free(d);
    if (err)
        op->error = 1;
 }

 static void
 cm_td(void *arg)
 {
    struct cm_comm *comm = arg;
    struct cm_op *op;
    for (;;) {
        pthread_mutex_lock(&comm->lock);
        while (BTPDQ_EMPTY(&comm->q))
            pthread_cond_wait(&comm->cond, &comm->lock);

        op = BTPDQ_FIRST(&comm->q);
        BTPDQ_REMOVE(&comm->q, op, td_entry);
        op->received = 1;
        pthread_mutex_unlock(&comm->lock);

        switch (op->type) {
        case CM_ALLOC:
            cm_td_alloc(op);
            break;
        case CM_SAVE:
            cm_td_save(op);
            break;
        case CM_START:
            cm_td_start(op);
            break;
        case CM_TEST:
            cm_td_test(op);
            break;
        case CM_WRITE:
            cm_td_write(op);
            break;
        default:
            abort();
        }
        td_post_begin();
        td_post(cm_td_cb, op);
        td_post_end();
    }
 }

 void
 cm_init(void)
 {
    pthread_t td;
    BTPDQ_INIT(&m_long_comm.q);
    pthread_mutex_init(&m_long_comm.lock, NULL);
    pthread_cond_init(&m_long_comm.cond, NULL);
    pthread_create(&td, NULL, (void *(*)(void *))cm_td, &m_long_comm);
    BTPDQ_INIT(&m_short_comm.q);
    pthread_mutex_init(&m_short_comm.lock, NULL);
    pthread_cond_init(&m_short_comm.cond, NULL);
    pthread_create(&td, NULL, (void *(*)(void *))cm_td, &m_short_comm);
 }
--- a/btpd/download_subr.c
+++ b/btpd/download_subr.c
@@ -37,9 +37,8 @@ piece_alloc(struct net *n, uint32_t index)
    struct piece *pc;
    size_t mem, field, blocks;
    unsigned nblocks;
    off_t piece_length = torrent_piece_size(n->tp, index);

    nblocks = (unsigned)ceil((double)piece_length / PIECE_BLOCKLEN);
    nblocks = torrent_piece_blocks(n->tp, index);
    blocks = sizeof(pc->blocks[0]) * nblocks;
    field = (size_t)ceil(nblocks / 8.0);
    mem = sizeof(*pc) + field + blocks;
@@ -63,7 +62,7 @@ piece_alloc(struct net *n, uint32_t index)
    pc->blocks = (struct block *)(pc->down_field + field);
    for (unsigned i = 0; i < nblocks; i++) {
        uint32_t start = i * PIECE_BLOCKLEN;
        uint32_t len = torrent_block_size(pc, i);
        uint32_t len = torrent_block_size(n->tp, index, nblocks, i);
        struct block *blk = &pc->blocks[i];
        blk->pc = pc;
        BTPDQ_INIT(&blk->reqs);
--- a/btpd/torrent.c
+++ b/btpd/torrent.c
@@ -31,13 +31,20 @@ torrent_piece_size(struct torrent *tp, uint32_t index)
 }

 uint32_t
 torrent_block_size(struct piece *pc, uint32_t index)
 torrent_piece_blocks(struct torrent *tp, uint32_t piece)
 {
    if (index < pc->nblocks - 1)
    return ceil(torrent_piece_size(tp, piece) / (double)PIECE_BLOCKLEN);
 }

 uint32_t
 torrent_block_size(struct torrent *tp, uint32_t piece, uint32_t nblocks,
    uint32_t block)
 {
    if (block < nblocks - 1)
        return PIECE_BLOCKLEN;
    else {
        uint32_t allbutlast = PIECE_BLOCKLEN * (pc->nblocks - 1);
        return torrent_piece_size(pc->n->tp, pc->index) - allbutlast;
        uint32_t allbutlast = PIECE_BLOCKLEN * (nblocks - 1);
        return torrent_piece_size(tp, piece) - allbutlast;
    }
 }

@@ -52,11 +59,14 @@ torrent_activate(struct torrent *tp)
 void
 torrent_deactivate(struct torrent *tp)
 {

    tp->state = T_STOPPING;
    tr_stop(tp);
    net_del_torrent(tp);
    cm_stop(tp);
 }

 int
 torrent_create(struct torrent **res, const char *path)
 torrent_load(struct torrent **res, const char *path)
 {
    struct metainfo *mi;
    int error;
@@ -88,16 +98,17 @@ torrent_create(struct torrent **res, const char *path)
    return error;
 }

 void torrent_cm_cb(struct torrent *tp, enum cm_state state)
 void
 torrent_on_cm_started(struct torrent *tp)
 {
    switch (state) {
    case CM_STARTED:
        net_add_torrent(tp);
        tr_start(tp);
        break;
    case CM_STOPPED:
        break;
    case CM_ERROR:
        break;
    }
    net_add_torrent(tp);
    tr_start(tp);
    tp->state = T_ACTIVE;
 }

 void
 torrent_on_cm_stopped(struct torrent *tp)
 {
    assert(tp->state == T_STOPPING);
    tp->state = T_INACTIVE;
 }
--- a/btpd/torrent.h
+++ b/btpd/torrent.h
@@ -25,19 +25,16 @@ struct torrent {

 BTPDQ_HEAD(torrent_tq, torrent);

 int torrent_create(struct torrent **res, const char *path);
 int torrent_load(struct torrent **res, const char *path);
 void torrent_activate(struct torrent *tp);
 void torrent_deactivate(struct torrent *tp);

 off_t torrent_piece_size(struct torrent *tp, uint32_t index);
 uint32_t torrent_block_size(struct piece *pc, uint32_t index);
 off_t torrent_piece_size(struct torrent *tp, uint32_t piece);
 uint32_t torrent_piece_blocks(struct torrent *tp, uint32_t piece);
 uint32_t torrent_block_size(struct torrent *tp, uint32_t piece,
    uint32_t nblocks, uint32_t block);

 enum cm_state {
    CM_STARTED,
    CM_STOPPED,
    CM_ERROR
 };

 void torrent_cm_cb(struct torrent *tp, enum cm_state state);
 void torrent_on_cm_stopped(struct torrent *tp);
 void torrent_on_cm_started(struct torrent *tp);

 #endif