<commit_before>#include "block_writer.h"
#include "util.h"

#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <pthread.h>

static pthread_t writer_thread;
static pthread_mutex_t mutex;
static pthread_mutex_t mutex_back;

static bool running = false;
static char *async_buffer = NULL;
static char *unaligned_async_buffer = NULL;
static size_t amount_to_write = 0;

static void *writer_worker(void *void_bw)
{
    struct block_writer *bw = (struct block_writer *) void_bw;

    for (;;) {
        pthread_mutex_lock(&mutex);

        while (amount_to_write > 0) {
            ssize_t rc = write(bw->fd, async_buffer, amount_to_write);
            if (rc < 0) {
                fwup_errx(EXIT_FAILURE, "Need to handle this");

            }
            fprintf(stderr, "wrote %d\n", rc);
            amount_to_write -= rc;
        }

        if (!running)
            break;

        pthread_mutex_unlock(&mutex_back);
    }
    fprintf(stderr, "done %d\n", amount_to_write);
    return NULL;
}

/**
 * @brief Helper for buffering writes into block-sized pieces
 *
 * This is needed since the unzip process creates variable sized chunks that need
 * to be combined before being written. File I/O to Flash media works best in large
 * sized chunks, so cache the writes until they get to that size. Writes MUST be
 * monotically increasing in offset. Seeking backwards is NOT supported.
 *
 * Based on benchmarks and inspection of how Linux writes to Flash media at a low level,
 * the chunks don't need to be on Flash erase block boundaries or Flash erase block size.
 * This doesn't appear to hurt, but I found out that Linux was sending 120 KB blocks (NOT
 * 128 KB blocks) via a USB sniffer no matter how much larger blocks were written. Performance
 * was still very fast.
 *
 * This code also pads files out to block boundaries (512 bytes for all supported media now).
 * This is needed on OSX when writing to /dev/rdiskN devices.
 *
 * @param aw
 * @param fd              the target file to write to
 * @param buffer_size     the size of the buffer (rounded to multiple of 2^log2_block_size)
 * @param log2_block_size 9 for 512 byte blocks
 * @return 0 if ok, -1 if not
 */
int block_writer_init(struct block_writer *bw, int fd, int buffer_size, int log2_block_size)
{
    bw->fd = fd;
    bw->block_size = (1 << log2_block_size);
    bw->block_size_mask = ~((off_t) bw->block_size - 1);
    bw->buffer_size = (buffer_size + ~bw->block_size_mask) & bw->block_size_mask;

    // Buffer alignment required on Linux when files are opened with O_DIRECT.
    bw->unaligned_buffer = (char *) malloc(bw->buffer_size + 4095);
    if (bw->unaligned_buffer == 0)
        ERR_RETURN("Cannot allocate write buffer of %d bytes.", bw->buffer_size);
    bw->buffer = (char *) (((uint64_t) (bw->unaligned_buffer + 4095)) & ~4095);
    bw->write_offset = 0;
    bw->last_write_offset = -1;
    bw->buffer_index = 0;
    bw->added_bytes = 0;

    unaligned_async_buffer = (char *) malloc(bw->buffer_size + 4095);
    if (unaligned_async_buffer == 0)
        ERR_RETURN("Cannot allocate write buffer of %d bytes.", bw->buffer_size);
    async_buffer = (char *) (((uint64_t) (unaligned_async_buffer + 4095)) & ~4095);
    running = true;

    pthread_mutex_init(&mutex, NULL);
    pthread_mutex_lock(&mutex);
    pthread_mutex_init(&mutex_back, NULL);
    if (pthread_create(&writer_thread, NULL, writer_worker, bw))
        fwup_errx(EXIT_FAILURE, "pthread_create");

    return 0;
}

static ssize_t maybe_pwrite(struct block_writer *bw, const char *buf, size_t count)
{
    pthread_mutex_lock(&mutex_back);
    fprintf(stderr, "maybe_pwrite %d\n", amount_to_write);
    if (bw->write_offset != bw->last_write_offset) {
        if (lseek(bw->fd, bw->write_offset, SEEK_SET) < 0) {
            pthread_mutex_unlock(&mutex);
            return -1;
        }

        bw->last_write_offset = bw->write_offset;
    }

    size_t amount_left = count;
    while (amount_left > bw->buffer_size) {
        ssize_t rc = write(bw->fd, buf, bw->buffer_size);
        if (rc <= 0) {
            pthread_mutex_unlock(&mutex);
            return -1;
        }

        bw->last_write_offset += rc;
        buf += rc;
        amount_left -= rc;
    }
    memcpy(async_buffer, buf, amount_left);
    amount_to_write = amount_left;
    pthread_mutex_unlock(&mutex);

    return count;
}

