snapraid/cmdline/stream.c

// SPDX-License-Identifier: GPL-3.0-or-later
// Copyright (C) 2011 Andrea Mazzoleni

#include "os/portable.h"

#include "support.h"
#include "util.h"
#include "stream.h"

/****************************************************************************/
/* stream */

unsigned STREAM_SIZE = 64 * 1024;

STREAM* sopen_read(const char* file, int flags)
{
	STREAM* s = malloc_nofail(sizeof(STREAM));
	struct stat st;
	int ret;

	s->handle_size = 1;
	s->handle = malloc_nofail(sizeof(struct stream_handle));
	s->advise = malloc_nofail(sizeof(struct advise_struct));
	s->flags = flags;
	advise_init(&s->advise[0], s->flags & STREAM_FLAGS_ADVISE_MASK);

	int open_flags = O_RDONLY | O_BINARY | advise_flags(&s->advise[0]);
	if ((s->flags & STREAM_FLAGS_NOFOLLOW) != 0)
		open_flags |= O_NOFOLLOW;

	pathcpy(s->handle[0].path, sizeof(s->handle[0].path), file);
	s->handle[0].f = open(file, open_flags);
	if (s->handle[0].f == -1) {
		free(s->advise);
		free(s->handle);
		free(s);
		return 0;
	}

	/* get file info */
	if (fstat(s->handle[0].f, &st) != 0) {
		close(s->handle[0].f);
		free(s->advise);
		free(s->handle);
		free(s);
		return 0;
	}

	s->size = st.st_size;
	ret = advise_open(&s->advise[0], s->handle[0].f);
	if (ret != 0) {
		/* LCOV_EXCL_START */
		close(s->handle[0].f);
		free(s->advise);
		free(s->handle);
		free(s);
		return 0;
		/* LCOV_EXCL_STOP */
	}

	s->buffer_size = STREAM_SIZE;
	if (s->buffer_size > s->size)
		s->buffer_size = s->size; /* allocate only what is necessary */
	s->buffer = malloc_nofail(s->buffer_size);
	mtest_vector(1, s->buffer_size, (void**)&s->buffer);
	s->pos = s->buffer;
	s->end = s->buffer;
	s->state = STREAM_STATE_READ;
	s->state_index = 0;
	s->offset = 0;
	s->offset_uncached = 0;
	s->crc = 0;
	s->crc_uncached = 0;

	return s;
}

STREAM* sopen_multi_write(unsigned count, int flags)
{
	unsigned i;

	STREAM* s = malloc_nofail(sizeof(STREAM));

	s->handle_size = count;
	s->handle = nalloc_nofail(count, sizeof(struct stream_handle));
	s->advise = nalloc_nofail(count, sizeof(struct advise_struct));
	s->flags = flags;

	for (i = 0; i < count; ++i) {
		s->handle[i].f = -1;
		advise_init(&s->advise[i], s->flags & STREAM_FLAGS_ADVISE_MASK);
	}

	s->buffer_size = STREAM_SIZE;
	s->buffer = malloc_nofail(s->buffer_size);
	mtest_vector(1, s->buffer_size, (void**)&s->buffer);
	s->pos = s->buffer;
	s->end = s->buffer + s->buffer_size;
	s->state = STREAM_STATE_WRITE;
	s->state_index = 0;
	s->size = 0;
	s->offset = 0;
	s->offset_uncached = 0;
	s->crc = 0;
	s->crc_uncached = 0;

	return s;
}

int sopen_multi_file(STREAM* s, unsigned i, const char* file)
{
	int f;

	pathcpy(s->handle[i].path, sizeof(s->handle[i].path), file);

	int open_flags = O_WRONLY | O_CREAT | O_EXCL | O_BINARY | advise_flags(&s->advise[i]);

	/* O_EXCL to be resilient ensure to always create a new file and not use a stale link to the original file */
	f = open(file, open_flags, 0600);
	if (f == -1) {
		/* LCOV_EXCL_START */
		return -1;
		/* LCOV_EXCL_STOP */
	}

	{
		int ret;

		ret = advise_open(&s->advise[i], f);
		if (ret != 0) {
			/* LCOV_EXCL_START */
			close(f);
			return -1;
			/* LCOV_EXCL_STOP */
		}
	}

	s->handle[i].f = f;

	return 0;
}

