be35ca792540936b3ad4982cc00792420ad90aa3
[openafs.git] / src / vol / daemon_com.c
1 /*
2  * Copyright 2006-2008, Sine Nomine Associates and others.
3  * All Rights Reserved.
4  *
5  * This software has been released under the terms of the IBM Public
6  * License.  For details, see the LICENSE file in the top-level source
7  * directory or online at http://www.openafs.org/dl/license10.html
8  */
9
10 /*
11  * localhost interprocess communication for servers
12  *
13  * currently handled by a localhost socket
14  * (yes, this needs to be replaced someday)
15  */
16
17 #ifndef _WIN32
18 #define FD_SETSIZE 65536
19 #endif
20
21 #include <afsconfig.h>
22 #include <afs/param.h>
23
24 #include <roken.h>
25 #include <afs/opr.h>
26
27 #include <rx/xdr.h>
28 #include <afs/afsint.h>
29 #include <afs/errors.h>
30
31 #include "nfs.h"
32 #include "daemon_com.h"
33 #include "lwp.h"
34 #include "lock.h"
35 #include <afs/afssyscalls.h>
36 #include "ihandle.h"
37 #include "vnode.h"
38 #include "volume.h"
39 #include "partition.h"
40 #include "common.h"
41 #include <rx/rx_queue.h>
42
43 #ifdef USE_UNIX_SOCKETS
44 #include <afs/afsutil.h>
45 #include <sys/un.h>
46 #endif
47
48 int (*V_BreakVolumeCallbacks) (VolumeId);
49
50 #define MAXHANDLERS     4       /* Up to 4 clients; must be at least 2, so that
51                                  * move = dump+restore can run on single server */
52
53 #define MAX_BIND_TRIES  5       /* Number of times to retry socket bind */
54
55 static int SYNC_ask_internal(SYNC_client_state * state, SYNC_command * com, SYNC_response * res);
56
57
58 /*
59  * On AIX, connect() and bind() require use of SUN_LEN() macro;
60  * sizeof(struct sockaddr_un) will not suffice.
61  */
62 #if defined(AFS_AIX_ENV) && defined(USE_UNIX_SOCKETS)
63 #define AFS_SOCKADDR_LEN(sa)  SUN_LEN(sa)
64 #else
65 #define AFS_SOCKADDR_LEN(sa)  sizeof(*sa)
66 #endif
67
68
69 /* daemon com SYNC general interfaces */
70
71 /**
72  * fill in sockaddr structure.
73  *
74  * @param[in]  endpoint pointer to sync endpoint object
75  * @param[out] addr     pointer to sockaddr structure
76  *
77  * @post sockaddr structure populated using information from
78  *       endpoint structure.
79  */
80 void
81 SYNC_getAddr(SYNC_endpoint_t * endpoint, SYNC_sockaddr_t * addr)
82 {
83 #ifdef USE_UNIX_SOCKETS
84     char tbuffer[AFSDIR_PATH_MAX];
85 #endif /* USE_UNIX_SOCKETS */
86
87     memset(addr, 0, sizeof(*addr));
88
89 #ifdef USE_UNIX_SOCKETS
90     strcompose(tbuffer, AFSDIR_PATH_MAX, AFSDIR_SERVER_LOCAL_DIRPATH, "/",
91                endpoint->un, NULL);
92     addr->sun_family = AF_UNIX;
93     strncpy(addr->sun_path, tbuffer, (sizeof(struct sockaddr_un) - sizeof(short)));
94 #else  /* !USE_UNIX_SOCKETS */
95 #ifdef STRUCT_SOCKADDR_HAS_SA_LEN
96     addr->sin_len = sizeof(struct sockaddr_in);
97 #endif
98     addr->sin_addr.s_addr = htonl(0x7f000001);
99     addr->sin_family = AF_INET; /* was localhost->h_addrtype */
100     addr->sin_port = htons(endpoint->in);       /* XXXX htons not _really_ neccessary */
101 #endif /* !USE_UNIX_SOCKETS */
102 }
103
104 /**
105  * get a socket descriptor of the appropriate domain.
106  *
107  * @param[in]  endpoint pointer to sync endpoint object
108  *
109  * @return socket descriptor
110  *
111  * @post socket of domain specified in endpoint structure is created and
112  *       returned to caller.
113  */
114 osi_socket
115 SYNC_getSock(SYNC_endpoint_t * endpoint)
116 {
117     osi_socket sd;
118     osi_Assert((sd = socket(endpoint->domain, SOCK_STREAM, 0)) >= 0);
119     return sd;
120 }
121
122 /* daemon com SYNC client interface */
123
124 /**
125  * open a client connection to a sync server
126  *
127  * @param[in] state  pointer to sync client handle
128  *
129  * @return operation status
130  *    @retval 1 success
131  *
132  * @note at present, this routine aborts rather than returning an error code
133  */
134 int
135 SYNC_connect(SYNC_client_state * state)
136 {
137     SYNC_sockaddr_t addr;
138     /* I can't believe the following is needed for localhost connections!! */
139     static time_t backoff[] =
140         { 3, 3, 3, 5, 5, 5, 7, 15, 16, 24, 32, 40, 48, 0 };
141     time_t *timeout = &backoff[0];
142
143     if (state->fd != OSI_NULLSOCKET) {
144         return 1;
145     }
146
147     SYNC_getAddr(&state->endpoint, &addr);
148
149     for (;;) {
150         state->fd = SYNC_getSock(&state->endpoint);
151         if (connect(state->fd, (struct sockaddr *)&addr, AFS_SOCKADDR_LEN(&addr)) >= 0)
152             return 1;
153         if (!*timeout)
154             break;
155         if (!(*timeout & 1))
156             Log("SYNC_connect: temporary failure on circuit '%s' (will retry)\n",
157                 state->proto_name);
158         SYNC_disconnect(state);
159         sleep(*timeout++);
160     }
161     perror("SYNC_connect failed (giving up!)");
162     return 0;
163 }
164
165 /**
166  * forcibly disconnect a sync client handle.
167  *
168  * @param[in] state  pointer to sync client handle
169  *
170  * @retval operation status
171  *    @retval 0 success
172  */
173 int
174 SYNC_disconnect(SYNC_client_state * state)
175 {
176     rk_closesocket(state->fd);
177     state->fd = OSI_NULLSOCKET;
178     return 0;
179 }
180
181 /**
182  * gracefully disconnect a sync client handle.
183  *
184  * @param[in] state  pointer to sync client handle
185  *
186  * @return operation status
187  *    @retval SYNC_OK success
188  */
189 afs_int32
190 SYNC_closeChannel(SYNC_client_state * state)
191 {
192     SYNC_command com;
193     SYNC_response res;
194     SYNC_PROTO_BUF_DECL(ores);
195
196     if (state->fd == OSI_NULLSOCKET)
197         return SYNC_OK;
198
199     memset(&com, 0, sizeof(com));
200     memset(&res, 0, sizeof(res));
201
202     res.payload.len = SYNC_PROTO_MAX_LEN;
203     res.payload.buf = ores;
204
205     com.hdr.command = SYNC_COM_CHANNEL_CLOSE;
206     com.hdr.command_len = sizeof(SYNC_command_hdr);
207     com.hdr.flags |= SYNC_FLAG_CHANNEL_SHUTDOWN;
208
209     /* in case the other end dropped, don't do any retries */
210     state->retry_limit = 0;
211     state->hard_timeout = 0;
212
213     SYNC_ask(state, &com, &res);
214     SYNC_disconnect(state);
215
216     return SYNC_OK;
217 }
218
219 /**
220  * forcibly break a client connection, and then create a new connection.
221  *
222  * @param[in] state  pointer to sync client handle
223  *
224  * @post old connection dropped; new connection established
225  *
226  * @return @see SYNC_connect()
227  */
228 int
229 SYNC_reconnect(SYNC_client_state * state)
230 {
231     SYNC_disconnect(state);
232     return SYNC_connect(state);
233 }
234
235 /**
236  * send a command to a sync server and wait for a response.
237  *
238  * @param[in]  state  pointer to sync client handle
239  * @param[in]  com    command object
240  * @param[out] res    response object
241  *
242  * @return operation status
243  *    @retval SYNC_OK success
244  *    @retval SYNC_COM_ERROR communications error
245  *    @retval SYNC_BAD_COMMAND server did not recognize command code
246  *
247  * @note this routine merely handles error processing; SYNC_ask_internal()
248  *       handles the low-level details of communicating with the SYNC server.
249  *
250  * @see SYNC_ask_internal
251  */
252 afs_int32
253 SYNC_ask(SYNC_client_state * state, SYNC_command * com, SYNC_response * res)
254 {
255     int tries;
256     afs_uint32 now, timeout, code=SYNC_OK;
257
258     if (state->fatal_error) {
259         return SYNC_COM_ERROR;
260     }
261
262     if (state->fd == OSI_NULLSOCKET) {
263         SYNC_connect(state);
264     }
265
266     if (state->fd == OSI_NULLSOCKET) {
267         state->fatal_error = 1;
268         return SYNC_COM_ERROR;
269     }
270
271 #ifdef AFS_DEMAND_ATTACH_FS
272     com->hdr.flags |= SYNC_FLAG_DAFS_EXTENSIONS;
273 #endif
274
275     now = FT_ApproxTime();
276     timeout = now + state->hard_timeout;
277     for (tries = 0;
278          (tries <= state->retry_limit) && (now <= timeout);
279          tries++, now = FT_ApproxTime()) {
280         code = SYNC_ask_internal(state, com, res);
281         if (code == SYNC_OK) {
282             break;
283         } else if (code == SYNC_BAD_COMMAND) {
284             Log("SYNC_ask: protocol mismatch on circuit '%s'; make sure "
285                 "fileserver, volserver, salvageserver and salvager are same "
286                 "version\n", state->proto_name);
287             break;
288         } else if ((code == SYNC_COM_ERROR) && (tries < state->retry_limit)) {
289             Log("SYNC_ask: protocol communications failure on circuit '%s'; "
290                 "attempting reconnect to server\n", state->proto_name);
291             SYNC_reconnect(state);
292             /* try again */
293         } else {
294             /*
295              * unknown (probably protocol-specific) response code, pass it up to
296              * the caller, and let them deal with it
297              */
298             break;
299         }
300     }
301
302     if (code == SYNC_COM_ERROR) {
303         Log("SYNC_ask: fatal protocol error on circuit '%s'; disabling sync "
304             "protocol until next server restart\n",
305             state->proto_name);
306         state->fatal_error = 1;
307     }
308
309     return code;
310 }
311
312 /**
313  * send a command to a sync server and wait for a response.
314  *
315  * @param[in]  state  pointer to sync client handle
316  * @param[in]  com    command object
317  * @param[out] res    response object
318  *
319  * @return operation status
320  *    @retval SYNC_OK success
321  *    @retval SYNC_COM_ERROR communications error
322  *
323  * @internal
324  */
325 static afs_int32
326 SYNC_ask_internal(SYNC_client_state * state, SYNC_command * com, SYNC_response * res)
327 {
328     int n;
329     SYNC_PROTO_BUF_DECL(buf);
330 #ifndef AFS_NT40_ENV
331     int iovcnt;
332     struct iovec iov[2];
333 #endif
334
335     if (state->fd == OSI_NULLSOCKET) {
336         Log("SYNC_ask:  invalid sync file descriptor on circuit '%s'\n",
337             state->proto_name);
338         res->hdr.response = SYNC_COM_ERROR;
339         goto done;
340     }
341
342     if (com->hdr.command_len > SYNC_PROTO_MAX_LEN) {
343         Log("SYNC_ask:  internal SYNC buffer too small on circuit '%s'; "
344             "please file a bug\n", state->proto_name);
345         res->hdr.response = SYNC_COM_ERROR;
346         goto done;
347     }
348
349     /*
350      * fill in some common header fields
351      */
352     com->hdr.proto_version = state->proto_version;
353     com->hdr.pkt_seq = ++state->pkt_seq;
354     com->hdr.com_seq = ++state->com_seq;
355 #ifdef AFS_NT40_ENV
356     com->hdr.pid = 0;
357     com->hdr.tid = 0;
358 #else
359     com->hdr.pid = getpid();
360 #ifdef AFS_PTHREAD_ENV
361     com->hdr.tid = afs_pointer_to_int(pthread_self());
362 #else
363     {
364         PROCESS handle = LWP_ThreadId();
365         com->hdr.tid = (handle) ? handle->index : 0;
366     }
367 #endif /* !AFS_PTHREAD_ENV */
368 #endif /* !AFS_NT40_ENV */
369
370     memcpy(buf, &com->hdr, sizeof(com->hdr));
371     if (com->payload.len) {
372         memcpy(buf + sizeof(com->hdr), com->payload.buf,
373                com->hdr.command_len - sizeof(com->hdr));
374     }
375
376 #ifdef AFS_NT40_ENV
377     n = send(state->fd, buf, com->hdr.command_len, 0);
378     if (n != com->hdr.command_len) {
379         Log("SYNC_ask:  write failed on circuit '%s'\n", state->proto_name);
380         res->hdr.response = SYNC_COM_ERROR;
381         goto done;
382     }
383
384     if (com->hdr.command == SYNC_COM_CHANNEL_CLOSE) {
385         /* short circuit close channel requests */
386         res->hdr.response = SYNC_OK;
387         goto done;
388     }
389
390     n = recv(state->fd, buf, SYNC_PROTO_MAX_LEN, 0);
391     if (n == 0 || (n < 0 && WSAEINTR != WSAGetLastError())) {
392         Log("SYNC_ask:  No response on circuit '%s'\n", state->proto_name);
393         res->hdr.response = SYNC_COM_ERROR;
394         goto done;
395     }
396 #else /* !AFS_NT40_ENV */
397     n = write(state->fd, buf, com->hdr.command_len);
398     if (com->hdr.command_len != n) {
399         Log("SYNC_ask: write failed on circuit '%s'\n", state->proto_name);
400         res->hdr.response = SYNC_COM_ERROR;
401         goto done;
402     }
403
404     if (com->hdr.command == SYNC_COM_CHANNEL_CLOSE) {
405         /* short circuit close channel requests */
406         res->hdr.response = SYNC_OK;
407         goto done;
408     }
409
410     /* receive the response */
411     iov[0].iov_base = (char *)&res->hdr;
412     iov[0].iov_len = sizeof(res->hdr);
413     if (res->payload.len) {
414         iov[1].iov_base = (char *)res->payload.buf;
415         iov[1].iov_len = res->payload.len;
416         iovcnt = 2;
417     } else {
418         iovcnt = 1;
419     }
420     n = readv(state->fd, iov, iovcnt);
421     if (n == 0 || (n < 0 && errno != EINTR)) {
422         Log("SYNC_ask: No response on circuit '%s'\n", state->proto_name);
423         res->hdr.response = SYNC_COM_ERROR;
424         goto done;
425     }
426 #endif /* !AFS_NT40_ENV */
427
428     res->recv_len = n;
429
430     if (n < sizeof(res->hdr)) {
431         Log("SYNC_ask:  response too short on circuit '%s'\n",
432             state->proto_name);
433         res->hdr.response = SYNC_COM_ERROR;
434         goto done;
435     }
436 #ifdef AFS_NT40_ENV
437     memcpy(&res->hdr, buf, sizeof(res->hdr));
438 #endif
439
440     if ((n - sizeof(res->hdr)) > res->payload.len) {
441         Log("SYNC_ask:  response too long on circuit '%s'\n",
442             state->proto_name);
443         res->hdr.response = SYNC_COM_ERROR;
444         goto done;
445     }
446 #ifdef AFS_NT40_ENV
447     memcpy(res->payload.buf, buf + sizeof(res->hdr), n - sizeof(res->hdr));
448 #endif
449
450     if (res->hdr.response_len != n) {
451         Log("SYNC_ask:  length field in response inconsistent "
452             "on circuit '%s'\n", state->proto_name);
453         res->hdr.response = SYNC_COM_ERROR;
454         goto done;
455     }
456     if (res->hdr.response == SYNC_DENIED) {
457         Log("SYNC_ask: negative response on circuit '%s'\n", state->proto_name);
458     }
459
460   done:
461     return res->hdr.response;
462 }
463
464
465 /*
466  * daemon com SYNC server-side interfaces
467  */
468
469 /**
470  * receive a command structure off a sync socket.
471  *
472  * @param[in]  state  pointer to server-side state object
473  * @param[in]  fd     file descriptor on which to perform i/o
474  * @param[out] com    sync command object to be populated
475  *
476  * @return operation status
477  *    @retval SYNC_OK command received
478  *    @retval SYNC_COM_ERROR there was a socket communications error
479  */
480 afs_int32
481 SYNC_getCom(SYNC_server_state_t * state,
482             osi_socket fd,
483             SYNC_command * com)
484 {
485     int n;
486     afs_int32 code = SYNC_OK;
487 #ifdef AFS_NT40_ENV
488     SYNC_PROTO_BUF_DECL(buf);
489 #else
490     struct iovec iov[2];
491     int iovcnt;
492 #endif
493
494 #ifdef AFS_NT40_ENV
495     n = recv(fd, buf, SYNC_PROTO_MAX_LEN, 0);
496
497     if (n == 0 || (n < 0 && WSAEINTR != WSAGetLastError())) {
498         Log("SYNC_getCom:  error receiving command\n");
499         code = SYNC_COM_ERROR;
500         goto done;
501     }
502 #else /* !AFS_NT40_ENV */
503     iov[0].iov_base = (char *)&com->hdr;
504     iov[0].iov_len = sizeof(com->hdr);
505     if (com->payload.len) {
506         iov[1].iov_base = (char *)com->payload.buf;
507         iov[1].iov_len = com->payload.len;
508         iovcnt = 2;
509     } else {
510         iovcnt = 1;
511     }
512
513     n = readv(fd, iov, iovcnt);
514     if (n == 0 || (n < 0 && errno != EINTR)) {
515         Log("SYNC_getCom:  error receiving command\n");
516         code = SYNC_COM_ERROR;
517         goto done;
518     }
519 #endif /* !AFS_NT40_ENV */
520
521     com->recv_len = n;
522
523     if (n < sizeof(com->hdr)) {
524         Log("SYNC_getCom:  command too short\n");
525         code = SYNC_COM_ERROR;
526         goto done;
527     }
528 #ifdef AFS_NT40_ENV
529     memcpy(&com->hdr, buf, sizeof(com->hdr));
530 #endif
531
532     if ((n - sizeof(com->hdr)) > com->payload.len) {
533         Log("SYNC_getCom:  command too long\n");
534         code = SYNC_COM_ERROR;
535         goto done;
536     }
537 #ifdef AFS_NT40_ENV
538     memcpy(com->payload.buf, buf + sizeof(com->hdr), n - sizeof(com->hdr));
539 #endif
540
541  done:
542     return code;
543 }
544
545 /**
546  * write a response structure to a sync socket.
547  *
548  * @param[in] state  handle to server-side state object
549  * @param[in] fd     file descriptor on which to perform i/o
550  * @param[in] res    handle to response packet
551  *
552  * @return operation status
553  *    @retval SYNC_OK
554  *    @retval SYNC_COM_ERROR
555  */
556 afs_int32
557 SYNC_putRes(SYNC_server_state_t * state,
558             osi_socket fd,
559             SYNC_response * res)
560 {
561     int n;
562     afs_int32 code = SYNC_OK;
563     SYNC_PROTO_BUF_DECL(buf);
564
565     if (res->hdr.response_len > (sizeof(res->hdr) + res->payload.len)) {
566         Log("SYNC_putRes:  response_len field in response header inconsistent\n");
567         code = SYNC_COM_ERROR;
568         goto done;
569     }
570
571     if (res->hdr.response_len > SYNC_PROTO_MAX_LEN) {
572         Log("SYNC_putRes:  internal SYNC buffer too small; please file a bug\n");
573         code = SYNC_COM_ERROR;
574         goto done;
575     }
576
577 #ifdef AFS_DEMAND_ATTACH_FS
578     res->hdr.flags |= SYNC_FLAG_DAFS_EXTENSIONS;
579 #endif
580     res->hdr.proto_version = state->proto_version;
581     res->hdr.pkt_seq = ++state->pkt_seq;
582     res->hdr.res_seq = ++state->res_seq;
583
584     memcpy(buf, &res->hdr, sizeof(res->hdr));
585     if (res->payload.len) {
586         memcpy(buf + sizeof(res->hdr), res->payload.buf,
587                res->hdr.response_len - sizeof(res->hdr));
588     }
589
590 #ifdef AFS_NT40_ENV
591     n = send(fd, buf, res->hdr.response_len, 0);
592 #else /* !AFS_NT40_ENV */
593     n = write(fd, buf, res->hdr.response_len);
594 #endif /* !AFS_NT40_ENV */
595
596     if (res->hdr.response_len != n) {
597         Log("SYNC_putRes: write failed\n");
598         res->hdr.response = SYNC_COM_ERROR;
599         goto done;
600     }
601
602  done:
603     return code;
604 }
605
606 /* return 0 for legal (null-terminated) string,
607  * 1 for illegal (unterminated) string */
608 int
609 SYNC_verifyProtocolString(char * buf, size_t len)
610 {
611     size_t s_len;
612
613     s_len = strnlen(buf, len);
614
615     return (s_len == len) ? 1 : 0;
616 }
617
618 /**
619  * clean up old sockets.
620  *
621  * @param[in]  state  server state object
622  *
623  * @post unix domain sockets are cleaned up
624  */
625 void
626 SYNC_cleanupSock(SYNC_server_state_t * state)
627 {
628 #ifdef USE_UNIX_SOCKETS
629     remove(state->addr.sun_path);
630 #endif
631 }
632
633 /**
634  * bind socket and set it to listen state.
635  *
636  * @param[in] state  server state object
637  *
638  * @return operation status
639  *    @retval 0 success
640  *    @retval nonzero failure
641  *
642  * @post socket bound and set to listen state
643  */
644 int
645 SYNC_bindSock(SYNC_server_state_t * state)
646 {
647     int code;
648     int on = 1;
649     int numTries;
650
651     /* Reuseaddr needed because system inexplicably leaves crud lying around */
652     code =
653         setsockopt(state->fd, SOL_SOCKET, SO_REUSEADDR, (char *)&on,
654                    sizeof(on));
655     if (code)
656         Log("SYNC_bindSock: setsockopt failed with (%d)\n", errno);
657
658     for (numTries = 0; numTries < state->bind_retry_limit; numTries++) {
659         code = bind(state->fd,
660                     (struct sockaddr *)&state->addr,
661                     AFS_SOCKADDR_LEN(&state->addr));
662         if (code == 0)
663             break;
664         Log("SYNC_bindSock: bind failed with (%d), will sleep and retry\n",
665             errno);
666         sleep(5);
667     }
668     listen(state->fd, state->listen_depth);
669
670     return code;
671 }