static ssize_t flush_buffer(struct block_writer *bw)
{
    assert(bw->buffer_index <= bw->buffer_size); // Math failed somewhere if we're buffering more than the block_size

    // Make sure that we're writing a whole block or pad it with 0s just in case.
    // Note: this isn't necessary when writing through a cache to a device, but that's
    //       not always the case.
    size_t block_boundary = (bw->buffer_index + ~bw->block_size_mask) & bw->block_size_mask;
    ssize_t added_bytes = bw->added_bytes;
    if (block_boundary != bw->buffer_index) {
        ssize_t padding_bytes = block_boundary - bw->buffer_index;
        memset(&bw->buffer[bw->buffer_index], 0, padding_bytes);
        added_bytes += padding_bytes;
        bw->buffer_index = block_boundary;
    }

    ssize_t rc = maybe_pwrite(bw, bw->buffer, bw->buffer_index); // pwrite(bw->fd, bw->buffer, bw->buffer_index, bw->write_offset);
    if (rc != (ssize_t) bw->buffer_index)
        return -1;

    // Success. Adjust rc to not count the padding.
    bw->write_offset += bw->buffer_index;
    bw->buffer_index = 0;
    bw->added_bytes = 0;
    return rc - added_bytes;
}

/**
 * @brief flush and free the aligned writer data structures
 * @param bw
 * @return number of bytes written or -1 if error
 */