STREAM* sopen_write(const char* file, int flags)
{
	STREAM* s = sopen_multi_write(1, flags);

	if (sopen_multi_file(s, 0, file) != 0) {
		sclose(s);
		return 0;
	}

	return s;
}

int sclose(STREAM* s)
{
	int fail = 0;
	unsigned i;

	if (s->state == STREAM_STATE_WRITE) {
		if (sflush(s) != 0) {
			/* LCOV_EXCL_START */
			fail = 1;
			/* LCOV_EXCL_STOP */
		}
	}

	for (i = 0; i < s->handle_size; ++i) {
		/* if open failed, there is nothing to close */
		if (s->handle[i].f != -1) {
			if (advise_close(&s->advise[i], s->handle[i].f) != 0) {
				/* LCOV_EXCL_START */
				fail = 1;
				/* LCOV_EXCL_STOP */
			}
			if (close(s->handle[i].f) != 0) {
				/* LCOV_EXCL_START */
				fail = 1;
				/* LCOV_EXCL_STOP */
			}
		}
	}

	free(s->advise);
	free(s->handle);
	free(s->buffer);
	free(s);

	if (fail) {
		/* LCOV_EXCL_START */
		return -1;
		/* LCOV_EXCL_STOP */
	}

	return 0;
}

int shandle(STREAM* s)
{
	if (!s->handle_size) {
		/* LCOV_EXCL_START */
		return -1;
		/* LCOV_EXCL_STOP */
	}

	return s->handle[0].f;
}

/**
 * Fill the read stream buffer.
 * \return 0 if at least on char is read, or EOF on error.
 */
static int sfill(STREAM* s)
{
	ssize_t ret;

	if (s->state != STREAM_STATE_READ) {
		/* LCOV_EXCL_START */
		return EOF;
		/* LCOV_EXCL_STOP */
	}

	/* expose the logical size boundary as EOF, just like EOF from the underlying file */
	if (s->offset >= s->size) {
		s->state = STREAM_STATE_EOF;
		return EOF;
	}

	ret = read(s->handle[0].f, s->buffer, s->buffer_size);

	if (ret < 0) {
		/* LCOV_EXCL_START */
		s->state = STREAM_STATE_ERROR;
		return EOF;
		/* LCOV_EXCL_STOP */
	}
	if (ret == 0) {
		/* nothing mode to read */
		s->state = STREAM_STATE_EOF;
		return EOF;
	}
	/* assume that it can be a shorter read, so ret < s->buffer_size doesn't mean that we reached EOF */

	if (advise_read(&s->advise[0], s->handle[0].f, s->offset, ret) != 0) {
		/* LCOV_EXCL_START */
		s->state = STREAM_STATE_ERROR;
		return EOF;
		/* LCOV_EXCL_STOP */
	}

	/* update the crc */
	if (s->flags & STREAM_FLAGS_CRC) {
		s->crc_uncached = s->crc;
		s->crc = crc32c(s->crc, s->buffer, ret);
	}

	/* update the offset */
	s->offset_uncached = s->offset;
	s->offset += ret;

	s->pos = s->buffer;
	s->end = s->buffer + ret;

	return 0;
}

int sdeplete(STREAM* s, unsigned char* last)
{
	/* last four bytes */
	last[0] = 0;
	last[1] = 0;
	last[2] = 0;
	last[3] = 0;

	while (1) {
		/* increase the position up to 4 bytes before the end */
		if (s->end - s->pos >= 4)
			s->pos = s->end - 4;

		/* insert the last 4 bytes */
		while (s->pos < s->end) {
			last[0] = last[1];
			last[1] = last[2];
			last[2] = last[3];
			last[3] = *s->pos++;
		}

		/* fill again the buffer until the end of the file */
		if (sfill(s) != 0) {
			/* on error fail */
			if (serror(s)) {
				/* LCOV_EXCL_START */
				return EOF;
				/* LCOV_EXCL_STOP */
			}

			/* on EOF terminate */
			break;
		}
	}

	return 0;
}

