4 Copyright (C) Andrew Tridgell 2006
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 "lib/events/events.h"
22 #include "lib/tdb/include/tdb.h"
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
51 void ctdb_tcp_tnode_cb(uint8_t *data, size_t cnt, void *private_data)
53 struct ctdb_node *node = talloc_get_type(private_data, struct ctdb_node);
54 struct ctdb_tcp_node *tnode = talloc_get_type(
55 node->private_data, struct ctdb_tcp_node);
58 node->ctdb->upcalls->node_dead(node);
61 ctdb_tcp_stop_connection(node);
62 tnode->connect_te = event_add_timed(node->ctdb->ev, tnode, timeval_zero(),
63 ctdb_tcp_node_connect, node);
67 called when socket becomes writeable on connect
69 static void ctdb_node_connect_write(struct event_context *ev, struct fd_event *fde,
70 uint16_t flags, void *private_data)
72 struct ctdb_node *node = talloc_get_type(private_data,
74 struct ctdb_tcp_node *tnode = talloc_get_type(node->private_data,
75 struct ctdb_tcp_node);
76 struct ctdb_context *ctdb = node->ctdb;
78 socklen_t len = sizeof(error);
81 talloc_free(tnode->connect_te);
82 tnode->connect_te = NULL;
84 if (getsockopt(tnode->fd, SOL_SOCKET, SO_ERROR, &error, &len) != 0 ||
86 ctdb_tcp_stop_connection(node);
87 tnode->connect_te = event_add_timed(ctdb->ev, tnode,
88 timeval_current_ofs(1, 0),
89 ctdb_tcp_node_connect, node);
93 talloc_free(tnode->connect_fde);
94 tnode->connect_fde = NULL;
96 setsockopt(tnode->fd,IPPROTO_TCP,TCP_NODELAY,(char *)&one,sizeof(one));
97 setsockopt(tnode->fd,SOL_SOCKET,SO_KEEPALIVE,(char *)&one,sizeof(one));
99 ctdb_queue_set_fd(tnode->out_queue, tnode->fd);
101 /* the queue subsystem now owns this fd */
106 static int ctdb_tcp_get_address(struct ctdb_context *ctdb,
107 const char *address, ctdb_sock_addr *addr)
109 if (parse_ip(address, addr) == 0) {
110 DEBUG(DEBUG_CRIT, (__location__ " Unparsable address : %s.\n", address));
117 called when we should try and establish a tcp connection to a node
119 void ctdb_tcp_node_connect(struct event_context *ev, struct timed_event *te,
120 struct timeval t, void *private_data)
122 struct ctdb_node *node = talloc_get_type(private_data,
124 struct ctdb_tcp_node *tnode = talloc_get_type(node->private_data,
125 struct ctdb_tcp_node);
126 struct ctdb_context *ctdb = node->ctdb;
127 ctdb_sock_addr sock_in;
128 ctdb_sock_addr sock_out;
130 ctdb_tcp_stop_connection(node);
132 ZERO_STRUCT(sock_out);
133 #ifdef HAVE_SOCK_SIN_LEN
134 sock_out.ip.sin_len = sizeof(sock_out);
136 if (ctdb_tcp_get_address(ctdb, node->address.address, &sock_out) != 0) {
139 switch (sock_out.sa.sa_family) {
141 sock_out.ip.sin_port = htons(node->address.port);
144 sock_out.ip6.sin6_port = htons(node->address.port);
147 DEBUG(DEBUG_ERR, (__location__ " unknown family %u\n",
148 sock_out.sa.sa_family));
152 tnode->fd = socket(sock_out.sa.sa_family, SOCK_STREAM, IPPROTO_TCP);
153 set_nonblocking(tnode->fd);
154 set_close_on_exec(tnode->fd);
156 /* Bind our side of the socketpair to the same address we use to listen
157 * on incoming CTDB traffic.
158 * We must specify this address to make sure that the address we expose to
159 * the remote side is actually routable in case CTDB traffic will run on
160 * a dedicated non-routeable network.
162 ZERO_STRUCT(sock_in);
163 #ifdef HAVE_SOCK_SIN_LEN
164 sock_in.ip.sin_len = sizeof(sock_in);
166 if (ctdb_tcp_get_address(ctdb, ctdb->address.address, &sock_in) != 0) {
169 bind(tnode->fd, (struct sockaddr *)&sock_in, sizeof(sock_in));
171 if (connect(tnode->fd, (struct sockaddr *)&sock_out, sizeof(sock_out)) != 0 &&
172 errno != EINPROGRESS) {
173 ctdb_tcp_stop_connection(node);
174 tnode->connect_te = event_add_timed(ctdb->ev, tnode,
175 timeval_current_ofs(1, 0),
176 ctdb_tcp_node_connect, node);
180 /* non-blocking connect - wait for write event */
181 tnode->connect_fde = event_add_fd(node->ctdb->ev, tnode, tnode->fd,
182 EVENT_FD_WRITE|EVENT_FD_READ,
183 ctdb_node_connect_write, node);
185 /* don't give it long to connect - retry in one second. This ensures
186 that we find a node is up quickly (tcp normally backs off a syn reply
187 delay by quite a lot) */
188 tnode->connect_te = event_add_timed(ctdb->ev, tnode, timeval_current_ofs(1, 0),
189 ctdb_tcp_node_connect, node);
193 called when we get contacted by another node
194 currently makes no attempt to check if the connection is really from a ctdb
197 static void ctdb_listen_event(struct event_context *ev, struct fd_event *fde,
198 uint16_t flags, void *private_data)
200 struct ctdb_context *ctdb = talloc_get_type(private_data, struct ctdb_context);
201 struct ctdb_tcp *ctcp = talloc_get_type(ctdb->private_data, struct ctdb_tcp);
205 struct ctdb_incoming *in;
207 const char *incoming_node;
209 memset(&addr, 0, sizeof(addr));
211 fd = accept(ctcp->listen_fd, (struct sockaddr *)&addr, &len);
212 if (fd == -1) return;
214 incoming_node = ctdb_addr_to_str(&addr);
215 nodeid = ctdb_ip_to_nodeid(ctdb, incoming_node);
218 DEBUG(DEBUG_ERR, ("Refused connection from unknown node %s\n", incoming_node));
223 in = talloc_zero(ctcp, struct ctdb_incoming);
227 set_nonblocking(in->fd);
228 set_close_on_exec(in->fd);
230 setsockopt(in->fd,SOL_SOCKET,SO_KEEPALIVE,(char *)&one,sizeof(one));
232 in->queue = ctdb_queue_setup(ctdb, in, in->fd, CTDB_TCP_ALIGNMENT,
233 ctdb_tcp_read_cb, in);
238 automatically find which address to listen on
240 static int ctdb_tcp_listen_automatic(struct ctdb_context *ctdb)
242 struct ctdb_tcp *ctcp = talloc_get_type(ctdb->private_data,
246 const char *lock_path = "/tmp/.ctdb_socket_lock";
250 /* in order to ensure that we don't get two nodes with the
251 same adddress, we must make the bind() and listen() calls
252 atomic. The SO_REUSEADDR setsockopt only prevents double
253 binds if the first socket is in LISTEN state */
254 lock_fd = open(lock_path, O_RDWR|O_CREAT, 0666);
256 DEBUG(DEBUG_CRIT,("Unable to open %s\n", lock_path));
260 lock.l_type = F_WRLCK;
261 lock.l_whence = SEEK_SET;
266 if (fcntl(lock_fd, F_SETLKW, &lock) != 0) {
267 DEBUG(DEBUG_CRIT,("Unable to lock %s\n", lock_path));
272 for (i=0;i<ctdb->num_nodes;i++) {
273 /* if node_ip is specified we will only try to bind to that
276 if (ctdb->node_ip != NULL) {
277 if (strcmp(ctdb->node_ip, ctdb->nodes[i]->address.address)) {
283 #ifdef HAVE_SOCK_SIN_LEN
284 sock.ip.sin_len = sizeof(sock);
286 if (ctdb_tcp_get_address(ctdb,
287 ctdb->nodes[i]->address.address,
292 switch (sock.sa.sa_family) {
294 sock.ip.sin_port = htons(ctdb->nodes[i]->address.port);
297 sock.ip6.sin6_port = htons(ctdb->nodes[i]->address.port);
300 DEBUG(DEBUG_ERR, (__location__ " unknown family %u\n",
305 ctcp->listen_fd = socket(sock.sa.sa_family, SOCK_STREAM, IPPROTO_TCP);
306 if (ctcp->listen_fd == -1) {
307 ctdb_set_error(ctdb, "socket failed\n");
311 set_close_on_exec(ctcp->listen_fd);
313 setsockopt(ctcp->listen_fd,SOL_SOCKET,SO_REUSEADDR,(char *)&one,sizeof(one));
315 if (bind(ctcp->listen_fd, (struct sockaddr * )&sock,
316 sizeof(sock)) == 0) {
321 if (i == ctdb->num_nodes) {
322 DEBUG(DEBUG_CRIT,("Unable to bind to any of the node addresses - giving up\n"));
325 ctdb->address.address = talloc_strdup(ctdb, ctdb->nodes[i]->address.address);
326 ctdb->address.port = ctdb->nodes[i]->address.port;
327 ctdb->name = talloc_asprintf(ctdb, "%s:%u",
328 ctdb->address.address,
330 ctdb->pnn = ctdb->nodes[i]->pnn;
331 ctdb->nodes[i]->flags &= ~NODE_FLAGS_DISCONNECTED;
332 DEBUG(DEBUG_INFO,("ctdb chose network address %s:%u pnn %u\n",
333 ctdb->address.address,
336 /* do we start out in DISABLED mode? */
337 if (ctdb->start_as_disabled != 0) {
338 DEBUG(DEBUG_INFO, ("This node is configured to start in DISABLED state\n"));
339 ctdb->nodes[i]->flags |= NODE_FLAGS_DISABLED;
342 if (listen(ctcp->listen_fd, 10) == -1) {
346 event_add_fd(ctdb->ev, ctcp, ctcp->listen_fd, EVENT_FD_READ|EVENT_FD_AUTOCLOSE,
347 ctdb_listen_event, ctdb);
354 close(ctcp->listen_fd);
355 ctcp->listen_fd = -1;
361 listen on our own address
363 int ctdb_tcp_listen(struct ctdb_context *ctdb)
365 struct ctdb_tcp *ctcp = talloc_get_type(ctdb->private_data,
370 /* we can either auto-bind to the first available address, or we can
371 use a specified address */
372 if (!ctdb->address.address) {
373 return ctdb_tcp_listen_automatic(ctdb);
377 #ifdef HAVE_SOCK_SIN_LEN
378 sock.ip.sin_len = sizeof(sock);
380 if (ctdb_tcp_get_address(ctdb, ctdb->address.address,
385 switch (sock.sa.sa_family) {
387 sock.ip.sin_port = htons(ctdb->address.port);
390 sock.ip6.sin6_port = htons(ctdb->address.port);
393 DEBUG(DEBUG_ERR, (__location__ " unknown family %u\n",
398 ctcp->listen_fd = socket(sock.sa.sa_family, SOCK_STREAM, IPPROTO_TCP);
399 if (ctcp->listen_fd == -1) {
400 ctdb_set_error(ctdb, "socket failed\n");
404 set_close_on_exec(ctcp->listen_fd);
406 setsockopt(ctcp->listen_fd,SOL_SOCKET,SO_REUSEADDR,(char *)&one,sizeof(one));
408 if (bind(ctcp->listen_fd, (struct sockaddr * )&sock, sizeof(sock)) != 0) {
412 if (listen(ctcp->listen_fd, 10) == -1) {
416 event_add_fd(ctdb->ev, ctcp, ctcp->listen_fd, EVENT_FD_READ|EVENT_FD_AUTOCLOSE,
417 ctdb_listen_event, ctdb);
422 if (ctcp->listen_fd != -1) {
423 close(ctcp->listen_fd);
425 ctcp->listen_fd = -1;