1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
|
/*
* (C) 2010-2016 by Holger Hans Peter Freyther
* (C) 2010 by On-Waves
*
* All Rights Reserved
*
* SPDX-License-Identifier: GPL-2.0+
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 2 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
*/
#include <errno.h>
#include <osmocom/core/write_queue.h>
#include <osmocom/core/logging.h>
/*! \addtogroup write_queue
* @{
* Write queue for writing \ref msgb to sockets/fds.
*
* \file write_queue.c */
/*! Select loop function for write queue handling
* \param[in] fd osmocom file descriptor
* \param[in] what bit-mask of events that have happened
* \returns 0 on success; negative on error
*
* This function is provided so that it can be registered with the
* select loop abstraction code (\ref osmo_fd::cb).
*/
int osmo_wqueue_bfd_cb(struct osmo_fd *fd, unsigned int what)
{
struct osmo_wqueue *queue;
int rc;
queue = container_of(fd, struct osmo_wqueue, bfd);
if (what & OSMO_FD_READ) {
rc = queue->read_cb(fd);
if (rc == -EBADF)
goto err_badfd;
}
if (what & OSMO_FD_EXCEPT) {
rc = queue->except_cb(fd);
if (rc == -EBADF)
goto err_badfd;
}
if (what & OSMO_FD_WRITE) {
struct msgb *msg;
fd->when &= ~OSMO_FD_WRITE;
msg = msgb_dequeue_count(&queue->msg_queue, &queue->current_length);
/* the queue might have been emptied */
if (msg) {
rc = queue->write_cb(fd, msg);
if (rc == -EBADF) {
msgb_free(msg);
goto err_badfd;
} else if (rc == -EAGAIN) {
/* re-enqueue the msgb to the head of the queue */
llist_add(&msg->list, &queue->msg_queue);
queue->current_length++;
} else
msgb_free(msg);
if (!llist_empty(&queue->msg_queue))
fd->when |= OSMO_FD_WRITE;
}
}
err_badfd:
/* Return value is not checked in osmo_select_main() */
return 0;
}
/*! Initialize a \ref osmo_wqueue structure
* \param[in] queue Write queue to operate on
* \param[in] max_length Maximum length of write queue
*/
void osmo_wqueue_init(struct osmo_wqueue *queue, int max_length)
{
queue->max_length = max_length;
queue->current_length = 0;
queue->read_cb = NULL;
queue->write_cb = NULL;
queue->except_cb = NULL;
queue->bfd.cb = osmo_wqueue_bfd_cb;
INIT_LLIST_HEAD(&queue->msg_queue);
}
/*! Enqueue a new \ref msgb into a write queue (without logging full queue events)
* \param[in] queue Write queue to be used
* \param[in] data to-be-enqueued message buffer
* \returns 0 on success; negative on error (MESSAGE NOT FREED IN CASE OF ERROR).
*/
int osmo_wqueue_enqueue_quiet(struct osmo_wqueue *queue, struct msgb *data)
{
if (queue->current_length >= queue->max_length)
return -ENOSPC;
msgb_enqueue_count(&queue->msg_queue, data, &queue->current_length);
queue->bfd.when |= OSMO_FD_WRITE;
return 0;
}
/*! Enqueue a new \ref msgb into a write queue
* \param[in] queue Write queue to be used
* \param[in] data to-be-enqueued message buffer
* \returns 0 on success; negative on error (MESSAGE NOT FREED IN CASE OF ERROR).
*/
int osmo_wqueue_enqueue(struct osmo_wqueue *queue, struct msgb *data)
{
if (queue->current_length >= queue->max_length) {
LOGP(DLGLOBAL, LOGL_ERROR,
"wqueue(%p) is full. Rejecting msgb\n", queue);
return -ENOSPC;
}
return osmo_wqueue_enqueue_quiet(queue, data);
}
/*! Clear a \ref osmo_wqueue
* \param[in] queue Write queue to be cleared
*
* This function will clear (remove/release) all messages in it.
*/
void osmo_wqueue_clear(struct osmo_wqueue *queue)
{
while (!llist_empty(&queue->msg_queue)) {
struct msgb *msg = msgb_dequeue(&queue->msg_queue);
msgb_free(msg);
}
queue->current_length = 0;
queue->bfd.when &= ~OSMO_FD_WRITE;
}
/*! Update write queue length & drop excess messages.
* \param[in] queue linked list header of message queue
* \param[in] len new max. wqueue length
* \returns Number of messages dropped.
*
* Messages beyond the new maximum message queue size will be dropped.
*/
size_t osmo_wqueue_set_maxlen(struct osmo_wqueue *queue, unsigned int len)
{
size_t dropped_msgs = 0;
struct msgb *msg;
queue->max_length = len;
while (queue->current_length > queue->max_length) {
msg = msgb_dequeue_count(&queue->msg_queue, &queue->current_length);
msgb_free(msg);
dropped_msgs++;
}
return dropped_msgs;
}
/*! @} */
|