int sflush(STREAM* s)
{
	ssize_t ret;
	ssize_t size;
	ssize_t count;
	unsigned i;

	if (s->state != STREAM_STATE_WRITE) {
		/* LCOV_EXCL_START */
		return EOF;
		/* LCOV_EXCL_STOP */
	}

	size = s->pos - s->buffer;
	if (!size)
		return 0;

	for (i = 0; i < s->handle_size; ++i) {
		count = 0;
		do {
			ret = write(s->handle[i].f, s->buffer + count, size - count);

			if (ret < 0) {
				if (errno == EINTR)
					continue;

				/* LCOV_EXCL_START */
				s->state = STREAM_STATE_ERROR;
				s->state_index = i;
				return EOF;
				/* LCOV_EXCL_STOP */
			}

			if (ret == 0) {
				/* LCOV_EXCL_START */
				errno = EIO;
				s->state = STREAM_STATE_ERROR;
				s->state_index = i;
				return EOF;
				/* LCOV_EXCL_STOP */
			}

			count += ret;
		} while (count < size);
	}

	for (i = 0; i < s->handle_size; ++i) {
		/* ADVISE_DISCARD modes only discard ranges after waiting for writeback. */
		if (advise_write(&s->advise[i], s->handle[i].f, s->offset, size) != 0) {
			/* LCOV_EXCL_START */
			s->state = STREAM_STATE_ERROR;
			s->state_index = i;
			return EOF;
			/* LCOV_EXCL_STOP */
		}
	}

	/*
	 * Update the crc *after* writing the data.
	 *
	 * This must be done after the file write,
	 * to be able to detect memory errors on the buffer,
	 * happening during the write.
	 */
	if (s->flags & STREAM_FLAGS_CRC) {
		s->crc = crc32c(s->crc, s->buffer, size);
		s->crc_uncached = s->crc;
	}

	/* update the offset */
	s->offset += size;
	s->offset_uncached = s->offset;

	s->pos = s->buffer;

	return 0;
}

int64_t stell(STREAM* s)
{
	return s->offset_uncached + (s->pos - s->buffer);
}

uint32_t scrc(STREAM* s)
{
	assert(s->flags & STREAM_FLAGS_CRC);
	return crc32c(s->crc_uncached, s->buffer, s->pos - s->buffer);
}

int sgetc_uncached(STREAM* s)
{
	/* if at the end of the buffer, fill it */
	if (s->pos == s->end && sfill(s) != 0)
		return EOF;
	return *s->pos++;
}

ssize_t sgettok(STREAM* f, char* str, int size)
{
	char* i = str;
	char* send = str + size;
	int c;

	while (1) {
		c = sgetc(f);
		if (c == EOF) {
			break;
		}
		if (c == ' ' || c == '\t') {
			sungetc(c, f);
			break;
		}
		if (c == '\n') {
			/* remove ending carriage return to support the Windows CR+LF format */
			if (i != str && i[-1] == '\r')
				--i;
			sungetc(c, f);
			break;
		}

		*i++ = c;

		if (i == send) {
			/* LCOV_EXCL_START */
			return -1;
			/* LCOV_EXCL_STOP */
		}
	}

	*i = 0;

	return i - str;
}

int sread_uncached(STREAM* f, void* void_data, size_t size)
{
	unsigned char* data = void_data;

	/* standard version using sgetc() */
	while (size--) {
		int c = sgetc(f);
		if (c == EOF) {
			/* LCOV_EXCL_START */
			return -1;
			/* LCOV_EXCL_STOP */
		}

		*data++ = c;
	}

	return 0;
}

ssize_t sgetline(STREAM* f, char* str, int size)
{
	char* i = str;
	char* send = str + size;
	int c;

	/* if there is enough data in memory */
	if (sptrlookup(f, size)) {
		/* optimized version with all the data in memory */
		unsigned char* pos = sptrget(f);

		while (1) {
			c = *pos++;
			if (c == '\n') {
				/* remove ending carriage return to support the Windows CR+LF format */
				if (i != str && i[-1] == '\r')
					--i;
				--pos;
				break;
			}

			if (i + 1 >= send) {
				/* LCOV_EXCL_START */
				*i = 0;
				--pos;
				sptrset(f, pos);
				return -1;
				/* LCOV_EXCL_STOP */
			}

			*i++ = c;
		}

		sptrset(f, pos);
	} else {
		while (1) {
			c = sgetc(f);
			if (c == EOF) {
				/* LCOV_EXCL_START */
				break;
				/* LCOV_EXCL_STOP */
			}

			if (c == '\n') {
				/* remove ending carriage return to support the Windows CR+LF format */
				if (i != str && i[-1] == '\r')
					--i;
				sungetc(c, f);
				break;
			}

			if (i + 1 >= send) {
				/* LCOV_EXCL_START */
				*i = 0;
				sungetc(c, f);
				return -1;
				/* LCOV_EXCL_STOP */
			}

			*i++ = c;
		}
	}

	*i = 0;

	return i - str;
}

