<commit_before>/// @file       rm.c
/// @brief      Simple rsync algorithm implementation
///		as described in Tridgell A., "Efficient Algorithms
///		for Sorting and Synchronization", 1999.
/// @details	Common definitions used by sender and receiver.
/// @author     <NAME> <EMAIL> at <EMAIL>
/// @version    0.1.2
/// @date       1 Jan 2016 07:50 PM
/// @copyright  LGPLv2.1v2.1


#include "rm.h"


uint32_t
rm_fast_check_block(const unsigned char *data, uint32_t len)
{
#ifdef DEBUG
	uint32_t res;
#endif
	uint32_t	r1, r2, i;

	assert(data != NULL);
	r1 = 0;
	r2 = 0;
	i = 0;
	for (; i < len; ++i)
	{
		r1 = (r1 + data[i]) % RM_FASTCHECK_MODULUS;
		r2 = (r2 + r1) % RM_FASTCHECK_MODULUS;
	}
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}

uint32_t
rm_adler32_1(const unsigned char *data, uint32_t len)
{
#ifdef DEBUG
	uint32_t res;
#endif
	uint32_t	r1, r2, i;

	assert(data != NULL);
	r1 = 1;
	r2 = 0;
	i = 0;
	for (; i < len; ++i)
	{
		r1 = (r1 + data[i]) % RM_ADLER32_MODULUS;
		r2 = (r2 + r1) % RM_ADLER32_MODULUS;
	}
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}

#define RM_DO1(buf, i)	{ r1 += buf[i]; r2 += r1; }
#define RM_DO2(buf, i)	RM_DO1(buf, i); RM_DO1(buf, i + 1)
#define RM_DO4(buf, i)	RM_DO2(buf, i); RM_DO2(buf, i + 2)
#define RM_DO8(buf, i)	RM_DO4(buf, i); RM_DO4(buf, i + 4)
#define RM_DO16(buf)	RM_DO8(buf,0); RM_DO8(buf,8)

uint32_t
rm_adler32_2(uint32_t adler, const unsigned char *data, uint32_t L)
{
#ifdef DEBUG
	uint32_t res;
#endif
	uint32_t	k;
	uint32_t	r1 = adler & 0xffff;
	uint32_t	r2 = (adler >> 16) & 0xffff;

	assert(data != NULL);
	while(L)
	{
		k = L < RM_ADLER32_NMAX ? L : RM_ADLER32_NMAX;
		L -= k;
		// process 16bits blocks
		while (k >= 16)
		{
			RM_DO16(data);
			data += 16;
			k-=16;
		}
		// remainder
		if (k > 0)
		{
			do
			{
				r1 += *data++;
				r2 += r1;
			} while (--k);
		}

		r1 = r1  % RM_ADLER32_MODULUS;
		r2 = r2  % RM_ADLER32_MODULUS;
	}
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}

// rolling adler with prime modulus won't work
uint32_t
rm_adler32_roll(uint32_t adler, unsigned char a_k,
		unsigned char a_kL, uint32_t L)
{
#ifdef DEBUG
	uint32_t res;
#endif
	uint32_t	r1, r2;
	// r1 and r2 from adler on block [k,k+L]]
	r1 = adler & 0xFFFF;
	r2 = (adler >> 16) & 0xFFFF;
	// update
	r1 = (r1 - a_k + a_kL) % RM_ADLER32_MODULUS;
	r2 = (r2 + r1 - L*a_k - 1) % RM_ADLER32_MODULUS;
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}

// modulus MUST be even
uint32_t
rm_fast_check_roll(uint32_t adler, unsigned char a_k,
		unsigned char a_kL, uint32_t L)
{
#ifdef DEBUG
	uint32_t res;
#endif
	uint32_t	r1, r2;
	// r1 and r2 from adler on block [k,k+L]]
	r1 = adler & 0xFFFF;
	r2 = (adler >> 16) & 0xFFFF;
	// update
	r1 = (r1 - a_k + a_kL) % RM_FASTCHECK_MODULUS;
	r2 = (r2 + r1 - L*a_k) % RM_FASTCHECK_MODULUS;
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}

