2819b052f414c7b400e208e5bfcd03ed6a8a0b26
[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  * NOTE: A hammer2 mount point can only accomodate one connection at a time
183  *       so this will disconnect any existing connection during the
184  *       reconnect.
185  */
186 static
187 void
188 master_reconnect(const char *mntpt)
189 {
190         struct hammer2_ioc_recluster recls;
191         hammer2_master_service_info_t *info;
192         pthread_t thread;
193         int fd;
194         int pipefds[2];
195
196         fd = open(mntpt, O_RDONLY);
197         if (fd < 0) {
198                 fprintf(stderr, "reconnect %s: no access to mount\n", mntpt);
199                 return;
200         }
201         if (pipe(pipefds) < 0) {
202                 fprintf(stderr, "reconnect %s: pipe() failed\n", mntpt);
203                 return;
204         }
205         bzero(&recls, sizeof(recls));
206         recls.fd = pipefds[0];
207         if (ioctl(fd, HAMMER2IOC_RECLUSTER, &recls) < 0) {
208                 fprintf(stderr, "reconnect %s: ioctl failed\n", mntpt);
209                 close(pipefds[0]);
210                 close(pipefds[1]);
211                 close(fd);
212                 return;
213         }
214         close(pipefds[0]);
215
216         info = malloc(sizeof(*info));
217         bzero(info, sizeof(*info));
218         info->fd = pipefds[1];
219         info->detachme = 1;
220         pthread_create(&thread, NULL, master_service, info);
221 }
222
223 /*
224  * Service an accepted connection (runs as a pthread)
225  *
226  * (also called from a couple of other places)
227  */
228 void *
229 master_service(void *data)
230 {
231         hammer2_master_service_info_t *info = data;
232         hammer2_iocom_t iocom;
233
234         if (info->detachme)
235                 pthread_detach(pthread_self());
236
237         hammer2_iocom_init(&iocom, info->fd, -1,
238                            master_auth_signal,
239                            master_auth_rxmsg,
240                            NULL);
241         hammer2_iocom_core(&iocom);
242
243         fprintf(stderr,
244                 "iocom on fd %d terminated error rx=%d, tx=%d\n",
245                 info->fd, iocom.ioq_rx.error, iocom.ioq_tx.error);
246         close(info->fd);
247         info->fd = -1;  /* safety */
248         free(info);
249
250         return (NULL);
251 }
252
253 /************************************************************************
254  *                          AUTHENTICATION                              *
255  ************************************************************************
256  *
257  * Callback via hammer2_iocom_core().
258  *
259  * Additional messaging-based authentication must occur before normal
260  * message operation.  The connection has already been encrypted at
261  * this point.
262  */
263 static void master_auth_conn_rx(hammer2_msg_t *msg);
264
265 static
266 void
267 master_auth_signal(hammer2_router_t *router)
268 {
269         hammer2_msg_t *msg;
270
271         /*
272          * Transmit LNK_CONN, enabling the SPAN protocol if both sides
273          * agree.
274          *
275          * XXX put additional authentication states here
276          */
277         msg = hammer2_msg_alloc(router, 0, HAMMER2_LNK_CONN |
278                                                    HAMMER2_MSGF_CREATE,
279                                 master_auth_conn_rx, NULL);
280         snprintf(msg->any.lnk_conn.label, sizeof(msg->any.lnk_conn.label), "*");
281         hammer2_msg_write(msg);
282
283         hammer2_router_restate(router,
284                               master_link_signal,
285                               master_link_rxmsg,
286                               NULL);
287 }
288
289 static
290 void
291 master_auth_conn_rx(hammer2_msg_t *msg)
292 {
293         if (msg->any.head.cmd & HAMMER2_MSGF_DELETE)
294                 hammer2_msg_reply(msg, 0);
295 }
296
297 static
298 void
299 master_auth_rxmsg(hammer2_msg_t *msg __unused)
300 {
301 }
302
303 /************************************************************************
304  *                      POST-AUTHENTICATION SERVICE MSGS                *
305  ************************************************************************
306  *
307  * Callback via hammer2_iocom_core().
308  */
309 static
310 void
311 master_link_signal(hammer2_router_t *router)
312 {
313         hammer2_msg_lnk_signal(router);
314 }
315
316 static
317 void
318 master_link_rxmsg(hammer2_msg_t *msg)
319 {
320         hammer2_state_t *state;
321         uint32_t cmd;
322
323         /*
324          * If the message state has a function established we just
325          * call the function, otherwise we call the appropriate
326          * link-level protocol related to the original command and
327          * let it sort it out.
328          *
329          * Non-transactional one-off messages, on the otherhand,
330          * might have REPLY set.
331          */
332         state = msg->state;
333         cmd = state ? state->msg->any.head.cmd : msg->any.head.cmd;
334
335         fprintf(stderr, "service-receive: %s\n", hammer2_msg_str(msg));
336
337         if (state && state->func) {
338                 assert(state->func != NULL);
339                 state->func(msg);
340         } else {
341                 switch(cmd & HAMMER2_MSGF_PROTOS) {
342                 case HAMMER2_MSG_PROTO_LNK:
343                         hammer2_msg_lnk(msg);
344                         break;
345                 case HAMMER2_MSG_PROTO_DBG:
346                         hammer2_msg_dbg(msg);
347                         break;
348                 default:
349                         hammer2_msg_reply(msg, HAMMER2_MSG_ERR_NOSUPP);
350                         break;
351                 }
352         }
353 }