ssize_t sgetlasttok(STREAM* f, char* str, int size)
{
	ssize_t ret;

	ret = sgetline(f, str, size);
	if (ret < 0) {
		/* LCOV_EXCL_START */
		return ret;
		/* LCOV_EXCL_STOP */
	}

	while (ret > 0 && (str[ret - 1] == ' ' || str[ret - 1] == '\t'))
		--ret;

	str[ret] = 0;

	return ret;
}

int sgetu32(STREAM* f, uint32_t* value)
{
	int c;

	c = sgetc(f);
	if (c == '0') {
		*value = 0;
		return 0;
	} else if (c >= '1' && c <= '9') {
		uint32_t v;

		v = c - '0';

		c = sgetc(f);
		while (c >= '0' && c <= '9') {
			uint32_t digit;
			if (v > 0xFFFFFFFFU / 10) {
				/* LCOV_EXCL_START */
				/* overflow */
				return -1;
				/* LCOV_EXCL_STOP */
			}
			v *= 10;

			digit = c - '0';
			if (v > 0xFFFFFFFFU - digit) {
				/* LCOV_EXCL_START */
				/* overflow */
				return -1;
				/* LCOV_EXCL_STOP */
			}
			v += digit;

			c = sgetc(f);
		}

		*value = v;

		sungetc(c, f);
		return 0;
	} else {
		/* LCOV_EXCL_START */
		/* nothing read */
		return -1;
		/* LCOV_EXCL_STOP */
	}
}

int sgetb32(STREAM* f, uint32_t* value)
{
	uint32_t v;
	unsigned char b;
	unsigned s;
	int c;

	v = 0;
	s = 0;
loop:
	c = sgetc(f);
	if (c == EOF) {
		/* LCOV_EXCL_START */
		return -1;
		/* LCOV_EXCL_STOP */
	}

	b = (unsigned char)c;
	if ((b & 0x80) == 0) {
		v |= (uint32_t)b << s;
		s += 7;
		if (tommy_unlikely(s >= 32)) {
			/* LCOV_EXCL_START */
			return -1;
			/* LCOV_EXCL_STOP */
		}
		goto loop;
	}

	/*
	 * Check for overflow: at s == 28 only 4 bits (0x0f) fit in uint32_t
	 */
	if (tommy_unlikely(s == 28 && (b & 0x70) != 0)) {
		return -1;
	}

	v |= (uint32_t)(b & 0x7f) << s;

	*value = v;

	return 0;
}

int sgetb64(STREAM* f, uint64_t* value)
{
	uint64_t v;
	unsigned char b;
	unsigned s;
	int c;

	v = 0;
	s = 0;
loop:
	c = sgetc(f);
	if (c == EOF) {
		/* LCOV_EXCL_START */
		return -1;
		/* LCOV_EXCL_STOP */
	}

	b = (unsigned char)c;
	if ((b & 0x80) == 0) {
		v |= (uint64_t)b << s;
		s += 7;
		if (tommy_unlikely(s >= 64)) {
			/* LCOV_EXCL_START */
			return -1;
			/* LCOV_EXCL_STOP */
		}
		goto loop;
	}

	/*
	 * Check for overflow: at s == 63 only 1 bit (0x01) fits in uint64_t
	 */
	if (tommy_unlikely(s == 63 && (b & 0x7e) != 0)) {
		return -1;
	}

	v |= (uint64_t)(b & 0x7f) << s;

	*value = v;

	return 0;
}