uint32_t
rm_rolling_ch(const unsigned char *data, uint32_t len,
				uint32_t M)
{
	uint32_t	r1, r2, i;

#ifdef DEBUG
	uint32_t res;
#endif
	assert(data != NULL);
	r1 = r2 = 0;
	i = 0;
	for (; i < len; ++i)
	{
		r1 = (r1 + data[i]) % M;
		r2 = (r2 + r1) % M;
	}
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}
 

void
rm_md5(const unsigned char *data, uint32_t len,
		unsigned char res[16])
{
	MD5_CTX ctx;
	md5_init(&ctx);
	md5_update(&ctx, data, len);
	md5_final(&ctx, res);
}

int
rm_copy_buffered(FILE *x, FILE *y, size_t bytes_n)
{
    size_t read, read_exp;
    char buf[RM_L1_CACHE_RECOMMENDED];

    read_exp = RM_L1_CACHE_RECOMMENDED < bytes_n ?
                        RM_L1_CACHE_RECOMMENDED : bytes_n;
    while ((read = fread(buf, 1, read_exp, x)) == read_exp)
    {
        if (fwrite(buf, 1, read_exp, y) != read_exp)
            return -1;
        bytes_n -= read;
        read_exp = RM_L1_CACHE_RECOMMENDED < bytes_n ?
                        RM_L1_CACHE_RECOMMENDED : bytes_n;
    }

    if (read == 0)
    {
        /* EOF reached */
        return 0;
    }

    if (ferror(x) != 0)
        return -2;
    if (ferror(y) != 0)
        return -3;

    return 0;
}

size_t
rm_fpread(void *buf, size_t size, size_t items_n, size_t offset, FILE *f)
{
    if (fseek(f, offset, SEEK_SET) != 0)
        return 0;
    return fread(buf, size, items_n, f);
}

