2 Generic Unix-domain Socket I/O
4 Copyright (C) Amitay Isaacs 2016
6 This program is free software; you can redistribute it and/or modify
7 it under the terms of the GNU General Public License as published by
8 the Free Software Foundation; either version 3 of the License, or
9 (at your option) any later version.
11 This program is distributed in the hope that it will be useful,
12 but WITHOUT ANY WARRANTY; without even the implied warranty of
13 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 GNU General Public License for more details.
16 You should have received a copy of the GNU General Public License
17 along with this program; if not, see <http://www.gnu.org/licenses/>.
21 #include "system/filesys.h"
22 #include "system/network.h"
27 #include "lib/util/sys_rw.h"
28 #include "lib/util/debug.h"
29 #include "lib/util/blocking.h"
31 #include "common/logging.h"
32 #include "common/sock_io.h"
34 bool sock_clean(const char *sockpath
)
38 ret
= unlink(sockpath
);
40 D_WARNING("Removed stale socket %s\n", sockpath
);
41 } else if (errno
!= ENOENT
) {
42 D_ERR("Failed to remove stale socket %s\n", sockpath
);
49 int sock_connect(const char *sockpath
)
51 struct sockaddr_un addr
;
55 if (sockpath
== NULL
) {
56 D_ERR("Invalid socket path\n");
60 memset(&addr
, 0, sizeof(addr
));
61 addr
.sun_family
= AF_UNIX
;
62 len
= strlcpy(addr
.sun_path
, sockpath
, sizeof(addr
.sun_path
));
63 if (len
>= sizeof(addr
.sun_path
)) {
64 D_ERR("Socket path too long, len=%zu\n", strlen(sockpath
));
68 fd
= socket(AF_UNIX
, SOCK_STREAM
, 0);
70 D_ERR("socket() failed, errno=%d\n", errno
);
74 ret
= connect(fd
, (struct sockaddr
*)&addr
, sizeof(addr
));
76 D_ERR("connect() failed, errno=%d\n", errno
);
85 struct tevent_context
*ev
;
86 sock_queue_callback_fn_t callback
;
90 struct tevent_immediate
*im
;
91 struct tevent_queue
*queue
;
92 struct tevent_fd
*fde
;
94 size_t buflen
, begin
, end
;
97 static bool sock_queue_set_fd(struct sock_queue
*queue
, int fd
);
98 static int sock_queue_destructor(struct sock_queue
*queue
);
99 static void sock_queue_handler(struct tevent_context
*ev
,
100 struct tevent_fd
*fde
, uint16_t flags
,
102 static void sock_queue_process(struct sock_queue
*queue
);
103 static void sock_queue_process_event(struct tevent_context
*ev
,
104 struct tevent_immediate
*im
,
107 struct sock_queue
*sock_queue_setup(TALLOC_CTX
*mem_ctx
,
108 struct tevent_context
*ev
,
110 sock_queue_callback_fn_t callback
,
113 struct sock_queue
*queue
;
115 queue
= talloc_zero(mem_ctx
, struct sock_queue
);
121 queue
->callback
= callback
;
122 queue
->private_data
= private_data
;
124 queue
->im
= tevent_create_immediate(queue
);
125 if (queue
->im
== NULL
) {
130 queue
->queue
= tevent_queue_create(queue
, "out-queue");
131 if (queue
->queue
== NULL
) {
136 if (! sock_queue_set_fd(queue
, fd
)) {
141 talloc_set_destructor(queue
, sock_queue_destructor
);
146 static bool sock_queue_set_fd(struct sock_queue
*queue
, int fd
)
148 TALLOC_FREE(queue
->fde
);
154 ret
= set_blocking(fd
, false);
159 queue
->fde
= tevent_add_fd(queue
->ev
, queue
, fd
,
161 sock_queue_handler
, queue
);
162 if (queue
->fde
== NULL
) {
165 tevent_fd_set_auto_close(queue
->fde
);
171 static int sock_queue_destructor(struct sock_queue
*queue
)
173 TALLOC_FREE(queue
->fde
);
179 static void sock_queue_handler(struct tevent_context
*ev
,
180 struct tevent_fd
*fde
, uint16_t flags
,
183 struct sock_queue
*queue
= talloc_get_type_abort(
184 private_data
, struct sock_queue
);
188 ret
= ioctl(queue
->fd
, FIONREAD
, &num_ready
);
194 if (num_ready
== 0) {
195 /* descriptor has been closed */
199 if (num_ready
> queue
->buflen
- queue
->end
) {
200 queue
->buf
= talloc_realloc_size(queue
, queue
->buf
,
201 queue
->end
+ num_ready
);
202 if (queue
->buf
== NULL
) {
205 queue
->buflen
= queue
->end
+ num_ready
;
208 nread
= sys_read(queue
->fd
, queue
->buf
+ queue
->end
, num_ready
);
214 sock_queue_process(queue
);
218 queue
->callback(NULL
, 0, queue
->private_data
);
221 static void sock_queue_process(struct sock_queue
*queue
)
225 if ((queue
->end
- queue
->begin
) < sizeof(uint32_t)) {
226 /* not enough data */
230 pkt_size
= *(uint32_t *)(queue
->buf
+ queue
->begin
);
232 D_ERR("Invalid packet of length 0\n");
233 queue
->callback(NULL
, 0, queue
->private_data
);
236 if ((queue
->end
- queue
->begin
) < pkt_size
) {
237 /* not enough data */
241 queue
->callback(queue
->buf
+ queue
->begin
, pkt_size
,
242 queue
->private_data
);
243 queue
->begin
+= pkt_size
;
245 if (queue
->begin
< queue
->end
) {
246 /* more data to be processed */
247 tevent_schedule_immediate(queue
->im
, queue
->ev
,
248 sock_queue_process_event
, queue
);
250 TALLOC_FREE(queue
->buf
);
257 static void sock_queue_process_event(struct tevent_context
*ev
,
258 struct tevent_immediate
*im
,
261 struct sock_queue
*queue
= talloc_get_type_abort(
262 private_data
, struct sock_queue
);
264 sock_queue_process(queue
);
267 struct sock_queue_write_state
{
272 static void sock_queue_trigger(struct tevent_req
*req
, void *private_data
);
274 int sock_queue_write(struct sock_queue
*queue
, uint8_t *buf
, size_t buflen
)
276 struct tevent_req
*req
;
277 struct sock_queue_write_state
*state
;
280 if (buflen
>= INT32_MAX
) {
284 req
= tevent_req_create(queue
, &state
, struct sock_queue_write_state
);
290 state
->pkt_size
= (uint32_t)buflen
;
292 status
= tevent_queue_add_entry(queue
->queue
, queue
->ev
, req
,
293 sock_queue_trigger
, queue
);
302 static void sock_queue_trigger(struct tevent_req
*req
, void *private_data
)
304 struct sock_queue
*queue
= talloc_get_type_abort(
305 private_data
, struct sock_queue
);
306 struct sock_queue_write_state
*state
= tevent_req_data(
307 req
, struct sock_queue_write_state
);
313 nwritten
= sys_write(queue
->fd
, state
->pkt
+ offset
,
314 state
->pkt_size
- offset
);
316 queue
->callback(NULL
, 0, queue
->private_data
);
321 } while (offset
< state
->pkt_size
);
323 tevent_req_done(req
);