int sgetble32(STREAM* f, uint32_t* value)
{
	unsigned char buf[4];

	if (sread(f, buf, 4) != 0) {
		/* LCOV_EXCL_START */
		return -1;
		/* LCOV_EXCL_STOP */
	}

	*value = buf[0] | (uint32_t)buf[1] << 8 | (uint32_t)buf[2] << 16 | (uint32_t)buf[3] << 24;

	return 0;
}

ssize_t sgetbs(STREAM* f, char* str, size_t size)
{
	uint64_t len;

	if (sgetb64(f, &len) < 0) {
		/* LCOV_EXCL_START */
		return -1;
		/* LCOV_EXCL_STOP */
	}

	if (len >= size) {
		/* LCOV_EXCL_START */
		return -1;
		/* LCOV_EXCL_STOP */
	}

	str[len] = 0;

	if (sread(f, str, len) != 0) {
		/* LCOV_EXCL_START */
		return -1;
		/* LCOV_EXCL_STOP */
	}

	return len;
}

int swrite_uncached(const void* void_data, size_t size, STREAM* f)
{
	const unsigned char* data = void_data;

	while (size) {
		size_t available;

		if (f->pos == f->end) {
			if (sflush(f) != 0) {
				/* LCOV_EXCL_START */
				return -1;
				/* LCOV_EXCL_STOP */
			}
		}

		available = (size_t)(f->end - f->pos);
		if (available > size)
			available = size;

		memcpy(f->pos, data, available);
		f->pos += available;
		data += available;
		size -= available;
	}

	return 0;
}

int sputb32(uint32_t value, STREAM* s)
{
	unsigned char b;
	unsigned char buf[16];
	unsigned i;

	i = 0;
loop:
	b = value & 0x7f;
	value >>= 7;

	if (value) {
		buf[i++] = b;
		goto loop;
	}

	buf[i++] = b | 0x80;

	return swrite(buf, i, s);
}

int sputb64(uint64_t value, STREAM* s)
{
	unsigned char b;
	unsigned char buf[16];
	unsigned i;

	i = 0;
loop:
	b = value & 0x7f;
	value >>= 7;

	if (value) {
		buf[i++] = b;
		goto loop;
	}

	buf[i++] = b | 0x80;

	return swrite(buf, i, s);
}

int sputble32(uint32_t value, STREAM* s)
{
	unsigned char buf[4];

	buf[0] = value & 0xFF;
	buf[1] = (value >> 8) & 0xFF;
	buf[2] = (value >> 16) & 0xFF;
	buf[3] = (value >> 24) & 0xFF;

	return swrite(buf, 4, s);
}

int sputbs(const char* str, STREAM* f)
{
	size_t len = strlen(str);

	if (sputb64(len, f) != 0) {
		/* LCOV_EXCL_START */
		return -1;
		/* LCOV_EXCL_STOP */
	}

	return swrite(str, len, f);
}

#if HAVE_FSYNC
int ssync(STREAM* s)
{
	unsigned i;

	for (i = 0; i < s->handle_size; ++i) {
		int ret;

		/*
		 * On macOS, standard fsync() only pushes data from kernel buffers to the
		 * drive's volatile cache without requesting a physical media flush. F_FULLFSYNC
		 * forces the drive controller to commit all cached data to permanent storage.
		 */
#if defined(__APPLE__) && defined(F_FULLFSYNC)
		ret = fcntl(s->handle[i].f, F_FULLFSYNC);
		if (ret != 0 && (errno == ENOTSUP || errno == EINVAL))
			ret = fsync(s->handle[i].f);
#else
		ret = fsync(s->handle[i].f);
#endif
		if (ret != 0) {
			/* LCOV_EXCL_START */
			s->state = STREAM_STATE_ERROR;
			s->state_index = i;
			return -1;
			/* LCOV_EXCL_STOP */
		}
	}

	return 0;
}
#endif

int smtime(STREAM* s, int64_t mtime_sec, int mtime_nsec)
{
	unsigned i;

	for (i = 0; i < s->handle_size; ++i) {
		if (s->handle[i].f != -1 && fmtime(s->handle[i].f, mtime_sec, mtime_nsec) != 0) {
			/* LCOV_EXCL_START */
			s->state = STREAM_STATE_ERROR;
			s->state_index = i;
			return -1;
			/* LCOV_EXCL_STOP */
		}
	}

	return 0;
}