int
rm_rolling_ch_proc(const struct rm_session *s, const struct twhlist_head *h,
        FILE *f_x, rm_delta_f *delta_f, uint32_t L, size_t from)
{
    uint32_t        hash;
    unsigned char   buf[L];
    int             fd;
	struct stat     fs;
    size_t          file_sz, read_left, read_now, read;
    uint8_t         match;
    const struct rm_ch_ch_ref_hlink   *e;
    struct rm_ch_ch ch;
    struct rm_roll_proc_cb_arg  cb_arg;         /* callback argument */
    struct rm_delta_e           *delta_e;
    size_t                      raw_bytes_n;
    unsigned char               *raw_bytes;     /* buffer */
    size_t                      a_k_pos, a_kL_pos;
    unsigned char               a_k, a_kL;      /* bytes to remove/add from rolling checksum */
    size_t          collisions_1st_level = 0;
    size_t          collisions_2nd_level = 0;

    if (L == 0 || s == NULL)
        return -1;
    /* setup callback argument */
    cb_arg.s = s;

    /* get file size */
    fd = fileno(f_x);
    if (fstat(fd, &fs) != 0)
    {
        /* Can't fstat file */
        return -2;
    }
    file_sz = fs.st_size;
    read_left = file_sz - from;
    if (read_left == 0)
    {
        goto end;
    }
    read_now = rm_min(L, read_left);

    read = fread(buf, 1, read_now, f_x);
    if (read != read_now)
    {
        return -3;
    }

    /* 1. initial checksum and it's hash */
    ch.f_ch = rm_fast_check_block(buf, L);
    hash = twhash_min(ch.f_ch, RM_NONOVERLAPPING_HASH_BITS);

    a_k_pos = 0;
    a_kL_pos = L;

    /* roll hash, lookup in table, produce delta elemenets */
    do
    {
        match = 0;

        /* buffer for a raw bytes, callback will take ownership of this! */
        raw_bytes = malloc(L * sizeof(*raw_bytes));
        if (raw_bytes == NULL)
        {
            return -4;
        }
        memset(raw_bytes, 0, L * sizeof(*raw_bytes));
        raw_bytes_n = 0;

        /* hash lookup */
        twhlist_for_each_entry(e, &h[hash], hlink)
        {
            /* hit 1, 1st Level match, hashtable hash match */
            if (e->data.ch_ch.f_ch == ch.f_ch)
            {
                /* hit 2, 2nd Level match, fast rolling checksum match */
                /* compute strong checksum */
                rm_md5(buf, read, ch.s_ch.data);
                if (0 == memcmp(&e->data.ch_ch.s_ch.data, &ch.s_ch.data,
                                                        RM_STRONG_CHECK_BYTES))
                {
                    /* hit 3, 3rd Level match, strong checksum match */
                    /* tx e->data.ref */
                    /* OK, FOUND */
                    match = 1;
                    break;
                } else {
                    /* 2nd Level collision, fast checksum match but strong checksum doesn't */
                    ++collisions_2nd_level;
                }
            } else {
                /* 1st Level collision, fast checksums are different but hashed
                * to the same bucket */
                ++collisions_1st_level;
            }
        }
        /* build delta, TODO free in callback! */
        /* found a match? */
        if (match == 1)
        {
            // TODO tx RM_DELTA_ELEMENT_REFERENCE, move file pointer accordingly
            /* any raw bytes buffered? */
            if (raw_bytes_n > 0)
            {
                /* send them first */
                delta_e = malloc(sizeof *delta_e);
                if (delta_e == NULL)
                {
                    return -5;
                }
                delta_e->type = RM_DELTA_ELEMENT_RAW_BYTES;
                delta_e->ref = 0;
                delta_e->raw_bytes = raw_bytes;
                delta_e->raw_bytes_n = raw_bytes_n;         /* move ownership, TODO cleanup in callback! */
                /* tx, signal delta_rx_tid, etc */
                cb_arg.delta_e = delta_e;
                delta_f(&cb_arg);                           /* TX, enqueue delta */
                /* cleanup */
                raw_bytes = NULL;
                raw_bytes_n = 0;
            }
            /* tx delta ref */
            delta_e = malloc(sizeof *delta_e);
            if (delta_e == NULL)
            {
                return -7;
            }
            delta_e->type = RM_DELTA_ELEMENT_REFERENCE;
            delta_e->ref = e->data.ref;
            delta_e->raw_bytes = NULL;
            delta_e->raw_bytes_n = 0;
            cb_arg.delta_e = delta_e;
            delta_f(&cb_arg);                               /* TX, enqueue delta */
        } else {
            // tx bytes, TODO consider some buffering strategy
            if (raw_bytes == NULL)
            {
                /* alloc new buffer */
                raw_bytes = malloc(L * sizeof(*raw_bytes));
                if (raw_bytes == NULL)
                {
                    return -8;
                }
                memset(raw_bytes, 0, L * sizeof(*raw_bytes));
                raw_bytes_n = 0;
            }
            /* copy byte into the buffer*/
            raw_bytes[raw_bytes_n] = a_k;
            ++raw_bytes_n;
            /* tx? */
            if ((raw_bytes_n == L) || (read_left == 0))     /* TODO there will be more conditions on final transmit here! */
            {
                /* tx */
                delta_e = malloc(sizeof *delta_e);
                if (delta_e == NULL)
                {
                    return -9;
                }
                delta_e->type = RM_DELTA_ELEMENT_RAW_BYTES;
                delta_e->ref = 0;
                delta_e->raw_bytes = raw_bytes;
                delta_e->raw_bytes_n = raw_bytes_n;         /* move ownership, TODO cleanup in callback! */
                /* tx, signal delta_rx_tid, etc */
                cb_arg.delta_e = delta_e;
                delta_f(&cb_arg);                           /* TX, enqueue delta */
                /* cleanup */
                raw_bytes = NULL;
                raw_bytes_n = 0;
            }
        }

        /* roll */
        /* read a_k, a_kL bytes */
        if (a_k_pos < L)
        {
            /* we have read L bytes at the beginning */
            a_k = buf[a_k_pos];
        } else {
            if (rm_fpread(&a_k, sizeof(unsigned char), 1, a_k_pos, f_x) != 1)
            {
                return -10;
            }
        }
        if (rm_fpread(&a_kL, sizeof(unsigned char), 1, a_kL_pos, f_x) != 1)
        {
            return -10;
        }
        ch.f_ch = rm_fast_check_roll(ch.f_ch, a_k, a_kL, L);

        /* move */
        ++a_k_pos;
        ++a_kL_pos;

        read_left = 0;// temporary
    } while (read_left > 0);

end:
    return 0;
}

