Branch data Line data Source code
1 : : /* SPDX-FileCopyrightText: Copyright 2026 Cozens Software Solutions Limited
2 : : * SPDX-License-Identifier: PolyForm-Noncommercial-1.0.0 OR LicenseRef-PolyForm-Internal-Use-1.0.0 OR LicenseRef-COSOSO-Commercial
3 : : */
4 : :
5 : : #include "SolidSyslogPosixMessageQueueBuffer.h"
6 : :
7 : : #include <errno.h>
8 : : #include <fcntl.h>
9 : : #include <mqueue.h>
10 : : #include <stdbool.h>
11 : : #include <stddef.h>
12 : : #include <stdint.h>
13 : : #include <sys/types.h>
14 : : #include <unistd.h>
15 : :
16 : : #include "SolidSyslogBufferCategories.h"
17 : : #include "SolidSyslogBufferDefinition.h"
18 : : #include "SolidSyslogError.h"
19 : : #include "SolidSyslogErrorCategory.h"
20 : : #include "SolidSyslogFormatter.h"
21 : : #include "SolidSyslogMutex.h"
22 : : #include "SolidSyslogNullBuffer.h"
23 : : #include "SolidSyslogPosixMessageQueueBufferErrors.h"
24 : : #include "SolidSyslogPosixMessageQueueBufferPrivate.h"
25 : : #include "SolidSyslogPrival.h"
26 : : #include "SolidSyslogTunables.h"
27 : :
28 : : const struct SolidSyslogErrorSource SolidSyslogPosixMessageQueueBufferErrorSource = {
29 : : "PosixMessageQueueBuffer",
30 : : SolidSyslogPosixMessageQueueBufferErrors_Name
31 : : };
32 : :
33 : : enum
34 : : {
35 : : /* 0600 in octal - owner read/write, equivalent to S_IRUSR | S_IWUSR. Hex form avoids MISRA 7.1. */
36 : : OWNER_READ_WRITE = 0x180U
37 : : };
38 : :
39 : : /* Which run of sends the queue would not take, if any, a write has just
40 : : * started. A run of evictions is noted apart, since a write can evict and
41 : : * still be refused. */
42 : : enum SolidSyslogPosixMessageQueueBufferRunStarted
43 : : {
44 : : RUN_STARTED_NONE,
45 : : RUN_STARTED_REFUSALS,
46 : : RUN_STARTED_FAILURES
47 : : };
48 : :
49 : : static bool PosixMessageQueueBuffer_Read(struct SolidSyslogBuffer* base, void* data, size_t maxSize, size_t* bytesRead);
50 : : static void PosixMessageQueueBuffer_Write(struct SolidSyslogBuffer* base, const void* data, size_t size);
51 : : static bool PosixMessageQueueBuffer_IsFull(struct SolidSyslogBuffer* base);
52 : : static inline void PosixMessageQueueBuffer_NoteRecordTaken(struct SolidSyslogPosixMessageQueueBuffer* self);
53 : : static inline void PosixMessageQueueBuffer_NoteReceiveFailed(struct SolidSyslogPosixMessageQueueBuffer* self);
54 : : static enum SolidSyslogPosixMessageQueueBufferRunStarted PosixMessageQueueBuffer_SendNotingRuns(
55 : : struct SolidSyslogPosixMessageQueueBuffer* self,
56 : : enum SolidSyslogDiscardPolicy policy,
57 : : const void* data,
58 : : size_t size,
59 : : bool* evictionRunStarted
60 : : );
61 : : static inline bool PosixMessageQueueBuffer_Send(
62 : : struct SolidSyslogPosixMessageQueueBuffer* self,
63 : : const void* data,
64 : : size_t size,
65 : : int* sendErrno
66 : : );
67 : : static inline bool PosixMessageQueueBuffer_EvictOne(const struct SolidSyslogPosixMessageQueueBuffer* self);
68 : : static inline bool PosixMessageQueueBuffer_NoteEvicted(struct SolidSyslogPosixMessageQueueBuffer* self, bool evicted);
69 : : static inline void PosixMessageQueueBuffer_ReportRecordsEvicted(void);
70 : : static void PosixMessageQueueBuffer_ReportRunStarted(enum SolidSyslogPosixMessageQueueBufferRunStarted started);
71 : :
72 : : static inline struct SolidSyslogPosixMessageQueueBuffer* PosixMessageQueueBuffer_SelfFromBase(
73 : : struct SolidSyslogBuffer* base
74 : : );
75 : : static inline void PosixMessageQueueBuffer_InstallDiscardPolicy(
76 : : struct SolidSyslogPosixMessageQueueBuffer* self,
77 : : SolidSyslogDiscardPolicyFunction configured,
78 : : void* context
79 : : );
80 : :
81 : : // NOLINTBEGIN(bugprone-easily-swappable-parameters) -- distinct semantic meaning; mirrors the public SolidSyslogPosixMessageQueueBuffer_Create signature plus a per-slot discriminator
82 : 66 : int SolidSyslogPosixMessageQueueBuffer_Initialise(
83 : : struct SolidSyslogBuffer* base,
84 : : struct SolidSyslogMutex* mutex,
85 : : size_t maxMessages,
86 : : size_t slotIndex,
87 : : SolidSyslogDiscardPolicyFunction getDiscardPolicy,
88 : : void* discardPolicyContext
89 : : )
90 : : // NOLINTEND(bugprone-easily-swappable-parameters)
91 : : {
92 : : static const char queueNamePrefix[] = "/solidsyslog_";
93 : :
94 : 66 : struct SolidSyslogPosixMessageQueueBuffer* self = PosixMessageQueueBuffer_SelfFromBase(base);
95 : :
96 : : /* Queue name: /solidsyslog_<pid>_<slotIndex>. The pid keeps the name
97 : : * unique per process; the slot index keeps multiple in-process pool
98 : : * entries from aliasing onto the same kernel queue object. */
99 : : SolidSyslogFormatterStorage
100 : : nameStorage[SOLIDSYSLOG_FORMATTER_STORAGE_SIZE(POSIX_MESSAGE_QUEUE_BUFFER_MAX_NAME_SIZE)];
101 : : struct SolidSyslogFormatter* name =
102 : 66 : SolidSyslogFormatter_Create(nameStorage, POSIX_MESSAGE_QUEUE_BUFFER_MAX_NAME_SIZE);
103 : 66 : SolidSyslogFormatter_BoundedString(name, queueNamePrefix, sizeof(queueNamePrefix) - 1U);
104 : 66 : SolidSyslogFormatter_Uint32(name, (uint32_t) getpid());
105 : 66 : SolidSyslogFormatter_AsciiCharacter(name, '_');
106 : 66 : SolidSyslogFormatter_Uint32(name, (uint32_t) slotIndex);
107 : 66 : const char* queueName = SolidSyslogFormatter_AsFormattedBuffer(name);
108 : :
109 : 66 : struct mq_attr attr = {0};
110 : 66 : attr.mq_maxmsg = (long) maxMessages;
111 : 66 : attr.mq_msgsize = (long) SOLIDSYSLOG_MAX_MESSAGE_SIZE;
112 : :
113 : 66 : self->Mq = mq_open(queueName, O_CREAT | O_EXCL | O_RDWR | O_NONBLOCK, OWNER_READ_WRITE, &attr);
114 : : /* errno is captured straight after mq_open, as Read does after
115 : : * mq_receive (MISRA C:2012 Rule 22.10). */
116 [ + + ]: 66 : int openErrno = (self->Mq == (mqd_t) -1) ? errno : 0;
117 : 66 : self->Mutex = mutex;
118 : 66 : PosixMessageQueueBuffer_InstallDiscardPolicy(self, getDiscardPolicy, discardPolicyContext);
119 : 66 : self->RefusalReported = false;
120 : 66 : self->FailureReported = false;
121 : 66 : self->EvictionReported = false;
122 : 66 : self->ReceiveFailureReported = false;
123 : 66 : self->Full = false;
124 : :
125 : : /* A slot whose open failed keeps the functions it had, so a handle left
126 : : * from an earlier Destroy of the slot stays a NullBuffer. */
127 [ + + ]: 66 : if (openErrno == 0)
128 : : {
129 : : /* The descriptor is all this process needs, and a forked child
130 : : * inherits it, so the name goes at once: nothing else can open the
131 : : * queue by it, and it is freed with the last descriptor. */
132 : 60 : (void) mq_unlink(queueName);
133 : 60 : self->Base.Write = PosixMessageQueueBuffer_Write;
134 : 60 : self->Base.Read = PosixMessageQueueBuffer_Read;
135 : 60 : self->Base.IsFull = PosixMessageQueueBuffer_IsFull;
136 : : }
137 : 66 : return openErrno;
138 : : }
139 : :
140 : 66 : static inline void PosixMessageQueueBuffer_InstallDiscardPolicy(
141 : : struct SolidSyslogPosixMessageQueueBuffer* self,
142 : : SolidSyslogDiscardPolicyFunction configured,
143 : : void* context
144 : : )
145 : : {
146 : 66 : self->GetDiscardPolicy = SolidSyslogDiscardPolicy_Default;
147 : 66 : self->DiscardPolicyContext = NULL;
148 [ + + ]: 66 : if (configured != NULL)
149 : : {
150 : 48 : self->GetDiscardPolicy = configured;
151 : 48 : self->DiscardPolicyContext = context;
152 : : }
153 : 66 : }
154 : :
155 : 60 : void SolidSyslogPosixMessageQueueBuffer_Cleanup(struct SolidSyslogBuffer* base)
156 : : {
157 : 60 : struct SolidSyslogPosixMessageQueueBuffer* self = PosixMessageQueueBuffer_SelfFromBase(base);
158 : 60 : mq_close(self->Mq);
159 : : /* Overwrite the abstract base with the shared NullBuffer vtable so
160 : : * use-after-destroy is a safe no-op rather than a NULL-fn-pointer crash. */
161 : 60 : *base = *SolidSyslogNullBuffer_Get();
162 : 60 : }
163 : :
164 : 68 : static bool PosixMessageQueueBuffer_Read(struct SolidSyslogBuffer* base, void* data, size_t maxSize, size_t* bytesRead)
165 : : {
166 : 68 : bool success = false;
167 [ + + ]: 68 : if (bytesRead != NULL)
168 : : {
169 : 67 : struct SolidSyslogPosixMessageQueueBuffer* self = PosixMessageQueueBuffer_SelfFromBase(base);
170 : 67 : ssize_t received = mq_receive(self->Mq, data, maxSize, NULL);
171 : 67 : success = received >= 0;
172 : :
173 : : /* Capture errno immediately after mq_receive so the EAGAIN test below
174 : : * stays a pure predicate and is decoupled from errno's lifetime
175 : : * between the errno-setting call and the read (MISRA C:2012 Rule 22.10).
176 : : * EAGAIN is the empty-queue poll signal - part of the happy path and
177 : : * must stay silent. Any other errno is a real failure worth surfacing. */
178 [ + + ]: 67 : int receiveErrno = success ? 0 : errno;
179 [ + + ]: 67 : if (success)
180 : : {
181 : 50 : self->ReceiveFailureReported = false;
182 : 50 : PosixMessageQueueBuffer_NoteRecordTaken(self);
183 : : }
184 [ + + ]: 17 : else if (receiveErrno != EAGAIN)
185 : : {
186 : 7 : PosixMessageQueueBuffer_NoteReceiveFailed(self);
187 : : }
188 : : else
189 : : {
190 : : /* Empty: the poll's quiet answer. */
191 : : }
192 : :
193 [ + + ]: 67 : *bytesRead = success ? (size_t) received : 0U;
194 : : }
195 : 68 : return success;
196 : : }
197 : :
198 : : /* A record taken out makes room, so the queue is no longer full. */
199 : 50 : static inline void PosixMessageQueueBuffer_NoteRecordTaken(struct SolidSyslogPosixMessageQueueBuffer* self)
200 : : {
201 : 50 : SolidSyslogMutex_Lock(self->Mutex);
202 : 50 : self->Full = false;
203 : 50 : SolidSyslogMutex_Unlock(self->Mutex);
204 : 50 : }
205 : :
206 : : /* Only the Service side receives, so the run needs no lock. It is marked
207 : : * before it is reported, as the send runs are. */
208 : 7 : static inline void PosixMessageQueueBuffer_NoteReceiveFailed(struct SolidSyslogPosixMessageQueueBuffer* self)
209 : : {
210 : 7 : bool runStarted = !self->ReceiveFailureReported;
211 : 7 : self->ReceiveFailureReported = true;
212 [ + + ]: 7 : if (runStarted)
213 : : {
214 : 5 : PosixMessageQueueBuffer_Report(
215 : : SOLIDSYSLOG_SEVERITY_ERROR,
216 : : SOLIDSYSLOG_CAT_BUFFER_BACKEND_FAILED,
217 : : SOLIDSYSLOG_POSIX_MESSAGE_QUEUE_BUFFER_ERROR_RECEIVE_FAILED
218 : : );
219 : : }
220 : 7 : }
221 : :
222 : 8 : static bool PosixMessageQueueBuffer_IsFull(struct SolidSyslogBuffer* base)
223 : : {
224 : 8 : struct SolidSyslogPosixMessageQueueBuffer* self = PosixMessageQueueBuffer_SelfFromBase(base);
225 : 8 : SolidSyslogMutex_Lock(self->Mutex);
226 : 8 : bool full = self->Full;
227 : 8 : SolidSyslogMutex_Unlock(self->Mutex);
228 : 8 : return full;
229 : : }
230 : :
231 : 176 : static void PosixMessageQueueBuffer_Write(struct SolidSyslogBuffer* base, const void* data, size_t size)
232 : : {
233 : 176 : struct SolidSyslogPosixMessageQueueBuffer* self = PosixMessageQueueBuffer_SelfFromBase(base);
234 : : /* Tasks logging at once share the run state. The policy is asked before
235 : : * the lock and the losses reported after it: both are integrator code that
236 : : * may log, which re-enters Write. */
237 : 176 : enum SolidSyslogDiscardPolicy policy = self->GetDiscardPolicy(self->DiscardPolicyContext);
238 : 176 : bool evictionRunStarted = false;
239 : 176 : SolidSyslogMutex_Lock(self->Mutex);
240 : : enum SolidSyslogPosixMessageQueueBufferRunStarted started =
241 : 176 : PosixMessageQueueBuffer_SendNotingRuns(self, policy, data, size, &evictionRunStarted);
242 : 176 : SolidSyslogMutex_Unlock(self->Mutex);
243 [ + + ]: 176 : if (evictionRunStarted)
244 : : {
245 : 12 : PosixMessageQueueBuffer_ReportRecordsEvicted();
246 : : }
247 : 176 : PosixMessageQueueBuffer_ReportRunStarted(started);
248 : 176 : }
249 : :
250 : : /* Sends the record, making room for it first where the discard policy says to,
251 : : * and says which run of losses, if any, this write starts: the one write of
252 : : * each run that is reported. A full queue, a queue that fails and a queue that
253 : : * had to evict are separate runs, so a queue that fills and then breaks still
254 : : * says it broke, and one that evicts and is still refused - another task filled
255 : : * the room - says both. The runs are marked here, before they are reported, so
256 : : * a handler that logs into the queue finds them already reported. */
257 : 176 : static enum SolidSyslogPosixMessageQueueBufferRunStarted PosixMessageQueueBuffer_SendNotingRuns(
258 : : struct SolidSyslogPosixMessageQueueBuffer* self,
259 : : enum SolidSyslogDiscardPolicy policy,
260 : : const void* data,
261 : : size_t size,
262 : : bool* evictionRunStarted
263 : : )
264 : : {
265 : 176 : enum SolidSyslogPosixMessageQueueBufferRunStarted started = RUN_STARTED_NONE;
266 : 176 : int sendErrno = 0;
267 : 176 : bool sent = PosixMessageQueueBuffer_Send(self, data, size, &sendErrno);
268 : 176 : bool evicted = false;
269 : :
270 [ + + + + : 176 : if (!sent && (sendErrno == EAGAIN) && (policy == SOLIDSYSLOG_DISCARD_POLICY_OLDEST))
+ + ]
271 : : {
272 : 14 : evicted = PosixMessageQueueBuffer_EvictOne(self);
273 [ + + + + ]: 14 : sent = evicted && PosixMessageQueueBuffer_Send(self, data, size, &sendErrno);
274 : : }
275 : 176 : *evictionRunStarted = PosixMessageQueueBuffer_NoteEvicted(self, evicted);
276 : :
277 [ + + ]: 176 : if (sent)
278 : : {
279 : 155 : self->RefusalReported = false;
280 : 155 : self->FailureReported = false;
281 : 155 : self->Full = evicted;
282 : : }
283 [ + + ]: 21 : else if (sendErrno == EAGAIN)
284 : : {
285 : 14 : started = self->RefusalReported ? RUN_STARTED_NONE : RUN_STARTED_REFUSALS;
286 : 14 : self->RefusalReported = true;
287 : 14 : self->Full = true;
288 : : }
289 : : else
290 : : {
291 [ + + ]: 7 : started = self->FailureReported ? RUN_STARTED_NONE : RUN_STARTED_FAILURES;
292 : 7 : self->FailureReported = true;
293 [ + + + + ]: 7 : self->Full = self->Full || evicted;
294 : : }
295 : :
296 : 176 : return started;
297 : : }
298 : :
299 : : /* A send that evicted continues or starts the run of evictions; one that
300 : : * evicted nothing ends it. Says whether a run started. */
301 : 176 : static inline bool PosixMessageQueueBuffer_NoteEvicted(struct SolidSyslogPosixMessageQueueBuffer* self, bool evicted)
302 : : {
303 [ + + + + ]: 176 : bool runStarted = evicted && !self->EvictionReported;
304 : 176 : self->EvictionReported = evicted;
305 : 176 : return runStarted;
306 : : }
307 : :
308 : 12 : static inline void PosixMessageQueueBuffer_ReportRecordsEvicted(void)
309 : : {
310 : 12 : PosixMessageQueueBuffer_Report(
311 : : SOLIDSYSLOG_SEVERITY_WARNING,
312 : : SOLIDSYSLOG_CAT_BUFFER_RECORDS_LOST,
313 : : SOLIDSYSLOG_POSIX_MESSAGE_QUEUE_BUFFER_ERROR_RECORDS_EVICTED
314 : : );
315 : 12 : }
316 : :
317 : : /* errno is captured straight after mq_send, as Read does, so the tests on it
318 : : * read errno only after the call that set it (MISRA C:2012 Rule 22.10). */
319 : 189 : static inline bool PosixMessageQueueBuffer_Send(
320 : : struct SolidSyslogPosixMessageQueueBuffer* self,
321 : : const void* data,
322 : : size_t size,
323 : : int* sendErrno
324 : : )
325 : : {
326 : 189 : bool sent = mq_send(self->Mq, data, size, 0) == 0;
327 [ + + ]: 189 : *sendErrno = sent ? 0 : errno;
328 : 189 : return sent;
329 : : }
330 : :
331 : : /* Takes the oldest record out of the queue and drops it. Fails when the queue
332 : : * has emptied meanwhile, or holds a record larger than the scratch; either
333 : : * way nothing was evicted and the send is not retried. */
334 : 14 : static inline bool PosixMessageQueueBuffer_EvictOne(const struct SolidSyslogPosixMessageQueueBuffer* self)
335 : : {
336 : : char evicted[SOLIDSYSLOG_MAX_MESSAGE_SIZE];
337 : 14 : return mq_receive(self->Mq, evicted, sizeof(evicted), NULL) >= 0;
338 : : }
339 : :
340 : : /* A full queue is a record lost to capacity; anything else is the queue
341 : : * failing. */
342 : 176 : static void PosixMessageQueueBuffer_ReportRunStarted(enum SolidSyslogPosixMessageQueueBufferRunStarted started)
343 : : {
344 [ + + ]: 176 : if (started == RUN_STARTED_REFUSALS)
345 : : {
346 : 12 : PosixMessageQueueBuffer_Report(
347 : : SOLIDSYSLOG_SEVERITY_WARNING,
348 : : SOLIDSYSLOG_CAT_BUFFER_RECORDS_LOST,
349 : : SOLIDSYSLOG_POSIX_MESSAGE_QUEUE_BUFFER_ERROR_RECORD_REFUSED
350 : : );
351 : : }
352 [ + + ]: 164 : else if (started == RUN_STARTED_FAILURES)
353 : : {
354 : 6 : PosixMessageQueueBuffer_Report(
355 : : SOLIDSYSLOG_SEVERITY_ERROR,
356 : : SOLIDSYSLOG_CAT_BUFFER_BACKEND_FAILED,
357 : : SOLIDSYSLOG_POSIX_MESSAGE_QUEUE_BUFFER_ERROR_SEND_FAILED
358 : : );
359 : : }
360 : : else
361 : : {
362 : : /* Sent, or a run already reported. */
363 : : }
364 : 176 : }
365 : :
366 : 377 : static inline struct SolidSyslogPosixMessageQueueBuffer* PosixMessageQueueBuffer_SelfFromBase(
367 : : struct SolidSyslogBuffer* base
368 : : )
369 : : {
370 : 377 : return (struct SolidSyslogPosixMessageQueueBuffer*) base;
371 : : }
|
