michaelh/zephyr
1
0
Fork 0
Code Releases Activity

kernel: Rewrite k_pipe_* API

Browse source 
The `k_pipe_*` API has been reworked to provide a more consistent and
intuitive interface. The new API aims to provide a simple to use byte
stream interface that is more in line with the POSIX pipe API.
The previous API has been deprecated and will be removed in a future
release.

Signed-off-by: Måns Ansgariusson <Mansgariusson@gmail.com>
This commit is contained in:
Måns Ansgariusson 2024-12-20 15:46:36 +01:00 • committed by Benjamin Cabé
commit c8a2a080ac
7 changed files with 414 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

159
include/zephyr/kernel.h
View file

@ -22,6 +22,7 @@
#include <zephyr/tracing/tracing_macros.h>
#include <zephyr/sys/mem_stats.h>
#include <zephyr/sys/iterable_sections.h>
#include <zephyr/sys/ring_buffer.h>
#ifdef __cplusplus
extern "C" {
@ -4991,6 +4992,18 @@ void k_mbox_data_get(struct k_mbox_msg *rx_msg, void *buffer);
* @{
*/
/**
* @brief initialize a pipe
*
* This routine initializes a pipe object, prior to its first use.
*
* @param pipe Address of the pipe.
* @param buffer Address of the pipe's buffer.
* @param buffer_size Size of the pipe's buffer.
*/
__syscall void k_pipe_init(struct k_pipe *pipe, uint8_t *buffer, size_t buffer_size);
#ifdef CONFIG_PIPES
/** Pipe Structure */
struct k_pipe {
unsigned char *buffer; /**< Pipe buffer: may be NULL */
@ -5061,19 +5074,7 @@ struct k_pipe {
Z_PIPE_INITIALIZER(name, _k_pipe_buf_##name, pipe_buffer_size)
/**
* @brief Initialize a pipe.
*
* This routine initializes a pipe object, prior to its first use.
*
* @param pipe Address of the pipe.
* @param buffer Address of the pipe's ring buffer, or NULL if no ring buffer
* is used.
* @param size Size of the pipe's ring buffer (in bytes), or zero if no ring
* buffer is used.
*/
void k_pipe_init(struct k_pipe *pipe, unsigned char *buffer, size_t size);
/**
* @deprecated Dynamic allocation of pipe buffers will be removed in the new k_pipe API.
* @brief Release a pipe's allocated buffer
*
* If a pipe object was given a dynamically allocated buffer via
@ -5084,9 +5085,10 @@ void k_pipe_init(struct k_pipe *pipe, unsigned char *buffer, size_t size);
* @retval 0 on success
* @retval -EAGAIN nothing to cleanup
*/
int k_pipe_cleanup(struct k_pipe *pipe);
__deprecated int k_pipe_cleanup(struct k_pipe *pipe);
/**
* @deprecated Dynamic allocation of pipe buffers will be removed in the new k_pipe API.
* @brief Initialize a pipe and allocate a buffer for it
*
* Storage for the buffer region will be allocated from the calling thread's
@ -5101,9 +5103,10 @@ int k_pipe_cleanup(struct k_pipe *pipe);
* @retval 0 on success
* @retval -ENOMEM if memory couldn't be allocated
*/
__syscall int k_pipe_alloc_init(struct k_pipe *pipe, size_t size);
__deprecated __syscall int k_pipe_alloc_init(struct k_pipe *pipe, size_t size);
/**
* @deprecated k_pipe_put() is replaced by k_pipe_write(...) in the new k_pipe API.
* @brief Write data to a pipe.
*
* This routine writes up to @a bytes_to_write bytes of data to @a pipe.
@ -5121,11 +5124,12 @@ __syscall int k_pipe_alloc_init(struct k_pipe *pipe, size_t size);
* @retval -EAGAIN Waiting period timed out; between zero and @a min_xfer
* minus one data bytes were written.
*/
__syscall int k_pipe_put(struct k_pipe *pipe, const void *data,
__deprecated __syscall int k_pipe_put(struct k_pipe *pipe, const void *data,
size_t bytes_to_write, size_t *bytes_written,
size_t min_xfer, k_timeout_t timeout);
/**
* @deprecated k_pipe_get() is replaced by k_pipe_read(...) in the new k_pipe API.
* @brief Read data from a pipe.
*
* This routine reads up to @a bytes_to_read bytes of data from @a pipe.
@ -5144,11 +5148,12 @@ __syscall int k_pipe_put(struct k_pipe *pipe, const void *data,
* @retval -EAGAIN Waiting period timed out; between zero and @a min_xfer
* minus one data bytes were read.
*/
__syscall int k_pipe_get(struct k_pipe *pipe, void *data,
__deprecated __syscall int k_pipe_get(struct k_pipe *pipe, void *data,
size_t bytes_to_read, size_t *bytes_read,
size_t min_xfer, k_timeout_t timeout);
/**
* @deprecated k_pipe_read_avail() will be removed in the new k_pipe API.
* @brief Query the number of bytes that may be read from @a pipe.
*
* @param pipe Address of the pipe.
@ -5156,9 +5161,10 @@ __syscall int k_pipe_get(struct k_pipe *pipe, void *data,
* @retval a number n such that 0 <= n <= @ref k_pipe.size; the
* result is zero for unbuffered pipes.
*/
__syscall size_t k_pipe_read_avail(struct k_pipe *pipe);
__deprecated __syscall size_t k_pipe_read_avail(struct k_pipe *pipe);
/**
* @deprecated k_pipe_write_avail() will be removed in the new k_pipe API.
* @brief Query the number of bytes that may be written to @a pipe
*
* @param pipe Address of the pipe.
@ -5166,9 +5172,10 @@ __syscall size_t k_pipe_read_avail(struct k_pipe *pipe);
* @retval a number n such that 0 <= n <= @ref k_pipe.size; the
* result is zero for unbuffered pipes.
*/
__syscall size_t k_pipe_write_avail(struct k_pipe *pipe);
__deprecated __syscall size_t k_pipe_write_avail(struct k_pipe *pipe);
/**
* @deprecated k_pipe_flush() will be removed in the new k_pipe API.
* @brief Flush the pipe of write data
*
* This routine flushes the pipe. Flushing the pipe is equivalent to reading
@ -5178,9 +5185,10 @@ __syscall size_t k_pipe_write_avail(struct k_pipe *pipe);
*
* @param pipe Address of the pipe.
*/
__syscall void k_pipe_flush(struct k_pipe *pipe);
__deprecated __syscall void k_pipe_flush(struct k_pipe *pipe);
/**
* @deprecated k_pipe_buffer_flush will be removed in the new k_pipe API.
* @brief Flush the pipe's internal buffer
*
* This routine flushes the pipe's internal buffer. This is equivalent to
@ -5191,14 +5199,121 @@ __syscall void k_pipe_flush(struct k_pipe *pipe);
*
* @param pipe Address of the pipe.
*/
__syscall void k_pipe_buffer_flush(struct k_pipe *pipe);
__deprecated __syscall void k_pipe_buffer_flush(struct k_pipe *pipe);
#else /* CONFIG_PIPES */
enum pipe_flags {
PIPE_FLAG_OPEN = BIT(0),
PIPE_FLAG_RESET = BIT(1),
};
struct k_pipe {
size_t waiting;
struct ring_buf buf;
struct k_spinlock lock;
_wait_q_t data;
_wait_q_t space;
uint8_t flags;
};
/**
* @cond INTERNAL_HIDDEN
*/
#define Z_PIPE_INITIALIZER(obj, pipe_buffer, pipe_buffer_size) \
{ \
.buf = RING_BUF_INIT(pipe_buffer, pipe_buffer_size), \
.data = Z_WAIT_Q_INIT(&obj.data), \
.space = Z_WAIT_Q_INIT(&obj.space), \
.flags = PIPE_FLAG_OPEN, \
.waiting = 0, \
}
/**
* INTERNAL_HIDDEN @endcond
*/
/**
* @brief Statically define and initialize a pipe.
*
* The pipe can be accessed outside the module where it is defined using:
*
* @code extern struct k_pipe <name>; @endcode
*
* @param name Name of the pipe.
* @param pipe_buffer_size Size of the pipe's ring buffer (in bytes).
* @param pipe_align Alignment of the pipe's ring buffer (power of 2).
*
*/
#define K_PIPE_DEFINE(name, pipe_buffer_size, pipe_align) \
static unsigned char __noinit __aligned(pipe_align) \
_k_pipe_buf_##name[pipe_buffer_size]; \
STRUCT_SECTION_ITERABLE(k_pipe, name) = \
Z_PIPE_INITIALIZER(name, _k_pipe_buf_##name, pipe_buffer_size)
/**
* @brief Write data to a pipe
*
* This routine writes up to @a len bytes of data to @a pipe.
* If the pipe is full, the routine will block until the data can be written or the timeout expires.
*
* @param pipe Address of the pipe.
* @param data Address of data to write.
* @param len Size of data (in bytes).
* @param timeout Waiting period to wait for the data to be written.
*
* @retval number of bytes written on success
* @retval -EAGAIN if no data could be written before the timeout expired
* @retval -ECANCELED if the write was interrupted by k_pipe_reset(..)
* @retval -EPIPE if the pipe was closed
*/
__syscall int k_pipe_write(struct k_pipe *pipe, const uint8_t *data, size_t len,
k_timeout_t timeout);
/**
* @brief Read data from a pipe
* This routine reads up to @a len bytes of data from @a pipe.
* If the pipe is empty, the routine will block until the data can be read or the timeout expires.
*
* @param pipe Address of the pipe.
* @param data Address to place the data read from pipe.
* @param len Requested number of bytes to read.
* @param timeout Waiting period to wait for the data to be read.
*
* @retval number of bytes read on success
* @retval -EAGAIN if no data could be read before the timeout expired
* @retval -ECANCELED if the read was interrupted by k_pipe_reset(..)
* @retval -EPIPE if the pipe was closed
*/
__syscall int k_pipe_read(struct k_pipe *pipe, uint8_t *data, size_t len,
k_timeout_t timeout);
/**
* @brief Reset a pipe
* This routine resets the pipe, discarding any unread data and unblocking any threads waiting to
* write or read, causing the waiting threads to return with -ECANCELED. Calling k_pipe_read(..) or
* k_pipe_write(..) when the pipe is resetting but not yet reset will return -ECANCELED.
* The pipe is left open after a reset and can be used as normal.
*
* @param pipe Address of the pipe.
*/
__syscall void k_pipe_reset(struct k_pipe *pipe);
/**
* @brief Close a pipe
*
* This routine closes a pipe. Any threads that were blocked on the pipe
* will be unblocked and receive an error code.
*
* @param pipe Address of the pipe.
*/
__syscall void k_pipe_close(struct k_pipe *pipe);
#endif /* CONFIG_PIPES */
/** @} */
/**
* @cond INTERNAL_HIDDEN
*/
struct k_mem_slab_info {
uint32_t num_blocks;
size_t block_size;

11
include/zephyr/sys/ring_buffer.h
View file

@ -7,7 +7,6 @@
#ifndef ZEPHYR_INCLUDE_SYS_RING_BUFFER_H_
#define ZEPHYR_INCLUDE_SYS_RING_BUFFER_H_
#include <zephyr/kernel.h>
#include <zephyr/sys/util.h>
#include <errno.h>
@ -62,6 +61,11 @@ static inline void ring_buf_internal_reset(struct ring_buf *buf, int32_t value)
buf->get_head = buf->get_tail = buf->get_base = value;
}
#define RING_BUF_INIT(buf, size8) \
{ \
.buffer = buf, \
.size = size8, \
}
/**
* @brief Define and initialize a ring buffer for byte data.
*
@ -80,10 +84,7 @@ static inline void ring_buf_internal_reset(struct ring_buf *buf, int32_t value)
BUILD_ASSERT(size8 < RING_BUFFER_MAX_SIZE,\
RING_BUFFER_SIZE_ASSERT_MSG); \
static uint8_t __noinit _ring_buffer_data_##name[size8]; \
struct ring_buf name = { \
.buffer = _ring_buffer_data_##name, \
.size = size8 \
}
struct ring_buf name = RING_BUF_INIT(_ring_buffer_data_##name, size8)
/**
* @brief Define and initialize an "item based" ring buffer.

4
kernel/CMakeLists.txt
View file

@ -84,6 +84,10 @@ list(APPEND kernel_files
thread.c
sched.c
)
# FIXME: Once the prior pipe implementation is removed, this should be included in the above list
if(NOT CONFIG_PIPES)
list(APPEND kernel_files pipe.c)
endif() # NOT CONFIG_PIPES
if(CONFIG_SMP)
list(APPEND kernel_files
smp.c

5
kernel/Kconfig
View file

@ -12,6 +12,7 @@ source "subsys/logging/Kconfig.template.log_config"
config MULTITHREADING
bool "Multi-threading" if ARCH_HAS_SINGLE_THREAD_SUPPORT
default y
select RING_BUFFER
help
If disabled, only the main thread is available, so a main() function
must be provided. Interrupts are available. Kernel objects will most
@ -713,6 +714,7 @@ config EVENTS
config PIPES
bool "Pipe objects"
select DEPRECATED
help
This option enables kernel pipes. A pipe is a kernel object that
allows a thread to send a byte stream to another thread. Pipes can
@ -720,6 +722,9 @@ config PIPES
Note that setting this option slightly increases the size of the
thread structure.
This Kconfig is deprecated and will be removed, by disabling this
kconfig another implementation of k_pipe will be available when
CONFIG_MULTITHREADING is enabled.
config KERNEL_MEM_POOL
bool "Use Kernel Memory Pool"

2
kernel/Kconfig.obj_core
View file

@ -77,7 +77,7 @@ config OBJ_CORE_SEM
config OBJ_CORE_PIPE
bool "Integrate pipe into object core framework"
default y if PIPES
default y
help
When enabled, this option integrates pipes into the object core
framework.

251
kernel/pipe.c Normal file
View file

@ -0,0 +1,251 @@
/*
* Copyright (c) 2024 Måns Ansgariusson <mansgariusson@gmail.com>
*
* SPDX-License-Identifier: Apache-2.0
*/
#include <zephyr/init.h>
#include <zephyr/kernel.h>
#include <zephyr/internal/syscall_handler.h>
#include <ksched.h>
#include <kthread.h>
#include <wait_q.h>
static inline bool pipe_closed(struct k_pipe *pipe)
{
return (pipe->flags & PIPE_FLAG_OPEN) == 0;
}
static inline bool pipe_resetting(struct k_pipe *pipe)
{
return (pipe->flags & PIPE_FLAG_RESET) != 0;
}
static inline bool pipe_full(struct k_pipe *pipe)
{
return ring_buf_space_get(&pipe->buf) == 0;
}
static inline bool pipe_empty(struct k_pipe *pipe)
{
return ring_buf_is_empty(&pipe->buf);
}
static inline int wait_for(_wait_q_t *waitq, struct k_pipe *pipe, k_spinlock_key_t *key,
k_timepoint_t time_limit)
{
k_timeout_t timeout = sys_timepoint_timeout(time_limit);
if (K_TIMEOUT_EQ(timeout, K_NO_WAIT)) {
return -EAGAIN;
}
pipe->waiting++;
z_pend_curr(&pipe->lock, *key, waitq, timeout);
*key = k_spin_lock(&pipe->lock);
pipe->waiting--;
if (unlikely(pipe_closed(pipe))) {
return -EPIPE;
} else if (unlikely(pipe_resetting(pipe))) {
if (pipe->waiting == 0) {
pipe->flags &= ~PIPE_FLAG_RESET;
}
return -ECANCELED;
} else if (sys_timepoint_expired(time_limit)) {
return -EAGAIN;
}
return 0;
}
static void notify_waiter(_wait_q_t *waitq)
{
struct k_thread *thread_to_unblock = z_unpend_first_thread(waitq);
if (likely(thread_to_unblock != NULL)) {
z_ready_thread(thread_to_unblock);
}
}
void z_impl_k_pipe_init(struct k_pipe *pipe, uint8_t *buffer, size_t buffer_size)
{
ring_buf_init(&pipe->buf, buffer_size, buffer);
pipe->flags = PIPE_FLAG_OPEN;
pipe->waiting = 0;
pipe->lock = (struct k_spinlock){};
z_waitq_init(&pipe->data);
z_waitq_init(&pipe->space);
k_object_init(pipe);
}
int z_impl_k_pipe_write(struct k_pipe *pipe, const uint8_t *data, size_t len, k_timeout_t timeout)
{
int rc;
size_t written = 0;
k_spinlock_key_t key;
k_timepoint_t end = sys_timepoint_calc(timeout);
while (written < len) {
key = k_spin_lock(&pipe->lock);
if (unlikely(pipe_closed(pipe))) {
k_spin_unlock(&pipe->lock, key);
rc = -EPIPE;
goto exit;
} else if (unlikely(pipe_resetting(pipe))) {
k_spin_unlock(&pipe->lock, key);
rc = -ECANCELED;
goto exit;
} else if (pipe_full(pipe)) {
rc = wait_for(&pipe->space, pipe, &key, end);
if (rc == -EAGAIN) {
/* the timeout expired */
k_spin_unlock(&pipe->lock, key);
rc = written ? written : -EAGAIN;
goto exit;
} else if (unlikely(rc != 0)) {
/* The pipe was closed or reseted while waiting for space */
k_spin_unlock(&pipe->lock, key);
goto exit;
} else if (unlikely(pipe_full(pipe))) {
/* Timeout has not elapsed, the pipe is open and not resetting,
* we've been notified of available space, but the notified space
* was consumed by another thread before the calling thread.
*/
k_spin_unlock(&pipe->lock, key);
continue;
}
/* The timeout has not elapsed, we've been notified of
* available space, and the pipe is not full. Continue writing.
*/
}
rc = ring_buf_put(&pipe->buf, &data[written], len - written);
if (likely(rc != 0)) {
notify_waiter(&pipe->data);
}
k_spin_unlock(&pipe->lock, key);
written += rc;
}
rc = written;
exit:
return rc;
}
int z_impl_k_pipe_read(struct k_pipe *pipe, uint8_t *data, size_t len, k_timeout_t timeout)
{
int rc;
size_t read = 0;
k_spinlock_key_t key;
k_timepoint_t end = sys_timepoint_calc(timeout);
while (read < len) {
key = k_spin_lock(&pipe->lock);
if (unlikely(pipe_resetting(pipe))) {
k_spin_unlock(&pipe->lock, key);
rc = -ECANCELED;
goto exit;
} else if (pipe_empty(pipe) && !pipe_closed(pipe)) {
rc = wait_for(&pipe->data, pipe, &key, end);
if (rc == -EAGAIN) {
/* The timeout elapsed */
k_spin_unlock(&pipe->lock, key);
rc = read ? read : -EAGAIN;
goto exit;
} else if (unlikely(rc == -ECANCELED)) {
/* The pipe is being rested. */
k_spin_unlock(&pipe->lock, key);
goto exit;
} else if (unlikely(rc == 0 && pipe_empty(pipe))) {
/* Timeout has not elapsed, we've been notified of available bytes
* but they have been consumed by another thread before the calling
* thread.
*/
k_spin_unlock(&pipe->lock, key);
continue;
}
/* The timeout has not elapsed, we've been notified of
* available bytes, and the pipe is not empty. Continue reading.
*/
}
if (unlikely(pipe_closed(pipe) && pipe_empty(pipe))) {
k_spin_unlock(&pipe->lock, key);
rc = read ? read : -EPIPE;
goto exit;
}
rc = ring_buf_get(&pipe->buf, &data[read], len - read);
if (likely(rc != 0)) {
notify_waiter(&pipe->space);
}
read += rc;
k_spin_unlock(&pipe->lock, key);
}
rc = read;
exit:
return rc;
}
void z_impl_k_pipe_reset(struct k_pipe *pipe)
{
K_SPINLOCK(&pipe->lock) {
ring_buf_reset(&pipe->buf);
if (likely(pipe->waiting != 0)) {
pipe->flags |= PIPE_FLAG_RESET;
z_unpend_all(&pipe->data);
z_unpend_all(&pipe->space);
}
}
}
void z_impl_k_pipe_close(struct k_pipe *pipe)
{
K_SPINLOCK(&pipe->lock) {
pipe->flags = 0;
z_unpend_all(&pipe->data);
z_unpend_all(&pipe->space);
}
}
#ifdef CONFIG_USERSPACE
void z_vrfy_k_pipe_init(struct k_pipe *pipe, uint8_t *buffer, size_t buffer_size)
{
K_OOPS(K_SYSCALL_OBJ(pipe, K_OBJ_PIPE));
K_OOPS(K_SYSCALL_MEMORY_WRITE(buffer, buffer_size));
z_impl_k_pipe_init(pipe, buffer, buffer_size);
}
#include <zephyr/syscalls/k_pipe_init_mrsh.c>
int z_vrfy_k_pipe_read(struct k_pipe *pipe, uint8_t *data, size_t len, k_timeout_t timeout)
{
K_OOPS(K_SYSCALL_OBJ(pipe, K_OBJ_PIPE));
K_OOPS(K_SYSCALL_MEMORY_WRITE(data, len));
return z_impl_k_pipe_read(pipe, data, len, timeout);
}
#include <zephyr/syscalls/k_pipe_read_mrsh.c>
int z_vrfy_k_pipe_write(struct k_pipe *pipe, const uint8_t *data, size_t len, k_timeout_t timeout)
{
K_OOPS(K_SYSCALL_OBJ(pipe, K_OBJ_PIPE));
K_OOPS(K_SYSCALL_MEMORY_READ(data, len));
return z_impl_k_pipe_write(pipe, data, len, timeout);
}
#include <zephyr/syscalls/k_pipe_write_mrsh.c>
void z_vrfy_k_pipe_reset(struct k_pipe *pipe)
{
K_OOPS(K_SYSCALL_OBJ(pipe, K_OBJ_PIPE));
z_impl_k_pipe_reset(pipe);
}
#include <zephyr/syscalls/k_pipe_reset_mrsh.c>
void z_vrfy_k_pipe_close(struct k_pipe *pipe)
{
K_OOPS(K_SYSCALL_OBJ(pipe, K_OBJ_PIPE));
z_impl_k_pipe_close(pipe);
}
#include <zephyr/syscalls/k_pipe_close_mrsh.c>
#endif /* CONFIG_USERSPACE */

11
kernel/pipes.c
View file

@ -36,7 +36,7 @@ static struct k_obj_type obj_type_pipe;
#endif /* CONFIG_OBJ_CORE_PIPE */
void k_pipe_init(struct k_pipe *pipe, unsigned char *buffer, size_t size)
void z_impl_k_pipe_init(struct k_pipe *pipe, unsigned char *buffer, size_t size)
{
pipe->buffer = buffer;
pipe->size = size;
@ -87,6 +87,15 @@ int z_impl_k_pipe_alloc_init(struct k_pipe *pipe, size_t size)
}
#ifdef CONFIG_USERSPACE
static inline void z_vrfy_k_pipe_init(struct k_pipe *pipe, unsigned char *buffer, size_t size)
{
K_OOPS(K_SYSCALL_OBJ_NEVER_INIT(pipe, K_OBJ_PIPE));
K_OOPS(K_SYSCALL_MEMORY_WRITE(buffer, size));
z_impl_k_pipe_init(pipe, buffer, size);
}
#include <zephyr/syscalls/k_pipe_init_mrsh.c>
static inline int z_vrfy_k_pipe_alloc_init(struct k_pipe *pipe, size_t size)
{
K_OOPS(K_SYSCALL_OBJ_NEVER_INIT(pipe, K_OBJ_PIPE));