ssize_t block_writer_free(struct block_writer *bw)
{
    // Write any data that's still in the buffer
    ssize_t rc = 0;
    if (bw->buffer_index)
        rc = flush_buffer(bw);

    pthread_mutex_lock(&mutex_back);
    fprintf(stderr, "stopping %d\n", amount_to_write);
    running = false;
    pthread_mutex_unlock(&mutex);

    if (pthread_join(writer_thread, NULL))
        fwup_errx(EXIT_FAILURE, "pthread_join");
    free(unaligned_async_buffer);
    pthread_mutex_destroy(&mutex);
    pthread_mutex_destroy(&mutex_back);

    // Free our buffer and clean up
    free(bw->unaligned_buffer);<commit_msg>Fewer prints<commit_after>#include "block_writer.h"
#include "util.h"

#include <assert.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include <pthread.h>

static pthread_t writer_thread;
static pthread_mutex_t mutex_to;
static pthread_mutex_t mutex_back;

static bool running = false;
static char *async_buffer = NULL;
static char *unaligned_async_buffer = NULL;
static size_t amount_to_write = 0;

static void *writer_worker(void *void_bw)
{
    struct block_writer *bw = (struct block_writer *) void_bw;

    for (;;) {
        pthread_mutex_lock(&mutex_to);

        while (amount_to_write > 0) {
            ssize_t rc = write(bw->fd, async_buffer, amount_to_write);
            if (rc < 0) {
                fwup_errx(EXIT_FAILURE, "Need to handle this");

            }
            amount_to_write -= rc;
        }

        if (!running)
            break;

        pthread_mutex_unlock(&mutex_back);
    }
    return NULL;
}

/**
 * @brief Helper for buffering writes into block-sized pieces
 *
 * This is needed since the unzip process creates variable sized chunks that need
 * to be combined before being written. File I/O to Flash media works best in large
 * sized chunks, so cache the writes until they get to that size. Writes MUST be
 * monotically increasing in offset. Seeking backwards is NOT supported.
 *
 * Based on benchmarks and inspection of how Linux writes to Flash media at a low level,
 * the chunks don't need to be on Flash erase block boundaries or Flash erase block size.
 * This doesn't appear to hurt, but I found out that Linux was sending 120 KB blocks (NOT
 * 128 KB blocks) via a USB sniffer no matter how much larger blocks were written. Performance
 * was still very fast.
 *
 * This code also pads files out to block boundaries (512 bytes for all supported media now).
 * This is needed on OSX when writing to /dev/rdiskN devices.
 *
 * @param aw
 * @param fd              the target file to write to
 * @param buffer_size     the size of the buffer (rounded to multiple of 2^log2_block_size)
 * @param log2_block_size 9 for 512 byte blocks
 * @return 0 if ok, -1 if not
 */
int block_writer_init(struct block_writer *bw, int fd, int buffer_size, int log2_block_size)
{
    bw->fd = fd;
    bw->block_size = (1 << log2_block_size);
    bw->block_size_mask = ~((off_t) bw->block_size - 1);
    bw->buffer_size = (buffer_size + ~bw->block_size_mask) & bw->block_size_mask;

    // Buffer alignment required on Linux when files are opened with O_DIRECT.
    bw->unaligned_buffer = (char *) malloc(bw->buffer_size + 4095);
    if (bw->unaligned_buffer == 0)
        ERR_RETURN("Cannot allocate write buffer of %d bytes.", bw->buffer_size);
    bw->buffer = (char *) (((uint64_t) (bw->unaligned_buffer + 4095)) & ~4095);
    bw->write_offset = 0;
    bw->last_write_offset = -1;
    bw->buffer_index = 0;
    bw->added_bytes = 0;

    unaligned_async_buffer = (char *) malloc(bw->buffer_size + 4095);
    if (unaligned_async_buffer == 0)
        ERR_RETURN("Cannot allocate write buffer of %d bytes.", bw->buffer_size);
    async_buffer = (char *) (((uint64_t) (unaligned_async_buffer + 4095)) & ~4095);
    running = true;

    pthread_mutex_init(&mutex_to, NULL);
    pthread_mutex_lock(&mutex_to);
    pthread_mutex_init(&mutex_back, NULL);
    if (pthread_create(&writer_thread, NULL, writer_worker, bw))
        fwup_errx(EXIT_FAILURE, "pthread_create");

    return 0;
}

static ssize_t maybe_pwrite(struct block_writer *bw, const char *buf, size_t count)
{
    pthread_mutex_lock(&mutex_back);
    if (bw->write_offset != bw->last_write_offset) {
        if (lseek(bw->fd, bw->write_offset, SEEK_SET) < 0) {
            pthread_mutex_unlock(&mutex_to);
            return -1;
        }

        bw->last_write_offset = bw->write_offset;
    }

    size_t amount_left = count;
    while (amount_left > bw->buffer_size) {
        ssize_t rc = write(bw->fd, buf, bw->buffer_size);
        if (rc <= 0) {
            pthread_mutex_unlock(&mutex_to);
            return -1;
        }

        bw->last_write_offset += rc;
        buf += rc;
        amount_left -= rc;
    }
    memcpy(async_buffer, buf, amount_left);
    amount_to_write = amount_left;
    pthread_mutex_unlock(&mutex_to);

    return count;
}

static ssize_t flush_buffer(struct block_writer *bw)
{
    assert(bw->buffer_index <= bw->buffer_size); // Math failed somewhere if we're buffering more than the block_size

    // Make sure that we're writing a whole block or pad it with 0s just in case.
    // Note: this isn't necessary when writing through a cache to a device, but that's
    //       not always the case.
    size_t block_boundary = (bw->buffer_index + ~bw->block_size_mask) & bw->block_size_mask;
    ssize_t added_bytes = bw->added_bytes;
    if (block_boundary != bw->buffer_index) {
        ssize_t padding_bytes = block_boundary - bw->buffer_index;
        memset(&bw->buffer[bw->buffer_index], 0, padding_bytes);
        added_bytes += padding_bytes;
        bw->buffer_index = block_boundary;
    }

    ssize_t rc = maybe_pwrite(bw, bw->buffer, bw->buffer_index); // pwrite(bw->fd, bw->buffer, bw->buffer_index, bw->write_offset);
    if (rc != (ssize_t) bw->buffer_index)
        return -1;

    // Success. Adjust rc to not count the padding.
    bw->write_offset += bw->buffer_index;
    bw->buffer_index = 0;
    bw->added_bytes = 0;
    return rc - added_bytes;
}

/**
 * @brief flush and free the aligned writer data structures
 * @param bw
 * @return number of bytes written or -1 if error
 */
ssize_t block_writer_free(struct block_writer *bw)
{
    // Write any data that's still in the buffer
    ssize_t rc = 0;
    if (bw->buffer_index)
        rc = flush_buffer(bw);

    pthread_mutex_lock(&mutex_back);
    running = false;
    pthread_mutex_unlock(&mutex_to);

    if (pthread_join(writer_thread, NULL))
        fwup_errx(EXIT_FAILURE, "pthread_join");
    free(unaligned_async_buffer);
    pthread_mutex_destroy(&mutex_to);
    pthread_mutex_destroy(&mutex_back);

    // Free our buffer and clean up
    free(bw->unaligned_buffer);
{'fraction_non_alphanumeric': 0.1078985624293329, 'fraction_numerical': 0.011468260377968019, 'mean_word_length': 3.1470194239785667, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 125, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2836929', 'n_tokens_mistral': 4201, 'n_tokens_neox': 3984, 'n_words': 1184}