4 Copyright (C) Andrew Tridgell 2006
5 Copyright (C) Ronnie Sahlberg 2008
7 This program is free software; you can redistribute it and/or modify
8 it under the terms of the GNU General Public License as published by
9 the Free Software Foundation; either version 3 of the License, or
10 (at your option) any later version.
12 This program is distributed in the hope that it will be useful,
13 but WITHOUT ANY WARRANTY; without even the implied warranty of
14 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 GNU General Public License for more details.
17 You should have received a copy of the GNU General Public License
18 along with this program; if not, see <http://www.gnu.org/licenses/>.
23 #include "system/network.h"
24 #include "system/filesys.h"
25 #include "../include/ctdb_private.h"
29 stop any connecting (established or pending) to a node
31 void ctdb_tcp_stop_connection(struct ctdb_node
*node
)
33 struct ctdb_tcp_node
*tnode
= talloc_get_type(
34 node
->private_data
, struct ctdb_tcp_node
);
36 ctdb_queue_set_fd(tnode
->out_queue
, -1);
37 talloc_free(tnode
->connect_te
);
38 talloc_free(tnode
->connect_fde
);
39 tnode
->connect_fde
= NULL
;
40 tnode
->connect_te
= NULL
;
41 if (tnode
->fd
!= -1) {
49 called when a complete packet has come in - should not happen on this socket
50 unless the other side closes the connection with RST or FIN
52 void ctdb_tcp_tnode_cb(uint8_t *data
, size_t cnt
, void *private_data
)
54 struct ctdb_node
*node
= talloc_get_type(private_data
, struct ctdb_node
);
55 struct ctdb_tcp_node
*tnode
= talloc_get_type(
56 node
->private_data
, struct ctdb_tcp_node
);
59 node
->ctdb
->upcalls
->node_dead(node
);
62 ctdb_tcp_stop_connection(node
);
63 tnode
->connect_te
= event_add_timed(node
->ctdb
->ev
, tnode
,
64 timeval_current_ofs(3, 0),
65 ctdb_tcp_node_connect
, node
);
69 called when socket becomes writeable on connect
71 static void ctdb_node_connect_write(struct event_context
*ev
, struct fd_event
*fde
,
72 uint16_t flags
, void *private_data
)
74 struct ctdb_node
*node
= talloc_get_type(private_data
,
76 struct ctdb_tcp_node
*tnode
= talloc_get_type(node
->private_data
,
77 struct ctdb_tcp_node
);
78 struct ctdb_context
*ctdb
= node
->ctdb
;
80 socklen_t len
= sizeof(error
);
83 talloc_free(tnode
->connect_te
);
84 tnode
->connect_te
= NULL
;
86 if (getsockopt(tnode
->fd
, SOL_SOCKET
, SO_ERROR
, &error
, &len
) != 0 ||
88 ctdb_tcp_stop_connection(node
);
89 tnode
->connect_te
= event_add_timed(ctdb
->ev
, tnode
,
90 timeval_current_ofs(1, 0),
91 ctdb_tcp_node_connect
, node
);
95 talloc_free(tnode
->connect_fde
);
96 tnode
->connect_fde
= NULL
;
98 setsockopt(tnode
->fd
,IPPROTO_TCP
,TCP_NODELAY
,(char *)&one
,sizeof(one
));
99 setsockopt(tnode
->fd
,SOL_SOCKET
,SO_KEEPALIVE
,(char *)&one
,sizeof(one
));
101 ctdb_queue_set_fd(tnode
->out_queue
, tnode
->fd
);
103 /* the queue subsystem now owns this fd */
108 static int ctdb_tcp_get_address(struct ctdb_context
*ctdb
,
109 const char *address
, ctdb_sock_addr
*addr
)
111 if (parse_ip(address
, NULL
, 0, addr
) == 0) {
112 DEBUG(DEBUG_CRIT
, (__location__
" Unparsable address : %s.\n", address
));
119 called when we should try and establish a tcp connection to a node
121 void ctdb_tcp_node_connect(struct event_context
*ev
, struct timed_event
*te
,
122 struct timeval t
, void *private_data
)
124 struct ctdb_node
*node
= talloc_get_type(private_data
,
126 struct ctdb_tcp_node
*tnode
= talloc_get_type(node
->private_data
,
127 struct ctdb_tcp_node
);
128 struct ctdb_context
*ctdb
= node
->ctdb
;
129 ctdb_sock_addr sock_in
;
132 ctdb_sock_addr sock_out
;
134 ctdb_tcp_stop_connection(node
);
136 ZERO_STRUCT(sock_out
);
137 #ifdef HAVE_SOCK_SIN_LEN
138 sock_out
.ip
.sin_len
= sizeof(sock_out
);
140 if (ctdb_tcp_get_address(ctdb
, node
->address
.address
, &sock_out
) != 0) {
143 switch (sock_out
.sa
.sa_family
) {
145 sock_out
.ip
.sin_port
= htons(node
->address
.port
);
148 sock_out
.ip6
.sin6_port
= htons(node
->address
.port
);
151 DEBUG(DEBUG_ERR
, (__location__
" unknown family %u\n",
152 sock_out
.sa
.sa_family
));
156 tnode
->fd
= socket(sock_out
.sa
.sa_family
, SOCK_STREAM
, IPPROTO_TCP
);
157 set_nonblocking(tnode
->fd
);
158 set_close_on_exec(tnode
->fd
);
160 DEBUG(DEBUG_DEBUG
, (__location__
" Created TCP SOCKET FD:%d\n", tnode
->fd
));
162 /* Bind our side of the socketpair to the same address we use to listen
163 * on incoming CTDB traffic.
164 * We must specify this address to make sure that the address we expose to
165 * the remote side is actually routable in case CTDB traffic will run on
166 * a dedicated non-routeable network.
168 ZERO_STRUCT(sock_in
);
169 if (ctdb_tcp_get_address(ctdb
, ctdb
->address
.address
, &sock_in
) != 0) {
170 DEBUG(DEBUG_ERR
, (__location__
" Failed to find our address. Failing bind.\n"));
175 /* AIX libs check to see if the socket address and length
176 arguments are consistent with each other on calls like
177 connect(). Can not get by with just sizeof(sock_in),
178 need sizeof(sock_in.ip).
180 switch (sock_in
.sa
.sa_family
) {
182 sockin_size
= sizeof(sock_in
.ip
);
183 sockout_size
= sizeof(sock_out
.ip
);
186 sockin_size
= sizeof(sock_in
.ip6
);
187 sockout_size
= sizeof(sock_out
.ip6
);
190 DEBUG(DEBUG_ERR
, (__location__
" unknown family %u\n",
191 sock_in
.sa
.sa_family
));
195 #ifdef HAVE_SOCK_SIN_LEN
196 sock_in
.ip
.sin_len
= sockin_size
;
197 sock_out
.ip
.sin_len
= sockout_size
;
199 bind(tnode
->fd
, (struct sockaddr
*)&sock_in
, sockin_size
);
201 if (connect(tnode
->fd
, (struct sockaddr
*)&sock_out
, sockout_size
) != 0 &&
202 errno
!= EINPROGRESS
) {
203 ctdb_tcp_stop_connection(node
);
204 tnode
->connect_te
= event_add_timed(ctdb
->ev
, tnode
,
205 timeval_current_ofs(1, 0),
206 ctdb_tcp_node_connect
, node
);
210 /* non-blocking connect - wait for write event */
211 tnode
->connect_fde
= event_add_fd(node
->ctdb
->ev
, tnode
, tnode
->fd
,
212 EVENT_FD_WRITE
|EVENT_FD_READ
,
213 ctdb_node_connect_write
, node
);
215 /* don't give it long to connect - retry in one second. This ensures
216 that we find a node is up quickly (tcp normally backs off a syn reply
217 delay by quite a lot) */
218 tnode
->connect_te
= event_add_timed(ctdb
->ev
, tnode
, timeval_current_ofs(1, 0),
219 ctdb_tcp_node_connect
, node
);
223 called when we get contacted by another node
224 currently makes no attempt to check if the connection is really from a ctdb
227 static void ctdb_listen_event(struct event_context
*ev
, struct fd_event
*fde
,
228 uint16_t flags
, void *private_data
)
230 struct ctdb_context
*ctdb
= talloc_get_type(private_data
, struct ctdb_context
);
231 struct ctdb_tcp
*ctcp
= talloc_get_type(ctdb
->private_data
, struct ctdb_tcp
);
235 struct ctdb_incoming
*in
;
237 const char *incoming_node
;
239 memset(&addr
, 0, sizeof(addr
));
241 fd
= accept(ctcp
->listen_fd
, (struct sockaddr
*)&addr
, &len
);
242 if (fd
== -1) return;
244 incoming_node
= ctdb_addr_to_str(&addr
);
245 nodeid
= ctdb_ip_to_nodeid(ctdb
, incoming_node
);
248 DEBUG(DEBUG_ERR
, ("Refused connection from unknown node %s\n", incoming_node
));
253 in
= talloc_zero(ctcp
, struct ctdb_incoming
);
257 set_nonblocking(in
->fd
);
258 set_close_on_exec(in
->fd
);
260 DEBUG(DEBUG_DEBUG
, (__location__
" Created SOCKET FD:%d to incoming ctdb connection\n", fd
));
262 setsockopt(in
->fd
,SOL_SOCKET
,SO_KEEPALIVE
,(char *)&one
,sizeof(one
));
264 in
->queue
= ctdb_queue_setup(ctdb
, in
, in
->fd
, CTDB_TCP_ALIGNMENT
,
265 ctdb_tcp_read_cb
, in
, "ctdbd-%s", incoming_node
);
270 automatically find which address to listen on
272 static int ctdb_tcp_listen_automatic(struct ctdb_context
*ctdb
)
274 struct ctdb_tcp
*ctcp
= talloc_get_type(ctdb
->private_data
,
278 const char *lock_path
= "/tmp/.ctdb_socket_lock";
282 struct tevent_fd
*fde
;
284 /* If there are no nodes, then it won't be possible to find
285 * the first one. Log a failure and short circuit the whole
288 if (ctdb
->num_nodes
== 0) {
289 DEBUG(DEBUG_CRIT
,("No nodes available to attempt bind to - is the nodes file empty?\n"));
293 /* in order to ensure that we don't get two nodes with the
294 same adddress, we must make the bind() and listen() calls
295 atomic. The SO_REUSEADDR setsockopt only prevents double
296 binds if the first socket is in LISTEN state */
297 lock_fd
= open(lock_path
, O_RDWR
|O_CREAT
, 0666);
299 DEBUG(DEBUG_CRIT
,("Unable to open %s\n", lock_path
));
303 lock
.l_type
= F_WRLCK
;
304 lock
.l_whence
= SEEK_SET
;
309 if (fcntl(lock_fd
, F_SETLKW
, &lock
) != 0) {
310 DEBUG(DEBUG_CRIT
,("Unable to lock %s\n", lock_path
));
315 for (i
=0; i
< ctdb
->num_nodes
; i
++) {
316 if (ctdb
->nodes
[i
]->flags
& NODE_FLAGS_DELETED
) {
320 if (ctdb_tcp_get_address(ctdb
,
321 ctdb
->nodes
[i
]->address
.address
,
326 switch (sock
.sa
.sa_family
) {
328 sock
.ip
.sin_port
= htons(ctdb
->nodes
[i
]->address
.port
);
329 sock_size
= sizeof(sock
.ip
);
332 sock
.ip6
.sin6_port
= htons(ctdb
->nodes
[i
]->address
.port
);
333 sock_size
= sizeof(sock
.ip6
);
336 DEBUG(DEBUG_ERR
, (__location__
" unknown family %u\n",
340 #ifdef HAVE_SOCK_SIN_LEN
341 sock
.ip
.sin_len
= sock_size
;
344 ctcp
->listen_fd
= socket(sock
.sa
.sa_family
, SOCK_STREAM
, IPPROTO_TCP
);
345 if (ctcp
->listen_fd
== -1) {
346 ctdb_set_error(ctdb
, "socket failed\n");
350 set_close_on_exec(ctcp
->listen_fd
);
352 setsockopt(ctcp
->listen_fd
,SOL_SOCKET
,SO_REUSEADDR
,(char *)&one
,sizeof(one
));
354 if (bind(ctcp
->listen_fd
, (struct sockaddr
* )&sock
, sock_size
) == 0) {
358 if (errno
== EADDRNOTAVAIL
) {
359 DEBUG(DEBUG_DEBUG
,(__location__
" Failed to bind() to socket. %s(%d)\n",
360 strerror(errno
), errno
));
362 DEBUG(DEBUG_ERR
,(__location__
" Failed to bind() to socket. %s(%d)\n",
363 strerror(errno
), errno
));
367 if (i
== ctdb
->num_nodes
) {
368 DEBUG(DEBUG_CRIT
,("Unable to bind to any of the node addresses - giving up\n"));
371 ctdb
->address
.address
= talloc_strdup(ctdb
, ctdb
->nodes
[i
]->address
.address
);
372 ctdb
->address
.port
= ctdb
->nodes
[i
]->address
.port
;
373 ctdb
->name
= talloc_asprintf(ctdb
, "%s:%u",
374 ctdb
->address
.address
,
376 ctdb
->pnn
= ctdb
->nodes
[i
]->pnn
;
377 DEBUG(DEBUG_INFO
,("ctdb chose network address %s:%u pnn %u\n",
378 ctdb
->address
.address
,
382 if (listen(ctcp
->listen_fd
, 10) == -1) {
386 fde
= event_add_fd(ctdb
->ev
, ctcp
, ctcp
->listen_fd
, EVENT_FD_READ
,
387 ctdb_listen_event
, ctdb
);
388 tevent_fd_set_auto_close(fde
);
396 close(ctcp
->listen_fd
);
397 ctcp
->listen_fd
= -1;
403 listen on our own address
405 int ctdb_tcp_listen(struct ctdb_context
*ctdb
)
407 struct ctdb_tcp
*ctcp
= talloc_get_type(ctdb
->private_data
,
412 struct tevent_fd
*fde
;
414 /* we can either auto-bind to the first available address, or we can
415 use a specified address */
416 if (!ctdb
->address
.address
) {
417 return ctdb_tcp_listen_automatic(ctdb
);
421 if (ctdb_tcp_get_address(ctdb
, ctdb
->address
.address
,
426 switch (sock
.sa
.sa_family
) {
428 sock
.ip
.sin_port
= htons(ctdb
->address
.port
);
429 sock_size
= sizeof(sock
.ip
);
432 sock
.ip6
.sin6_port
= htons(ctdb
->address
.port
);
433 sock_size
= sizeof(sock
.ip6
);
436 DEBUG(DEBUG_ERR
, (__location__
" unknown family %u\n",
440 #ifdef HAVE_SOCK_SIN_LEN
441 sock
.ip
.sin_len
= sock_size
;
444 ctcp
->listen_fd
= socket(sock
.sa
.sa_family
, SOCK_STREAM
, IPPROTO_TCP
);
445 if (ctcp
->listen_fd
== -1) {
446 ctdb_set_error(ctdb
, "socket failed\n");
450 set_close_on_exec(ctcp
->listen_fd
);
452 setsockopt(ctcp
->listen_fd
,SOL_SOCKET
,SO_REUSEADDR
,(char *)&one
,sizeof(one
));
454 if (bind(ctcp
->listen_fd
, (struct sockaddr
* )&sock
, sock_size
) != 0) {
455 DEBUG(DEBUG_ERR
,(__location__
" Failed to bind() to socket. %s(%d)\n", strerror(errno
), errno
));
459 if (listen(ctcp
->listen_fd
, 10) == -1) {
463 fde
= event_add_fd(ctdb
->ev
, ctcp
, ctcp
->listen_fd
, EVENT_FD_READ
,
464 ctdb_listen_event
, ctdb
);
465 tevent_fd_set_auto_close(fde
);
470 if (ctcp
->listen_fd
!= -1) {
471 close(ctcp
->listen_fd
);
473 ctcp
->listen_fd
= -1;