libiio/stream.c

// SPDX-License-Identifier: MIT
/*
 * libiio - Library for interfacing industrial I/O (IIO) devices
 *
 * Copyright (C) 2022 Analog Devices, Inc.
 * Author: Paul Cercueil <paul.cercueil@analog.com>
 */

#include "iio-private.h"
#include "iio/iio.h"

#include <errno.h>
#include <iio/iio-debug.h>
#include <stdbool.h>
#include <stdlib.h>

struct iio_stream {
	struct iio_buffer_stream *buf_stream;
	struct iio_block **blocks;
	size_t nb_blocks;
	bool started, buf_enabled, all_enqueued;
	unsigned int curr;
};

struct iio_stream *
iio_buffer_create_stream(struct iio_buffer *buffer, size_t nb_blocks,
			 size_t samples_count, struct iio_channels_mask *mask)
{
	struct iio_buffer_stream *buf_stream;
	size_t i, sample_size, buf_size;
	struct iio_stream *stream;
	int err;

	if (!nb_blocks || !samples_count || !mask)
		return iio_ptr(-EINVAL);

	stream = zalloc(sizeof(*stream));
	if (!stream)
		return iio_ptr(-ENOMEM);

	sample_size = iio_device_get_sample_size(buffer->dev, mask);
	buf_size = samples_count * sample_size;

	buf_stream = iio_buffer_open(buffer, mask);
	err = iio_err(buf_stream);
	if (err)
		goto err_free_stream;

	stream->blocks = calloc(nb_blocks, sizeof(*stream->blocks));
	if (!stream->blocks) {
		err = -ENOMEM;
		goto err_close_buffer;
	}

	for (i = 0; i < nb_blocks; i++) {
		stream->blocks[i] = iio_buffer_stream_create_block(buf_stream, buf_size);
		err = iio_err(stream->blocks[i]);
		if (err) {
			stream->blocks[i] = NULL;
			goto err_free_stream_blocks;
		}
	}

	stream->buf_stream = buf_stream;
	stream->nb_blocks = nb_blocks;

	return stream;

err_free_stream_blocks:
	for (i = 0; i < nb_blocks; i++)
		if (stream->blocks[i])
			iio_block_destroy(stream->blocks[i]);
	free(stream->blocks);
err_close_buffer:
	iio_buffer_close(buf_stream);
err_free_stream:
	free(stream);
	return iio_ptr(err);
}

void iio_stream_cancel(struct iio_stream *stream)
{
	iio_buffer_stream_cancel(stream->buf_stream);
}

void iio_stream_destroy(struct iio_stream *stream)
{
	size_t i;

	for (i = 0; i < stream->nb_blocks; i++) {
		if (stream->blocks[i]) {
			/* Make sure to dequeue any possible in flight block
			 * before destroying it. Note that if the block is not
			 * enqueued, this returns an error but we really don't
			 * care about it. Making sure the block is dequeued makes
			 * for a much cleaner exit specially in cases where freeing
			 * the blocks is deferred to the default client.
			 */
			iio_block_dequeue(stream->blocks[i], false);
			iio_block_destroy(stream->blocks[i]);
		}
	}

	free(stream->blocks);
	iio_buffer_close(stream->buf_stream);
	free(stream);
}

const struct iio_block *
iio_stream_get_next_block(struct iio_stream *stream)
{
	const struct iio_device *dev = stream->buf_stream->buf->dev;
	bool is_tx = iio_buffer_is_output(stream->buf_stream->buf);
	unsigned int i;
	int err;

	if (!stream->started) {
		for (i = 1; !is_tx && i < stream->nb_blocks; i++) {
			err = iio_block_enqueue(stream->blocks[i], 0, false);
			if (err) {
				dev_perror(dev, err, "Unable to enqueue block");
				return iio_ptr(err);
			}
		}

		stream->started = true;

		if (is_tx)
			return stream->blocks[0];

		stream->all_enqueued = true;
	}

	err = iio_block_enqueue(stream->blocks[stream->curr], 0, false);
	if (err < 0) {
		dev_perror(dev, err, "Unable to enqueue block");
		return iio_ptr(err);
	}

	if (!stream->buf_enabled) {
		err = iio_buffer_stream_start(stream->buf_stream);
		if (err) {
			dev_perror(dev, err, "Unable to enable buffer");
			return iio_ptr(err);
		}

		stream->buf_enabled = true;
	}

	stream->curr = (stream->curr + 1) % stream->nb_blocks;

	stream->all_enqueued |= stream->curr == 0;
	if (stream->all_enqueued) {
		err = iio_block_dequeue(stream->blocks[stream->curr], false);
		if (err < 0) {
			dev_perror(dev, err, "Unable to dequeue block");
			return iio_ptr(err);
		}
	}

	return stream->blocks[stream->curr];
}