int
rm_launch_thread(pthread_t *t, void*(*f)(void*), void *arg, int detachstate)
{
	int                 err;
	pthread_attr_t      attr;

	err = pthread_attr_init(&attr);
	if (err != 0)
    {
		return -1;
    }
	err = pthread_attr_setdetachstate(&attr, detachstate);
	if (err != 0)
    {
		goto fail;
    }
	err = pthread_create(t, &attr, f, arg);
	if (err != 0)
    {
        goto fail;
    }
    return 0;
fail:
	pthread_attr_destroy(&attr);
	return -1;
}
<commit_msg>4 35 45 0.1.2 Roll checksum<commit_after>/// @file       rm.c
/// @brief      Simple rsync algorithm implementation
///		as described in Tridgell A., "Efficient Algorithms
///		for Sorting and Synchronization", 1999.
/// @details	Common definitions used by sender and receiver.
/// @author     <NAME> <EMAIL>
/// @version    0.1.2
/// @date       1 Jan 2016 07:50 PM
/// @copyright  LGPLv2.1v2.1


#include "rm.h"


uint32_t
rm_fast_check_block(const unsigned char *data, uint32_t len)
{
#ifdef DEBUG
	uint32_t res;
#endif
	uint32_t	r1, r2, i;

	assert(data != NULL);
	r1 = 0;
	r2 = 0;
	i = 0;
	for (; i < len; ++i)
	{
		r1 = (r1 + data[i]) % RM_FASTCHECK_MODULUS;
		r2 = (r2 + r1) % RM_FASTCHECK_MODULUS;
	}
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}

uint32_t
rm_adler32_1(const unsigned char *data, uint32_t len)
{
#ifdef DEBUG
	uint32_t res;
#endif
	uint32_t	r1, r2, i;

	assert(data != NULL);
	r1 = 1;
	r2 = 0;
	i = 0;
	for (; i < len; ++i)
	{
		r1 = (r1 + data[i]) % RM_ADLER32_MODULUS;
		r2 = (r2 + r1) % RM_ADLER32_MODULUS;
	}
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}

#define RM_DO1(buf, i)	{ r1 += buf[i]; r2 += r1; }
#define RM_DO2(buf, i)	RM_DO1(buf, i); RM_DO1(buf, i + 1)
#define RM_DO4(buf, i)	RM_DO2(buf, i); RM_DO2(buf, i + 2)
#define RM_DO8(buf, i)	RM_DO4(buf, i); RM_DO4(buf, i + 4)
#define RM_DO16(buf)	RM_DO8(buf,0); RM_DO8(buf,8)

uint32_t
rm_adler32_2(uint32_t adler, const unsigned char *data, uint32_t L)
{
#ifdef DEBUG
	uint32_t res;
#endif
	uint32_t	k;
	uint32_t	r1 = adler & 0xffff;
	uint32_t	r2 = (adler >> 16) & 0xffff;

	assert(data != NULL);
	while(L)
	{
		k = L < RM_ADLER32_NMAX ? L : RM_ADLER32_NMAX;
		L -= k;
		// process 16bits blocks
		while (k >= 16)
		{
			RM_DO16(data);
			data += 16;
			k-=16;
		}
		// remainder
		if (k > 0)
		{
			do
			{
				r1 += *data++;
				r2 += r1;
			} while (--k);
		}

		r1 = r1  % RM_ADLER32_MODULUS;
		r2 = r2  % RM_ADLER32_MODULUS;
	}
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}

