Merge branches 'hammer2' and 'master' of ssh://crater.dragonflybsd.org/repository...
[dragonfly.git] / sbin / hammer2 / cmd_service.c
1 /*
2  * Copyright (c) 2011-2012 The DragonFly Project.  All rights reserved.
3  *
4  * This code is derived from software contributed to The DragonFly Project
5  * by Matthew Dillon <dillon@dragonflybsd.org>
6  * by Venkatesh Srinivas <vsrinivas@dragonflybsd.org>
7  *
8  * Redistribution and use in source and binary forms, with or without
9  * modification, are permitted provided that the following conditions
10  * are met:
11  *
12  * 1. Redistributions of source code must retain the above copyright
13  *    notice, this list of conditions and the following disclaimer.
14  * 2. Redistributions in binary form must reproduce the above copyright
15  *    notice, this list of conditions and the following disclaimer in
16  *    the documentation and/or other materials provided with the
17  *    distribution.
18  * 3. Neither the name of The DragonFly Project nor the names of its
19  *    contributors may be used to endorse or promote products derived
20  *    from this software without specific, prior written permission.
21  *
22  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
23  * ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
24  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
25  * FOR A PARTICULAR PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE
26  * COPYRIGHT HOLDERS OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
27  * INCIDENTAL, SPECIAL, EXEMPLARY OR CONSEQUENTIAL DAMAGES (INCLUDING,
28  * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
29  * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
30  * AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
31  * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
32  * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
33  * SUCH DAMAGE.
34  */
35
36 #include "hammer2.h"
37
38 static void *master_accept(void *data);
39 static void master_auth_signal(hammer2_router_t *router);
40 static void master_auth_rxmsg(hammer2_msg_t *msg);
41 static void master_link_signal(hammer2_router_t *router);
42 static void master_link_rxmsg(hammer2_msg_t *msg);
43 static void master_reconnect(const char *mntpt);
44
45 /*
46  * Start-up the master listener daemon for the machine.
47  *
48  * The master listener serves as a rendezvous point in the cluster, accepting
49  * connections, performing registrations and authentications, maintaining
50  * the spanning tree, and keeping track of message state so disconnects can
51  * be handled properly.
52  *
53  * Once authenticated only low-level messaging protocols (which includes
54  * tracking persistent messages) are handled by this daemon.  This daemon
55  * does not run the higher level quorum or locking protocols.
56  *
57  * This daemon can also be told to maintain connections to other nodes,
58  * forming a messaging backbone, which in turn allows PFS's (if desired) to
59  * simply connect to the master daemon via localhost if desired.
60  * Backbones are specified via /etc/hammer2.conf.
61  */
62 int
63 cmd_service(void)
64 {
65         struct sockaddr_in lsin;
66         int on;
67         int lfd;
68
69         /*
70          * Acquire socket and set options
71          */
72         if ((lfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
73                 fprintf(stderr, "master_listen: socket(): %s\n",
74                         strerror(errno));
75                 return 1;
76         }
77         on = 1;
78         setsockopt(lfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
79
80         /*
81          * Setup listen port and try to bind.  If the bind fails we assume
82          * that a master listener process is already running and silently
83          * fail.
84          */
85         bzero(&lsin, sizeof(lsin));
86         lsin.sin_family = AF_INET;
87         lsin.sin_addr.s_addr = INADDR_ANY;
88         lsin.sin_port = htons(HAMMER2_LISTEN_PORT);
89         if (bind(lfd, (struct sockaddr *)&lsin, sizeof(lsin)) < 0) {
90                 close(lfd);
91                 if (QuietOpt == 0) {
92                         fprintf(stderr,
93                                 "master listen: daemon already running\n");
94                 }
95                 return 0;
96         }
97         if (QuietOpt == 0)
98                 fprintf(stderr, "master listen: startup\n");
99         listen(lfd, 50);
100
101         /*
102          * Fork and disconnect the controlling terminal and parent process,
103          * executing the specified function as a pthread.
104          *
105          * Returns to the original process which can then continue running.
106          * In debug mode this call will create the pthread without forking
107          * and set NormalExit to 0, instead of fork.
108          */
109         hammer2_demon(master_accept, (void *)(intptr_t)lfd);
110         if (NormalExit)
111                 close(lfd);
112         return 0;
113 }
114
115 /*
116  * Master listen/accept thread.  Accept connections on the master socket,
117  * starting a pthread for each one.
118  */
119 static
120 void *
121 master_accept(void *data)
122 {
123         struct sockaddr_in asin;
124         socklen_t alen;
125         pthread_t thread;
126         hammer2_master_service_info_t *info;
127         int lfd = (int)(intptr_t)data;
128         int fd;
129         int i;
130         int count;
131         struct statfs *mntbuf = NULL;
132         struct statvfs *mntvbuf = NULL;
133
134         /*
135          * Nobody waits for us
136          */
137         setproctitle("hammer2 master listen");
138         pthread_detach(pthread_self());
139
140         /*
141          * Scan existing hammer2 mounts and reconnect to them using
142          * HAMMER2IOC_RECLUSTER.
143          */
144         count = getmntvinfo(&mntbuf, &mntvbuf, MNT_NOWAIT);
145         for (i = 0; i < count; ++i) {
146                 if (strcmp(mntbuf[i].f_fstypename, "hammer2") == 0)
147                         master_reconnect(mntbuf[i].f_mntonname);
148         }
149
150         /*
151          * Accept connections and create pthreads to handle them after
152          * validating the IP.
153          */
154         for (;;) {
155                 alen = sizeof(asin);
156                 fd = accept(lfd, (struct sockaddr *)&asin, &alen);
157                 if (fd < 0) {
158                         if (errno == EINTR)
159                                 continue;
160                         break;
161                 }
162                 thread = NULL;
163                 fprintf(stderr, "master_accept: accept fd %d\n", fd);
164                 info = malloc(sizeof(*info));
165                 bzero(info, sizeof(*info));
166                 info->fd = fd;
167                 info->detachme = 1;
168                 pthread_create(&thread, NULL, master_service, info);
169         }
170         return (NULL);
171 }
172
173 /*
174  * Normally the mount program supplies a cluster communications
175  * descriptor to the hammer2 vfs on mount, but if you kill the service
176  * daemon and restart it that link will be lost.
177  *
178  * This procedure attempts to [re]connect to existing mounts when
179  * the service daemon is started up before going into its accept
180  * loop.
181  */
182 static
183 void
184 master_reconnect(const char *mntpt)
185 {
186         struct hammer2_ioc_recluster recls;
187         hammer2_master_service_info_t *info;
188         pthread_t thread;
189         int fd;
190         int pipefds[2];
191
192         fd = open(mntpt, O_RDONLY);
193         if (fd < 0) {
194                 fprintf(stderr, "reconnect %s: no access to mount\n", mntpt);
195                 return;
196         }
197         if (pipe(pipefds) < 0) {
198                 fprintf(stderr, "reconnect %s: pipe() failed\n", mntpt);
199                 return;
200         }
201         bzero(&recls, sizeof(recls));
202         recls.fd = pipefds[0];
203         if (ioctl(fd, HAMMER2IOC_RECLUSTER, &recls) < 0) {
204                 fprintf(stderr, "reconnect %s: ioctl failed\n", mntpt);
205                 close(pipefds[0]);
206                 close(pipefds[1]);
207                 close(fd);
208                 return;
209         }
210         close(pipefds[0]);
211
212         info = malloc(sizeof(*info));
213         bzero(info, sizeof(*info));
214         info->fd = pipefds[1];
215         info->detachme = 1;
216         pthread_create(&thread, NULL, master_service, info);
217 }
218
219 /*
220  * Service an accepted connection (runs as a pthread)
221  *
222  * (also called from a couple of other places)
223  */
224 void *
225 master_service(void *data)
226 {
227         hammer2_master_service_info_t *info = data;
228         hammer2_iocom_t iocom;
229
230         if (info->detachme)
231                 pthread_detach(pthread_self());
232
233         hammer2_iocom_init(&iocom, info->fd, -1,
234                            master_auth_signal,
235                            master_auth_rxmsg,
236                            NULL);
237         hammer2_iocom_core(&iocom);
238
239         fprintf(stderr,
240                 "iocom on fd %d terminated error rx=%d, tx=%d\n",
241                 info->fd, iocom.ioq_rx.error, iocom.ioq_tx.error);
242         close(info->fd);
243         info->fd = -1;  /* safety */
244         free(info);
245
246         return (NULL);
247 }
248
249 /************************************************************************
250  *                          AUTHENTICATION                              *
251  ************************************************************************
252  *
253  * Callback via hammer2_iocom_core().
254  *
255  * Additional messaging-based authentication must occur before normal
256  * message operation.  The connection has already been encrypted at
257  * this point.
258  */
259 static void master_auth_conn_rx(hammer2_msg_t *msg);
260
261 static
262 void
263 master_auth_signal(hammer2_router_t *router)
264 {
265         hammer2_msg_t *msg;
266
267         /*
268          * Transmit LNK_CONN, enabling the SPAN protocol if both sides
269          * agree.
270          *
271          * XXX put additional authentication states here
272          */
273         msg = hammer2_msg_alloc(router, 0, HAMMER2_LNK_CONN |
274                                                    HAMMER2_MSGF_CREATE,
275                                 master_auth_conn_rx, NULL);
276         snprintf(msg->any.lnk_conn.label, sizeof(msg->any.lnk_conn.label), "*");
277         hammer2_msg_write(msg);
278
279         hammer2_router_restate(router,
280                               master_link_signal,
281                               master_link_rxmsg,
282                               NULL);
283 }
284
285 static
286 void
287 master_auth_conn_rx(hammer2_msg_t *msg)
288 {
289         if (msg->any.head.cmd & HAMMER2_MSGF_DELETE)
290                 hammer2_msg_reply(msg, 0);
291 }
292
293 static
294 void
295 master_auth_rxmsg(hammer2_msg_t *msg __unused)
296 {
297 }
298
299 /************************************************************************
300  *                      POST-AUTHENTICATION SERVICE MSGS                *
301  ************************************************************************
302  *
303  * Callback via hammer2_iocom_core().
304  */
305 static
306 void
307 master_link_signal(hammer2_router_t *router)
308 {
309         hammer2_msg_lnk_signal(router);
310 }
311
312 static
313 void
314 master_link_rxmsg(hammer2_msg_t *msg)
315 {
316         hammer2_state_t *state;
317         uint32_t cmd;
318
319         /*
320          * If the message state has a function established we just
321          * call the function, otherwise we call the appropriate
322          * link-level protocol related to the original command and
323          * let it sort it out.
324          *
325          * Non-transactional one-off messages, on the otherhand,
326          * might have REPLY set.
327          */
328         state = msg->state;
329         cmd = state ? state->msg->any.head.cmd : msg->any.head.cmd;
330
331         fprintf(stderr, "service-receive: %s\n", hammer2_msg_str(msg));
332
333         if (state && state->func) {
334                 assert(state->func != NULL);
335                 state->func(msg);
336         } else {
337                 switch(cmd & HAMMER2_MSGF_PROTOS) {
338                 case HAMMER2_MSG_PROTO_LNK:
339                         hammer2_msg_lnk(msg);
340                         break;
341                 case HAMMER2_MSG_PROTO_DBG:
342                         hammer2_msg_dbg(msg);
343                         break;
344                 default:
345                         hammer2_msg_reply(msg, HAMMER2_MSG_ERR_NOSUPP);
346                         break;
347                 }
348         }
349 }