merge from tridge
[sahlberg/ctdb.git] / tcp / tcp_connect.c
1 /* 
2    ctdb over TCP
3
4    Copyright (C) Andrew Tridgell  2006
5
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.
10    
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.
15    
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/>.
18 */
19
20 #include "includes.h"
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"
26 #include "ctdb_tcp.h"
27
28 /*
29   stop any connecting (established or pending) to a node
30  */
31 void ctdb_tcp_stop_connection(struct ctdb_node *node)
32 {
33         struct ctdb_tcp_node *tnode = talloc_get_type(
34                 node->private_data, struct ctdb_tcp_node);
35         
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) {
42                 close(tnode->fd);
43                 tnode->fd = -1;
44         }
45 }
46
47
48 /*
49   called when a complete packet has come in - should not happen on this socket
50  */
51 void ctdb_tcp_tnode_cb(uint8_t *data, size_t cnt, void *private_data)
52 {
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);
56
57         if (data == NULL) {
58                 node->ctdb->upcalls->node_dead(node);
59         }
60
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);
64 }
65
66 /*
67   called when socket becomes writeable on connect
68 */
69 static void ctdb_node_connect_write(struct event_context *ev, struct fd_event *fde, 
70                                     uint16_t flags, void *private_data)
71 {
72         struct ctdb_node *node = talloc_get_type(private_data,
73                                                  struct ctdb_node);
74         struct ctdb_tcp_node *tnode = talloc_get_type(node->private_data,
75                                                       struct ctdb_tcp_node);
76         struct ctdb_context *ctdb = node->ctdb;
77         int error = 0;
78         socklen_t len = sizeof(error);
79         int one = 1;
80
81         talloc_free(tnode->connect_te);
82         tnode->connect_te = NULL;
83
84         if (getsockopt(tnode->fd, SOL_SOCKET, SO_ERROR, &error, &len) != 0 ||
85             error != 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);
90                 return;
91         }
92
93         talloc_free(tnode->connect_fde);
94         tnode->connect_fde = NULL;
95
96         setsockopt(tnode->fd,IPPROTO_TCP,TCP_NODELAY,(char *)&one,sizeof(one));
97         setsockopt(tnode->fd,SOL_SOCKET,SO_KEEPALIVE,(char *)&one,sizeof(one));
98
99         ctdb_queue_set_fd(tnode->out_queue, tnode->fd);
100
101         /* the queue subsystem now owns this fd */
102         tnode->fd = -1;
103        
104         /* tell the ctdb layer we are connected */
105         node->ctdb->upcalls->node_connected(node);
106 }
107
108
109 static int ctdb_tcp_get_address(struct ctdb_context *ctdb,
110                                 const char *address, struct in_addr *addr)
111 {
112         if (inet_pton(AF_INET, address, addr) <= 0) {
113                 struct hostent *he = gethostbyname(address);
114                 if (he == NULL || he->h_length > sizeof(*addr)) {
115                         ctdb_set_error(ctdb, "invalid nework address '%s'\n", 
116                                        address);
117                         return -1;
118                 }
119                 memcpy(addr, he->h_addr, he->h_length);
120         }
121         return 0;
122 }
123
124 /*
125   called when we should try and establish a tcp connection to a node
126 */
127 void ctdb_tcp_node_connect(struct event_context *ev, struct timed_event *te, 
128                            struct timeval t, void *private_data)
129 {
130         struct ctdb_node *node = talloc_get_type(private_data,
131                                                  struct ctdb_node);
132         struct ctdb_tcp_node *tnode = talloc_get_type(node->private_data, 
133                                                       struct ctdb_tcp_node);
134         struct ctdb_context *ctdb = node->ctdb;
135         struct sockaddr_in sock_in;
136         struct sockaddr_in sock_out;
137
138         ctdb_tcp_stop_connection(node);
139
140         tnode->fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
141
142         set_nonblocking(tnode->fd);
143         set_close_on_exec(tnode->fd);
144
145         ZERO_STRUCT(sock_out);
146 #ifdef HAVE_SOCK_SIN_LEN
147         sock_out.sin_len = sizeof(sock_out);
148 #endif
149         if (ctdb_tcp_get_address(ctdb, node->address.address, &sock_out.sin_addr) != 0) {
150                 return;
151         }
152         sock_out.sin_port = htons(node->address.port);
153         sock_out.sin_family = PF_INET;
154
155
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.
161          */
162         ZERO_STRUCT(sock_in);
163 #ifdef HAVE_SOCK_SIN_LEN
164         sock_in.sin_len = sizeof(sock_in);
165 #endif
166         if (ctdb_tcp_get_address(ctdb, ctdb->address.address, &sock_in.sin_addr) != 0) {
167                 return;
168         }
169         sock_in.sin_port = htons(0); /* INPORT_ANY is not always available */
170         sock_in.sin_family = PF_INET;
171         bind(tnode->fd, (struct sockaddr *)&sock_in, sizeof(sock_in));
172
173         if (connect(tnode->fd, (struct sockaddr *)&sock_out, sizeof(sock_out)) != 0 &&
174             errno != EINPROGRESS) {
175                 ctdb_tcp_stop_connection(node);
176                 tnode->connect_te = event_add_timed(ctdb->ev, tnode, 
177                                                     timeval_current_ofs(1, 0),
178                                                     ctdb_tcp_node_connect, node);
179                 return;
180         }
181
182         /* non-blocking connect - wait for write event */
183         tnode->connect_fde = event_add_fd(node->ctdb->ev, tnode, tnode->fd, 
184                                           EVENT_FD_WRITE|EVENT_FD_READ, 
185                                           ctdb_node_connect_write, node);
186
187         /* don't give it long to connect - retry in one second. This ensures
188            that we find a node is up quickly (tcp normally backs off a syn reply
189            delay by quite a lot) */
190         tnode->connect_te = event_add_timed(ctdb->ev, tnode, timeval_current_ofs(1, 0), 
191                                             ctdb_tcp_node_connect, node);
192 }
193
194 /*
195   called when we get contacted by another node
196   currently makes no attempt to check if the connection is really from a ctdb
197   node in our cluster
198 */
199 static void ctdb_listen_event(struct event_context *ev, struct fd_event *fde, 
200                               uint16_t flags, void *private_data)
201 {
202         struct ctdb_context *ctdb = talloc_get_type(private_data, struct ctdb_context);
203         struct ctdb_tcp *ctcp = talloc_get_type(ctdb->private_data, struct ctdb_tcp);
204         struct sockaddr_in addr;
205         socklen_t len;
206         int fd, nodeid;
207         struct ctdb_incoming *in;
208         int one = 1;
209         const char *incoming_node;
210
211         memset(&addr, 0, sizeof(addr));
212         len = sizeof(addr);
213         fd = accept(ctcp->listen_fd, (struct sockaddr *)&addr, &len);
214         if (fd == -1) return;
215
216         incoming_node = inet_ntoa(addr.sin_addr);
217         for (nodeid=0;nodeid<ctdb->num_nodes;nodeid++) {
218                 if (!strcmp(incoming_node, ctdb->nodes[nodeid]->address.address)) {
219                         DEBUG(0, ("Incoming connection from node:%d %s\n",nodeid,incoming_node));
220                         break;
221                 }
222         }
223         if (nodeid>=ctdb->num_nodes) {
224                 DEBUG(0, ("Refused connection from unknown node %s\n", incoming_node));
225                 close(fd);
226                 return;
227         }
228
229         in = talloc_zero(ctcp, struct ctdb_incoming);
230         in->fd = fd;
231         in->ctdb = ctdb;
232
233         set_nonblocking(in->fd);
234         set_close_on_exec(in->fd);
235
236         setsockopt(in->fd,SOL_SOCKET,SO_KEEPALIVE,(char *)&one,sizeof(one));
237
238         in->queue = ctdb_queue_setup(ctdb, in, in->fd, CTDB_TCP_ALIGNMENT, 
239                                      ctdb_tcp_read_cb, in);
240 }
241
242
243 /*
244   automatically find which address to listen on
245 */
246 static int ctdb_tcp_listen_automatic(struct ctdb_context *ctdb)
247 {
248         struct ctdb_tcp *ctcp = talloc_get_type(ctdb->private_data,
249                                                 struct ctdb_tcp);
250         struct sockaddr_in sock;
251         int lock_fd, i;
252         const char *lock_path = "/tmp/.ctdb_socket_lock";
253         struct flock lock;
254
255         /* in order to ensure that we don't get two nodes with the
256            same adddress, we must make the bind() and listen() calls
257            atomic. The SO_REUSEADDR setsockopt only prevents double
258            binds if the first socket is in LISTEN state  */
259         lock_fd = open(lock_path, O_RDWR|O_CREAT, 0666);
260         if (lock_fd == -1) {
261                 DEBUG(0,("Unable to open %s\n", lock_path));
262                 return -1;
263         }
264
265         lock.l_type = F_WRLCK;
266         lock.l_whence = SEEK_SET;
267         lock.l_start = 0;
268         lock.l_len = 1;
269         lock.l_pid = 0;
270
271         if (fcntl(lock_fd, F_SETLKW, &lock) != 0) {
272                 DEBUG(0,("Unable to lock %s\n", lock_path));
273                 close(lock_fd);
274                 return -1;
275         }
276
277         for (i=0;i<ctdb->num_nodes;i++) {
278                 ZERO_STRUCT(sock);
279 #ifdef HAVE_SOCK_SIN_LEN
280                 sock.sin_len = sizeof(sock);
281 #endif
282                 sock.sin_port = htons(ctdb->nodes[i]->address.port);
283                 sock.sin_family = PF_INET;
284                 if (ctdb_tcp_get_address(ctdb, ctdb->nodes[i]->address.address, 
285                                          &sock.sin_addr) != 0) {
286                         continue;
287                 }
288                 
289                 if (bind(ctcp->listen_fd, (struct sockaddr * )&sock, 
290                          sizeof(sock)) == 0) {
291                         break;
292                 }
293         }
294         
295         if (i == ctdb->num_nodes) {
296                 DEBUG(0,("Unable to bind to any of the node addresses - giving up\n"));
297                 goto failed;
298         }
299         ctdb->address = ctdb->nodes[i]->address;
300         ctdb->name = talloc_asprintf(ctdb, "%s:%u", 
301                                      ctdb->address.address, 
302                                      ctdb->address.port);
303         ctdb->pnn = ctdb->nodes[i]->pnn;
304         ctdb->nodes[i]->flags &= ~NODE_FLAGS_DISCONNECTED;
305         DEBUG(1,("ctdb chose network address %s:%u pnn %u\n", 
306                  ctdb->address.address, 
307                  ctdb->address.port, 
308                  ctdb->pnn));
309
310         if (listen(ctcp->listen_fd, 10) == -1) {
311                 goto failed;
312         }
313
314         event_add_fd(ctdb->ev, ctcp, ctcp->listen_fd, EVENT_FD_READ|EVENT_FD_AUTOCLOSE, 
315                      ctdb_listen_event, ctdb);  
316
317         close(lock_fd);
318         return 0;
319         
320 failed:
321         close(lock_fd);
322         close(ctcp->listen_fd);
323         ctcp->listen_fd = -1;
324         return -1;
325 }
326
327
328 /*
329   listen on our own address
330 */
331 int ctdb_tcp_listen(struct ctdb_context *ctdb)
332 {
333         struct ctdb_tcp *ctcp = talloc_get_type(ctdb->private_data,
334                                                 struct ctdb_tcp);
335         struct sockaddr_in sock;
336         int one = 1;
337
338         ctcp->listen_fd = socket(PF_INET, SOCK_STREAM, IPPROTO_TCP);
339         if (ctcp->listen_fd == -1) {
340                 ctdb_set_error(ctdb, "socket failed\n");
341                 return -1;
342         }
343
344         set_close_on_exec(ctcp->listen_fd);
345
346         setsockopt(ctcp->listen_fd,SOL_SOCKET,SO_REUSEADDR,(char *)&one,sizeof(one));
347
348         /* we can either auto-bind to the first available address, or we can
349            use a specified address */
350         if (!ctdb->address.address) {
351                 return ctdb_tcp_listen_automatic(ctdb);
352         }
353
354         ZERO_STRUCT(sock);
355 #ifdef HAVE_SOCK_SIN_LEN
356         sock.sin_len = sizeof(sock);
357 #endif
358         sock.sin_port = htons(ctdb->address.port);
359         sock.sin_family = PF_INET;
360         
361         if (ctdb_tcp_get_address(ctdb, ctdb->address.address, 
362                                  &sock.sin_addr) != 0) {
363                 goto failed;
364         }
365         
366         if (bind(ctcp->listen_fd, (struct sockaddr * )&sock, sizeof(sock)) != 0) {
367                 goto failed;
368         }
369
370         if (listen(ctcp->listen_fd, 10) == -1) {
371                 goto failed;
372         }
373
374         event_add_fd(ctdb->ev, ctcp, ctcp->listen_fd, EVENT_FD_READ|EVENT_FD_AUTOCLOSE, 
375                      ctdb_listen_event, ctdb);  
376
377         return 0;
378
379 failed:
380         close(ctcp->listen_fd);
381         ctcp->listen_fd = -1;
382         return -1;
383 }
384