// rolling adler with prime modulus won't work
uint32_t
rm_adler32_roll(uint32_t adler, unsigned char a_k,
		unsigned char a_kL, uint32_t L)
{
#ifdef DEBUG
	uint32_t res;
#endif
	uint32_t	r1, r2;
	// r1 and r2 from adler on block [k,k+L]]
	r1 = adler & 0xFFFF;
	r2 = (adler >> 16) & 0xFFFF;
	// update
	r1 = (r1 - a_k + a_kL) % RM_ADLER32_MODULUS;
	r2 = (r2 + r1 - L*a_k - 1) % RM_ADLER32_MODULUS;
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}

// modulus MUST be even
uint32_t
rm_fast_check_roll(uint32_t adler, unsigned char a_k,
		unsigned char a_kL, uint32_t L)
{
#ifdef DEBUG
	uint32_t res;
#endif
	uint32_t	r1, r2;
	// r1 and r2 from adler on block [k,k+L]]
	r1 = adler & 0xFFFF;
	r2 = (adler >> 16) & 0xFFFF;
	// update
	r1 = (r1 - a_k + a_kL) % RM_FASTCHECK_MODULUS;
	r2 = (r2 + r1 - L*a_k) % RM_FASTCHECK_MODULUS;
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}

uint32_t
rm_rolling_ch(const unsigned char *data, uint32_t len,
				uint32_t M)
{
	uint32_t	r1, r2, i;

#ifdef DEBUG
	uint32_t res;
#endif
	assert(data != NULL);
	r1 = r2 = 0;
	i = 0;
	for (; i < len; ++i)
	{
		r1 = (r1 + data[i]) % M;
		r2 = (r2 + r1) % M;
	}
#ifdef DEBUG
	res = (r2 << 16) | r1;
	return res;
#endif
	return (r2 << 16) | r1;
}
 

void
rm_md5(const unsigned char *data, uint32_t len,
		unsigned char res[16])
{
	MD5_CTX ctx;
	md5_init(&ctx);
	md5_update(&ctx, data, len);
	md5_final(&ctx, res);
}

int
rm_copy_buffered(FILE *x, FILE *y, size_t bytes_n)
{
    size_t read, read_exp;
    char buf[RM_L1_CACHE_RECOMMENDED];

    read_exp = RM_L1_CACHE_RECOMMENDED < bytes_n ?
                        RM_L1_CACHE_RECOMMENDED : bytes_n;
    while ((read = fread(buf, 1, read_exp, x)) == read_exp)
    {
        if (fwrite(buf, 1, read_exp, y) != read_exp)
            return -1;
        bytes_n -= read;
        read_exp = RM_L1_CACHE_RECOMMENDED < bytes_n ?
                        RM_L1_CACHE_RECOMMENDED : bytes_n;
    }

    if (read == 0)
    {
        /* EOF reached */
        return 0;
    }

    if (ferror(x) != 0)
        return -2;
    if (ferror(y) != 0)
        return -3;

    return 0;
}

size_t
rm_fpread(void *buf, size_t size, size_t items_n, size_t offset, FILE *f)
{
    if (fseek(f, offset, SEEK_SET) != 0)
        return 0;
    return fread(buf, size, items_n, f);
}

