1 /*
2 * Copyright (c) 2024 Måns Ansgariusson <mansgariusson@gmail.com>
3 *
4 * SPDX-License-Identifier: Apache-2.0
5 */
6 #include <stdint.h>
7 #include <zephyr/kernel.h>
8 #include <zephyr/ztest.h>
9 #include <zephyr/random/random.h>
10 #include <zephyr/logging/log.h>
11
12 LOG_MODULE_REGISTER(k_pipe_concurrency, LOG_LEVEL_DBG);
13 ZTEST_SUITE(k_pipe_concurrency, NULL, NULL, NULL, NULL, NULL);
14
15 static const int partial_wait_time = 2000;
16 #define DUMMY_DATA_SIZE 16
17 static struct k_thread thread;
18 static K_THREAD_STACK_DEFINE(stack, 1024);
19 static struct k_pipe pipe;
20
thread_close(void * arg1,void * arg2,void * arg3)21 static void thread_close(void *arg1, void *arg2, void *arg3)
22 {
23 k_pipe_close((struct k_pipe *)arg1);
24 }
25
thread_reset(void * arg1,void * arg2,void * arg3)26 static void thread_reset(void *arg1, void *arg2, void *arg3)
27 {
28 k_pipe_reset((struct k_pipe *)arg1);
29 }
30
thread_write(void * arg1,void * arg2,void * arg3)31 static void thread_write(void *arg1, void *arg2, void *arg3)
32 {
33 uint8_t garbage[DUMMY_DATA_SIZE] = {};
34
35 zassert_true(k_pipe_write((struct k_pipe *)arg1, garbage, sizeof(garbage),
36 K_MSEC(partial_wait_time)) == sizeof(garbage), "Failed to write to pipe");
37 }
38
thread_read(void * arg1,void * arg2,void * arg3)39 static void thread_read(void *arg1, void *arg2, void *arg3)
40 {
41 uint8_t garbage[DUMMY_DATA_SIZE];
42
43 zassert_true(k_pipe_read((struct k_pipe *)arg1, garbage, sizeof(garbage),
44 K_MSEC(partial_wait_time)) == sizeof(garbage), "Failed to read from pipe");
45 }
46
ZTEST(k_pipe_concurrency,test_close_on_read)47 ZTEST(k_pipe_concurrency, test_close_on_read)
48 {
49 k_tid_t tid;
50 uint8_t buffer[DUMMY_DATA_SIZE];
51 uint8_t res;
52
53 k_pipe_init(&pipe, buffer, sizeof(buffer));
54 tid = k_thread_create(&thread, stack, K_THREAD_STACK_SIZEOF(stack),
55 thread_close, &pipe, NULL, NULL, K_PRIO_COOP(0), 0, K_MSEC(100));
56 zassert_true(tid, "k_thread_create failed");
57 zassert_true(k_pipe_read(&pipe, &res, sizeof(res), K_MSEC(1000)) == -EPIPE,
58 "Read on closed pipe should return -EPIPE");
59 k_thread_join(tid, K_FOREVER);
60 zassert_true((pipe.flags & PIPE_FLAG_OPEN) == 0,
61 "Pipe should continue to be closed after all waiters have been released");
62 }
63
ZTEST(k_pipe_concurrency,test_close_on_write)64 ZTEST(k_pipe_concurrency, test_close_on_write)
65 {
66 k_tid_t tid;
67 uint8_t buffer[DUMMY_DATA_SIZE];
68 uint8_t garbage[DUMMY_DATA_SIZE];
69
70 k_pipe_init(&pipe, buffer, sizeof(buffer));
71 zassert_true(sizeof(garbage) == k_pipe_write(&pipe, garbage, sizeof(garbage), K_MSEC(1000)),
72 "Failed to write to pipe");
73
74 tid = k_thread_create(&thread, stack, K_THREAD_STACK_SIZEOF(stack),
75 thread_close, &pipe, NULL, NULL, K_PRIO_COOP(0), 0, K_MSEC(100));
76 zassert_true(tid, "k_thread_create failed");
77 zassert_true(k_pipe_write(&pipe, garbage, sizeof(garbage), K_MSEC(1000)) == -EPIPE,
78 "write should return -EPIPE, when pipe is closed");
79 k_thread_join(tid, K_FOREVER);
80 zassert_true((pipe.flags & PIPE_FLAG_OPEN) == 0,
81 "pipe should continue to be closed after all waiters have been released");
82 }
83
ZTEST(k_pipe_concurrency,test_reset_on_read)84 ZTEST(k_pipe_concurrency, test_reset_on_read)
85 {
86 k_tid_t tid;
87 uint8_t buffer[DUMMY_DATA_SIZE];
88 uint8_t res;
89
90 k_pipe_init(&pipe, buffer, sizeof(buffer));
91
92 tid = k_thread_create(&thread, stack, K_THREAD_STACK_SIZEOF(stack),
93 thread_reset, &pipe, NULL, NULL, K_PRIO_COOP(0), 0, K_MSEC(100));
94 zassert_true(tid, "k_thread_create failed");
95 zassert_true(k_pipe_read(&pipe, &res, sizeof(res), K_MSEC(1000)) == -ECANCELED,
96 "reset on read should return -ECANCELED");
97 k_thread_join(tid, K_FOREVER);
98 zassert_true((pipe.flags & PIPE_FLAG_RESET) == 0,
99 "pipe should not have reset flag after all waiters are done");
100 zassert_true((pipe.flags & PIPE_FLAG_OPEN) != 0,
101 "pipe should continue to be open after pipe is reseted");
102 }
103
ZTEST(k_pipe_concurrency,test_reset_on_write)104 ZTEST(k_pipe_concurrency, test_reset_on_write)
105 {
106 k_tid_t tid;
107 uint8_t buffer[DUMMY_DATA_SIZE];
108 uint8_t garbage[DUMMY_DATA_SIZE];
109
110 k_pipe_init(&pipe, buffer, sizeof(buffer));
111 zassert_true(sizeof(garbage) == k_pipe_write(&pipe, garbage, sizeof(garbage), K_MSEC(1000)),
112 "Failed to write to pipe");
113
114 tid = k_thread_create(&thread, stack, K_THREAD_STACK_SIZEOF(stack),
115 thread_reset, &pipe, NULL, NULL, K_PRIO_COOP(0), 0, K_MSEC(100));
116 zassert_true(tid, "k_thread_create failed");
117 zassert_true(k_pipe_write(&pipe, garbage, sizeof(garbage), K_MSEC(1000)) == -ECANCELED,
118 "reset on write should return -ECANCELED");
119 k_thread_join(tid, K_FOREVER);
120 zassert_true((pipe.flags & PIPE_FLAG_RESET) == 0,
121 "pipe should not have reset flag after all waiters are done");
122 zassert_true((pipe.flags & PIPE_FLAG_OPEN) != 0,
123 "pipe should continue to be open after pipe is reseted");
124 }
125
ZTEST(k_pipe_concurrency,test_partial_read)126 ZTEST(k_pipe_concurrency, test_partial_read)
127 {
128 k_tid_t tid;
129 uint8_t buffer[DUMMY_DATA_SIZE];
130 uint8_t garbage[DUMMY_DATA_SIZE];
131 size_t write_size = sizeof(garbage)/2;
132
133 k_pipe_init(&pipe, buffer, sizeof(buffer));
134 tid = k_thread_create(&thread, stack, K_THREAD_STACK_SIZEOF(stack),
135 thread_read, &pipe, NULL, NULL, K_PRIO_COOP(0), 0, K_NO_WAIT);
136
137 zassert_true(k_pipe_write(&pipe, garbage, write_size, K_NO_WAIT) == write_size,
138 "write to pipe failed");
139 k_msleep(partial_wait_time/4);
140 zassert_true(k_pipe_write(&pipe, garbage, write_size, K_NO_WAIT) == write_size,
141 "k_k_pipe_write should return number of bytes written");
142 k_thread_join(tid, K_FOREVER);
143 }
144
ZTEST(k_pipe_concurrency,test_partial_write)145 ZTEST(k_pipe_concurrency, test_partial_write)
146 {
147 k_tid_t tid;
148 uint8_t buffer[DUMMY_DATA_SIZE];
149 uint8_t garbage[DUMMY_DATA_SIZE];
150 size_t read_size = sizeof(garbage)/2;
151
152 k_pipe_init(&pipe, buffer, sizeof(buffer));
153
154 zassert_true(k_pipe_write(&pipe, garbage, sizeof(garbage), K_NO_WAIT) == sizeof(garbage),
155 "Failed to write to pipe");
156 tid = k_thread_create(&thread, stack, K_THREAD_STACK_SIZEOF(stack),
157 thread_write, &pipe, NULL, NULL, K_PRIO_COOP(0), 0, K_NO_WAIT);
158
159 zassert_true(k_pipe_read(&pipe, garbage, read_size, K_NO_WAIT) == read_size,
160 "Failed to read from pipe");
161 k_msleep(partial_wait_time/2);
162 zassert_true(k_pipe_read(&pipe, garbage, read_size, K_NO_WAIT) == read_size,
163 "failed t read from pipe");
164 k_thread_join(tid, K_FOREVER);
165 }
166