int
rm_rolling_ch_proc(const struct rm_session *s, const struct twhlist_head *h,
        FILE *f_x, rm_delta_f *delta_f, uint32_t L, size_t from)
{
    uint32_t        hash;
    unsigned char   buf[L];
    int             fd;
	struct stat     fs;
    size_t          file_sz, read_left, read_now, read;
    uint8_t         match;
    const struct rm_ch_ch_ref_hlink   *e;
    struct rm_ch_ch ch;
    struct rm_roll_proc_cb_arg  cb_arg;         /* callback argument */
    struct rm_delta_e           *delta_e;
    size_t                      raw_bytes_n;
    unsigned char               *raw_bytes;     /* buffer */
    size_t                      a_k_pos, a_kL_pos;
    unsigned char               a_k, a_kL;      /* bytes to remove/add from rolling checksum */
    size_t          collisions_1st_level = 0;
    size_t          collisions_2nd_level = 0;

    if (L == 0 || s == NULL)
        return -1;
    /* setup callback argument */
    cb_arg.s = s;

    /* get file size */
    fd = fileno(f_x);
    if (fstat(fd, &fs) != 0)
    {
        /* Can't fstat file */
        return -2;
    }
    file_sz = fs.st_size;
    read_left = file_sz - from;
    if (read_left == 0)
    {
        goto end;
    }
    read_now = rm_min(L, read_left);

    read = fread(buf, 1, read_now, f_x);
    if (read != read_now)
    {
        return -3;
    }

    if (read_left < L)
    {
        /* TODO copy all raw bytes in single delta */
        goto copy_tail;
    }

    /* 1. initial checksum */
    ch.f_ch = rm_fast_check_block(buf, L);

    a_k_pos = 0;
    a_kL_pos = L;

    /* roll hash, lookup in table, produce delta elemenets */
    do
    {
        match = 0;

        /* buffer for a raw bytes, callback will take ownership of this! */
        raw_bytes = malloc(L * sizeof(*raw_bytes));
        if (raw_bytes == NULL)
        {
            return -4;
        }
        memset(raw_bytes, 0, L * sizeof(*raw_bytes));
        raw_bytes_n = 0;

        /* hash lookup */
        hash = twhash_min(ch.f_ch, RM_NONOVERLAPPING_HASH_BITS);
        twhlist_for_each_entry(e, &h[hash], hlink)
        {
            /* hit 1, 1st Level match, hashtable hash match */
            if (e->data.ch_ch.f_ch == ch.f_ch)
            {
                /* hit 2, 2nd Level match, fast rolling checksum match */
                /* compute strong checksum */
                rm_md5(buf, read, ch.s_ch.data);
                if (0 == memcmp(&e->data.ch_ch.s_ch.data, &ch.s_ch.data,
                                                        RM_STRONG_CHECK_BYTES))
                {
                    /* hit 3, 3rd Level match, strong checksum match */
                    /* tx e->data.ref */
                    /* OK, FOUND */
                    match = 1;
                    break;
                } else {
                    /* 2nd Level collision, fast checksum match but strong checksum doesn't */
                    ++collisions_2nd_level;
                }
            } else {
                /* 1st Level collision, fast checksums are different but hashed
                * to the same bucket */
                ++collisions_1st_level;
            }
        }
        /* build delta, TODO free in callback! */
        /* found a match? */
        if (match == 1)
        {
            // TODO tx RM_DELTA_ELEMENT_REFERENCE, move file pointer accordingly
            /* any raw bytes buffered? */
            if (raw_bytes_n > 0)
            {
                /* send them first */
                delta_e = malloc(sizeof *delta_e);
                if (delta_e == NULL)
                {
                    return -5;
                }
                delta_e->type = RM_DELTA_ELEMENT_RAW_BYTES;
                delta_e->ref = 0;
                delta_e->raw_bytes = raw_bytes;
                delta_e->raw_bytes_n = raw_bytes_n;         /* move ownership, TODO cleanup in callback! */
                /* tx, signal delta_rx_tid, etc */
                cb_arg.delta_e = delta_e;
                delta_f(&cb_arg);                           /* TX, enqueue delta */
                /* cleanup */
                raw_bytes = NULL;
                raw_bytes_n = 0;
            }
            /* tx delta ref */
            delta_e = malloc(sizeof *delta_e);
            if (delta_e == NULL)
            {
                return -7;
            }
            delta_e->type = RM_DELTA_ELEMENT_REFERENCE;
            delta_e->ref = e->data.ref;
            delta_e->raw_bytes = NULL;
            delta_e->raw_bytes_n = 0;
            cb_arg.delta_e = delta_e;
            delta_f(&cb_arg);                               /* TX, enqueue delta */
            read_left -= L;
        } else {
            /* raw byte */
            if (raw_bytes == NULL)
            {
                /* alloc new buffer */
                raw_bytes = malloc(L * sizeof(*raw_bytes));
                if (raw_bytes == NULL)
                {
                    return -8;
                }
                memset(raw_bytes, 0, L * sizeof(*raw_bytes));
                raw_bytes_n = 0;
            }
            /* copy byte into the buffer*/
            raw_bytes[raw_bytes_n] = a_k;
            read_left -= 1;
            ++raw_bytes_n;
            /* tx? */
            if ((raw_bytes_n == L) || (read_left == 0))     /* TODO there will be more conditions on final transmit here! */
            {
                /* tx */
                delta_e = malloc(sizeof *delta_e);
                if (delta_e == NULL)
                {
                    return -9;
                }
                delta_e->type = RM_DELTA_ELEMENT_RAW_BYTES;
                delta_e->ref = 0;
                delta_e->raw_bytes = raw_bytes;
                delta_e->raw_bytes_n = raw_bytes_n;         /* move ownership, TODO cleanup in callback! */
                /* tx, signal delta_rx_tid, etc */
                cb_arg.delta_e = delta_e;
                delta_f(&cb_arg);                           /* TX, enqueue delta */
                /* cleanup */
                raw_bytes = NULL;
                raw_bytes_n = 0;
            }
        }

        /* roll */
        if (match == 1)
        {
            read_now = rm_min(L, read_left);
            if (read_now < L)
            {
                goto copy_tail;
            }
            read = rm_fpread(buf, 1, L, a_kL_pos - 1, f_x);
            if (read != read_now)
            {
                return -10;
            }
            /* fast checksum */
            ch.f_ch = rm_fast_check_block(buf, L);

            /* move */
            a_k_pos = a_kL_pos;
            a_kL_pos = a_k + L;
        } else {
            /* read a_k, a_kL bytes */
            if (a_k_pos < L)
            {
                /* we have read L bytes at the beginning */
                a_k = buf[a_k_pos];
            } else {
                if (rm_fpread(&a_k, sizeof(unsigned char), 1, a_k_pos, f_x) != 1)
                {
                    return -11;
                }
            }
            if (a_kL_pos < file_sz)
            {
                /* usual case */
                if (rm_fpread(&a_kL, sizeof(unsigned char), 1, a_kL_pos, f_x) != 1)
                {
                    return -12;
                }
                ch.f_ch = rm_fast_check_roll(ch.f_ch, a_k, a_kL, L);

                /* move */
                ++a_k_pos;
                ++a_kL_pos;
            } else {
                /* TODO */
                /* we are on the tail, this must be handled in a different way */
            }
        }
    } while (read_left > 0);

end:
    return 0;
copy_tail:
    return 0;
}

int
rm_launch_thread(pthread_t *t, void*(*f)(void*), void *arg, int detachstate)
{
	int                 err;
	pthread_attr_t      attr;

	err = pthread_attr_init(&attr);
	if (err != 0)
    {
		return -1;
    }
	err = pthread_attr_setdetachstate(&attr, detachstate);
	if (err != 0)
    {
		goto fail;
    }
	err = pthread_create(t, &attr, f, arg);
	if (err != 0)
    {
        goto fail;
    }
    return 0;
fail:
	pthread_attr_destroy(&attr);
	return -1;
}

{'fraction_non_alphanumeric': 0.12093817697050492, 'fraction_numerical': 0.030244959173471087, 'mean_word_length': 1.5597142247814033, 'pattern_counts': {'":': 0, '<': 73, '<?xml version=': 0, '>': 60, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5839837', 'n_tokens_mistral': 9514, 'n_tokens_neox': 8765, 'n_words': 2193}