2 * Copyright (c) 2011-2012 The DragonFly Project. All rights reserved.
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>
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions
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
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.
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
38 #include <sys/xdiskioctl.h>
39 #include <machine/atomic.h>
41 struct hammer2_media_config {
42 hammer2_volconf_t copy_run;
43 hammer2_volconf_t copy_pend;
48 int pipefd[2]; /* signal stop */
50 pthread_t iocom_thread;
51 enum { H2MC_STOPPED, H2MC_CONNECT, H2MC_RUNNING } state;
54 typedef struct hammer2_media_config hammer2_media_config_t;
56 #define H2CONFCTL_STOP 0x00000001
57 #define H2CONFCTL_UPDATE 0x00000002
60 TAILQ_ENTRY(diskcon) entry;
64 struct service_node_opaque {
67 dmsg_media_block_t block;
74 TAILQ_ENTRY(autoconn) entry;
78 int pipefd[2]; /* {read,write} */
79 enum { AUTOCONN_INACTIVE, AUTOCONN_ACTIVE } state;
85 TAILQ_HEAD(, diskcon) diskconq = TAILQ_HEAD_INITIALIZER(diskconq);
86 static pthread_mutex_t diskmtx;
87 static pthread_mutex_t confmtx;
89 static void *service_thread(void *data);
90 static void *udev_thread(void *data);
91 static void *autoconn_thread(void *data);
92 static void master_reconnect(const char *mntpt);
93 static void disk_reconnect(const char *disk);
94 static void disk_disconnect(void *handle);
95 static void udev_check_disks(void);
96 static void hammer2_usrmsg_handler(dmsg_msg_t *msg, int unmanaged);
97 static void hammer2_node_handler(void **opaque, struct dmsg_msg *msg, int op);
98 static void *hammer2_volconf_thread(void *info);
99 static void hammer2_volconf_signal(dmsg_iocom_t *iocom);
100 static void hammer2_volconf_start(hammer2_media_config_t *conf,
101 const char *hostname);
102 static void hammer2_volconf_stop(hammer2_media_config_t *conf);
105 static void xdisk_reconnect(struct service_node_opaque *info);
106 static void xdisk_disconnect(void *handle);
107 static void *xdisk_attach_tmpthread(void *data);
110 * Start-up the master listener daemon for the machine. This daemon runs
111 * a UDP discovery protocol, a TCP rendezvous, and scans certain files
112 * and directories for work.
116 * The only purpose for the UDP discovery protocol is to determine what
117 * other IPs on the LAN are running the hammer2 service daemon. DNS is not
118 * required to operate, but hostnames (if assigned) must be unique. If
119 * no hostname is assigned the host's IP is used as the name. This name
120 * is broadcast along with the mtime of the originator's private key.
122 * Receiving hammer2 service daemons which are able to match the label against
123 * /etc/hammer2/remote/<label>.pub will initiate a persistent connection
124 * to the target. Removal of the file will cause a disconnection. A failed
125 * public key negotiation stops further connection attempts until either the
126 * file is updated or the remote mtime is updated.
128 * Generally speaking this results in a web of connections, typically a
129 * combination of point-to-point for the more important links and relayed
130 * (spanning tree) for less important or filtered links.
134 * The TCP listener serves as a rendezvous point in the cluster, accepting
135 * connections, performing registrations and authentications, maintaining
136 * the spanning tree, and keeping track of message state so disconnects can
137 * be handled properly.
139 * Once authenticated only low-level messaging protocols (which includes
140 * tracking persistent messages) are handled by this daemon. This daemon
141 * does not run the higher level quorum or locking protocols.
145 * The file /etc/hammer2/autoconn, if it exists, contains a list of targets
146 * to connect to (which do not have to be on the local lan). This list will
147 * be retried until a connection can be established. The file is not usually
148 * needed for linkages local to the LAN.
153 struct sockaddr_in lsin;
158 * Acquire socket and set options
160 if ((lfd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
161 fprintf(stderr, "master_listen: socket(): %s\n",
166 setsockopt(lfd, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
169 * Setup listen port and try to bind. If the bind fails we assume
170 * that a master listener process is already running and silently
173 bzero(&lsin, sizeof(lsin));
174 lsin.sin_family = AF_INET;
175 lsin.sin_addr.s_addr = INADDR_ANY;
176 lsin.sin_port = htons(DMSG_LISTEN_PORT);
177 if (bind(lfd, (struct sockaddr *)&lsin, sizeof(lsin)) < 0) {
181 "master listen: daemon already running\n");
186 fprintf(stderr, "master listen: startup\n");
190 * Fork and disconnect the controlling terminal and parent process,
191 * executing the specified function as a pthread.
193 * Returns to the original process which can then continue running.
194 * In debug mode this call will create the pthread without forking
195 * and set NormalExit to 0, instead of fork.
197 hammer2_demon(service_thread, (void *)(intptr_t)lfd);
204 * Master listen/accept thread. Accept connections on the master socket,
205 * starting a pthread for each one.
209 service_thread(void *data)
211 struct sockaddr_in asin;
214 dmsg_master_service_info_t *info;
215 int lfd = (int)(intptr_t)data;
219 struct statfs *mntbuf = NULL;
220 struct statvfs *mntvbuf = NULL;
223 * Nobody waits for us
225 setproctitle("hammer2 master listen");
226 pthread_detach(pthread_self());
229 * Start up a thread to handle block device monitoring
232 pthread_create(&thread, NULL, udev_thread, NULL);
235 * Start thread to manage /etc/hammer2/autoconn
238 pthread_create(&thread, NULL, autoconn_thread, NULL);
241 * Scan existing hammer2 mounts and reconnect to them using
242 * HAMMER2IOC_RECLUSTER.
244 count = getmntvinfo(&mntbuf, &mntvbuf, MNT_NOWAIT);
245 for (i = 0; i < count; ++i) {
246 if (strcmp(mntbuf[i].f_fstypename, "hammer2") == 0)
247 master_reconnect(mntbuf[i].f_mntonname);
251 * Accept connections and create pthreads to handle them after
256 fd = accept(lfd, (struct sockaddr *)&asin, &alen);
263 fprintf(stderr, "service_thread: accept fd %d\n", fd);
264 info = malloc(sizeof(*info));
265 bzero(info, sizeof(*info));
268 info->usrmsg_callback = hammer2_usrmsg_handler;
269 info->node_handler = hammer2_node_handler;
270 info->label = strdup("client");
271 pthread_create(&thread, NULL, dmsg_master_service, info);
277 * Handle/Monitor the dmsg stream. If unmanaged is set we are responsible
278 * for responding for the message, otherwise if it is not set libdmsg has
279 * already done some preprocessing and will respond to the message for us
282 * We primarily monitor for VOLCONFs
286 hammer2_usrmsg_handler(dmsg_msg_t *msg, int unmanaged)
289 hammer2_media_config_t *conf;
290 dmsg_lnk_hammer2_volconf_t *msgconf;
294 * Only process messages which are part of a LNK_CONN stream
298 (state->rxcmd & DMSGF_BASECMDMASK) != DMSG_LNK_CONN) {
299 hammer2_shell_parse(msg, unmanaged);
303 switch(msg->any.head.cmd & DMSGF_TRANSMASK) {
304 case DMSG_LNK_CONN | DMSGF_CREATE | DMSGF_DELETE:
305 case DMSG_LNK_CONN | DMSGF_DELETE:
306 case DMSG_LNK_ERROR | DMSGF_DELETE:
308 * Deleting connection, clean out all volume configs
310 if (state->media == NULL || state->media->usrhandle == NULL)
312 conf = state->media->usrhandle;
313 fprintf(stderr, "Shutting down media spans\n");
314 for (i = 0; i < HAMMER2_COPYID_COUNT; ++i) {
315 if (conf[i].thread) {
316 conf[i].ctl = H2CONFCTL_STOP;
317 pthread_cond_signal(&conf[i].cond);
320 for (i = 0; i < HAMMER2_COPYID_COUNT; ++i) {
321 if (conf[i].thread) {
322 pthread_join(conf[i].thread, NULL);
324 pthread_cond_destroy(&conf[i].cond);
327 state->media->usrhandle = NULL;
330 case DMSG_LNK_HAMMER2_VOLCONF:
332 * One-way volume-configuration message is transmitted
333 * over the open LNK_CONN transaction.
335 fprintf(stderr, "RECEIVED VOLCONF\n");
337 if ((conf = state->media->usrhandle) == NULL) {
338 conf = malloc(sizeof(*conf) * HAMMER2_COPYID_COUNT);
339 bzero(conf, sizeof(*conf) * HAMMER2_COPYID_COUNT);
340 state->media->usrhandle = conf;
342 msgconf = H2_LNK_VOLCONF(msg);
344 if (msgconf->index < 0 ||
345 msgconf->index >= HAMMER2_COPYID_COUNT) {
347 "VOLCONF: ILLEGAL INDEX %d\n",
351 if (msgconf->copy.path[sizeof(msgconf->copy.path) - 1] != 0 ||
352 msgconf->copy.path[0] == 0) {
354 "VOLCONF: ILLEGAL PATH %d\n",
358 conf += msgconf->index;
359 pthread_mutex_lock(&confmtx);
360 conf->copy_pend = msgconf->copy;
361 conf->ctl |= H2CONFCTL_UPDATE;
362 pthread_mutex_unlock(&confmtx);
363 if (conf->thread == NULL) {
364 fprintf(stderr, "VOLCONF THREAD STARTED\n");
365 pthread_cond_init(&conf->cond, NULL);
366 pthread_create(&conf->thread, NULL,
367 hammer2_volconf_thread, (void *)conf);
369 pthread_cond_signal(&conf->cond);
373 dmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
379 hammer2_volconf_thread(void *info)
381 hammer2_media_config_t *conf = info;
383 pthread_mutex_lock(&confmtx);
384 while ((conf->ctl & H2CONFCTL_STOP) == 0) {
385 if (conf->ctl & H2CONFCTL_UPDATE) {
386 fprintf(stderr, "VOLCONF UPDATE\n");
387 conf->ctl &= ~H2CONFCTL_UPDATE;
388 if (bcmp(&conf->copy_run, &conf->copy_pend,
389 sizeof(conf->copy_run)) == 0) {
390 fprintf(stderr, "VOLCONF: no changes\n");
394 * XXX TODO - auto reconnect on lookup failure or
395 * connect failure or stream failure.
398 pthread_mutex_unlock(&confmtx);
399 hammer2_volconf_stop(conf);
400 conf->copy_run = conf->copy_pend;
401 if (conf->copy_run.copyid != 0 &&
402 strncmp(conf->copy_run.path, "span:", 5) == 0) {
403 hammer2_volconf_start(conf,
404 conf->copy_run.path + 5);
406 pthread_mutex_lock(&confmtx);
407 fprintf(stderr, "VOLCONF UPDATE DONE state %d\n", conf->state);
409 if (conf->state == H2MC_CONNECT) {
410 hammer2_volconf_start(conf, conf->copy_run.path + 5);
411 pthread_mutex_unlock(&confmtx);
413 pthread_mutex_lock(&confmtx);
415 pthread_cond_wait(&conf->cond, &confmtx);
418 pthread_mutex_unlock(&confmtx);
419 hammer2_volconf_stop(conf);
425 hammer2_volconf_start(hammer2_media_config_t *conf, const char *hostname)
427 dmsg_master_service_info_t *info;
429 switch(conf->state) {
432 conf->fd = dmsg_connect(hostname);
434 fprintf(stderr, "Unable to connect to %s\n", hostname);
435 conf->state = H2MC_CONNECT;
436 } else if (pipe(conf->pipefd) < 0) {
438 fprintf(stderr, "pipe() failed during volconf\n");
439 conf->state = H2MC_CONNECT;
441 fprintf(stderr, "VOLCONF CONNECT\n");
442 info = malloc(sizeof(*info));
443 bzero(info, sizeof(*info));
445 info->altfd = conf->pipefd[0];
446 info->altmsg_callback = hammer2_volconf_signal;
447 info->usrmsg_callback = hammer2_usrmsg_handler;
449 conf->state = H2MC_RUNNING;
450 pthread_create(&conf->iocom_thread, NULL,
451 dmsg_master_service, info);
461 hammer2_volconf_stop(hammer2_media_config_t *conf)
463 switch(conf->state) {
467 conf->state = H2MC_STOPPED;
470 close(conf->pipefd[1]);
471 conf->pipefd[1] = -1;
472 pthread_join(conf->iocom_thread, NULL);
473 conf->iocom_thread = NULL;
474 conf->state = H2MC_STOPPED;
481 hammer2_volconf_signal(dmsg_iocom_t *iocom)
483 atomic_set_int(&iocom->flags, DMSG_IOCOMF_EOF);
488 * Node discovery code on received SPANs (or loss of SPANs). This code
489 * is used to track the availability of remote block devices and install
490 * or deinstall them using the xdisk driver (/dev/xdisk).
492 * An installed xdisk creates /dev/xa%d and /dev/serno/<blah> based on
493 * the data handed to it. When opened, a virtual circuit is forged and
494 * maintained to the block device server via DMSG. Temporary failures
495 * stall the device until successfully reconnected or explicitly destroyed.
499 hammer2_node_handler(void **opaquep, struct dmsg_msg *msg, int op)
501 struct service_node_opaque *info = *opaquep;
503 fprintf(stderr, "NODE HANLDER A **********************\n");
506 case DMSG_NODEOP_ADD:
507 if (msg->any.lnk_span.peer_type != DMSG_PEER_BLOCK)
509 fprintf(stderr, "NODE HANLDER B %d **********************\n",
510 msg->any.lnk_span.pfs_type);
511 if (msg->any.lnk_span.pfs_type != DMSG_PFSTYPE_SERVER)
513 fprintf(stderr, "NODE HANLDER C **********************\n");
515 info = malloc(sizeof(*info));
516 bzero(info, sizeof(*info));
519 snprintf(info->cl_label, sizeof(info->cl_label),
520 "%s", msg->any.lnk_span.cl_label);
521 snprintf(info->fs_label, sizeof(info->fs_label),
522 "%s", msg->any.lnk_span.fs_label);
523 info->block = msg->any.lnk_span.media.block;
524 fprintf(stderr, "NODE ADD %s serno %s\n",
525 info->cl_label, info->fs_label);
527 xdisk_reconnect(info);
529 case DMSG_NODEOP_DEL:
531 fprintf(stderr, "NODE DEL %s serno %s\n",
532 info->cl_label, info->fs_label);
533 pthread_mutex_lock(&diskmtx);
536 if (info->servicing == 0)
539 shutdown(info->servicefd, SHUT_RDWR);/*XXX*/
540 pthread_mutex_unlock(&diskmtx);
549 * Monitor block devices. Currently polls every ~10 seconds or so.
553 udev_thread(void *data __unused)
558 pthread_detach(pthread_self());
560 if ((fd = open(UDEV_DEVICE_PATH, O_RDWR)) < 0) {
561 fprintf(stderr, "udev_thread: unable to open \"%s\"\n",
566 while (ioctl(fd, UDEVWAIT, &seq) == 0) {
573 static void *autoconn_connect_thread(void *data);
574 static void autoconn_disconnect_signal(dmsg_iocom_t *iocom);
578 autoconn_thread(void *data __unused)
580 TAILQ_HEAD(, autoconn) autolist;
582 struct autoconn *next;
591 TAILQ_INIT(&autolist);
595 pthread_detach(pthread_self());
603 * Poll the file. Loop up if the synchronized state (lmod)
606 if (stat(HAMMER2_DEFAULT_DIR "/autoconn", &st) == 0) {
607 if (lmod == st.st_mtime)
609 fp = fopen(HAMMER2_DEFAULT_DIR "/autoconn", "r");
619 * Wait at least 5 seconds after the file is created or
622 * Do not update the synchronized state.
624 if (fp == NULL && found_last) {
627 } else if (fp && found_last == 0) {
634 * Don't scan the file until the time progresses past the
635 * file's mtime, so we can validate that the file was not
636 * further modified during our scan.
638 * Do not update the synchronized state.
642 if (t == st.st_mtime) {
652 * Set staging to disconnect, then scan the file.
654 TAILQ_FOREACH(ac, &autolist, entry)
656 while (fp && fgets(buf, sizeof(buf), fp) != NULL) {
659 if ((host = strtok(buf, " \t\r\n")) == NULL ||
663 TAILQ_FOREACH(ac, &autolist, entry) {
664 if (strcmp(host, ac->host) == 0)
668 ac = malloc(sizeof(*ac));
669 bzero(ac, sizeof(*ac));
670 ac->host = strdup(host);
671 ac->state = AUTOCONN_INACTIVE;
672 TAILQ_INSERT_TAIL(&autolist, ac, entry);
678 * Ignore the scan (and retry again) if the file was
679 * modified during the scan.
681 * Do not update the synchronized state.
684 if (fstat(fileno(fp), &st) < 0) {
689 if (t != st.st_mtime)
694 * Update the synchronized state and reconfigure the
695 * connect list as needed.
698 next = TAILQ_FIRST(&autolist);
699 while ((ac = next) != NULL) {
700 next = TAILQ_NEXT(ac, entry);
705 if (ac->stage && ac->state == AUTOCONN_INACTIVE) {
706 if (pipe(ac->pipefd) == 0) {
708 ac->state = AUTOCONN_ACTIVE;
710 pthread_create(&thread, NULL,
711 autoconn_connect_thread,
717 * Unstaging, stop active connection.
719 * We write to the pipe which causes the iocom_core
720 * to call autoconn_disconnect_signal().
722 if (ac->stage == 0 &&
723 ac->state == AUTOCONN_ACTIVE) {
724 if (ac->stopme == 0) {
727 write(ac->pipefd[1], &dummy, 1);
732 * Unstaging, delete inactive connection.
734 if (ac->stage == 0 &&
735 ac->state == AUTOCONN_INACTIVE) {
736 TAILQ_REMOVE(&autolist, ac, entry);
749 autoconn_connect_thread(void *data)
751 dmsg_master_service_info_t *info;
757 pthread_detach(pthread_self());
759 while (ac->stopme == 0) {
760 fd = dmsg_connect(ac->host);
762 fprintf(stderr, "autoconn: Connect failure: %s\n",
767 fprintf(stderr, "autoconn: Connect %s\n", ac->host);
769 info = malloc(sizeof(*info));
770 bzero(info, sizeof(*info));
772 info->altfd = ac->pipefd[0];
773 info->altmsg_callback = autoconn_disconnect_signal;
774 info->usrmsg_callback = hammer2_usrmsg_handler;
776 info->noclosealt = 1;
777 pthread_create(&ac->thread, NULL, dmsg_master_service, info);
778 pthread_join(ac->thread, &res);
780 close(ac->pipefd[0]);
781 ac->state = AUTOCONN_INACTIVE;
782 /* auto structure can be ripped out here */
788 autoconn_disconnect_signal(dmsg_iocom_t *iocom)
790 fprintf(stderr, "autoconn: Shutting down socket\n");
791 atomic_set_int(&iocom->flags, DMSG_IOCOMF_EOF);
795 * Retrieve the list of disk attachments and attempt to export
800 udev_check_disks(void)
810 error = sysctlbyname("kern.disks", NULL, &n, NULL, 0);
811 if (error < 0 || n == 0)
813 if (n >= sizeof(tmpbuf))
817 error = sysctlbyname("kern.disks", buf, &n, NULL, 0);
830 fprintf(stderr, "DISKS: %s\n", buf);
831 for (disk = strtok(buf, WS); disk; disk = strtok(NULL, WS)) {
832 disk_reconnect(disk);
840 * Normally the mount program supplies a cluster communications
841 * descriptor to the hammer2 vfs on mount, but if you kill the service
842 * daemon and restart it that link will be lost.
844 * This procedure attempts to [re]connect to existing mounts when
845 * the service daemon is started up before going into its accept
848 * NOTE: A hammer2 mount point can only accomodate one connection at a time
849 * so this will disconnect any existing connection during the
854 master_reconnect(const char *mntpt)
856 struct hammer2_ioc_recluster recls;
857 dmsg_master_service_info_t *info;
862 fd = open(mntpt, O_RDONLY);
864 fprintf(stderr, "reconnect %s: no access to mount\n", mntpt);
867 if (pipe(pipefds) < 0) {
868 fprintf(stderr, "reconnect %s: pipe() failed\n", mntpt);
872 bzero(&recls, sizeof(recls));
873 recls.fd = pipefds[0];
874 if (ioctl(fd, HAMMER2IOC_RECLUSTER, &recls) < 0) {
875 fprintf(stderr, "reconnect %s: ioctl failed\n", mntpt);
884 info = malloc(sizeof(*info));
885 bzero(info, sizeof(*info));
886 info->fd = pipefds[1];
888 info->usrmsg_callback = hammer2_usrmsg_handler;
889 info->node_handler = hammer2_node_handler;
890 info->label = strdup("hammer2");
891 pthread_create(&thread, NULL, dmsg_master_service, info);
895 * Reconnect a physical disk service to the mesh.
899 disk_reconnect(const char *disk)
901 struct disk_ioc_recluster recls;
903 dmsg_master_service_info_t *info;
910 * Urm, this will auto-create mdX+1, just ignore for now.
911 * This mechanic needs to be fixed. It might actually be nice
912 * to be able to export md disks.
914 if (strncmp(disk, "md", 2) == 0)
916 if (strncmp(disk, "xa", 2) == 0)
920 * Check if already connected
922 pthread_mutex_lock(&diskmtx);
923 TAILQ_FOREACH(dc, &diskconq, entry) {
924 if (strcmp(dc->disk, disk) == 0)
927 pthread_mutex_unlock(&diskmtx);
932 * Not already connected, create a connection to the kernel
935 asprintf(&path, "/dev/%s", disk);
936 fd = open(path, O_RDONLY);
938 fprintf(stderr, "reconnect %s: no access to disk\n", disk);
943 if (pipe(pipefds) < 0) {
944 fprintf(stderr, "reconnect %s: pipe() failed\n", disk);
948 bzero(&recls, sizeof(recls));
949 recls.fd = pipefds[0];
950 if (ioctl(fd, DIOCRECLUSTER, &recls) < 0) {
951 fprintf(stderr, "reconnect %s: ioctl failed\n", disk);
960 dc = malloc(sizeof(*dc));
961 dc->disk = strdup(disk);
962 pthread_mutex_lock(&diskmtx);
963 TAILQ_INSERT_TAIL(&diskconq, dc, entry);
964 pthread_mutex_unlock(&diskmtx);
966 info = malloc(sizeof(*info));
967 bzero(info, sizeof(*info));
968 info->fd = pipefds[1];
970 info->usrmsg_callback = hammer2_usrmsg_handler;
971 info->node_handler = hammer2_node_handler;
972 info->exit_callback = disk_disconnect;
974 info->label = strdup(dc->disk);
975 pthread_create(&thread, NULL, dmsg_master_service, info);
980 disk_disconnect(void *handle)
982 struct diskcon *dc = handle;
984 fprintf(stderr, "DISK_DISCONNECT %s\n", dc->disk);
986 pthread_mutex_lock(&diskmtx);
987 TAILQ_REMOVE(&diskconq, dc, entry);
988 pthread_mutex_unlock(&diskmtx);
994 * [re]connect a remote disk service to the local system via /dev/xdisk.
998 xdisk_reconnect(struct service_node_opaque *xdisk)
1000 struct xdisk_attach_ioctl *xaioc;
1001 dmsg_master_service_info_t *info;
1005 if (pipe(pipefds) < 0) {
1006 fprintf(stderr, "reconnect %s: pipe() failed\n",
1011 info = malloc(sizeof(*info));
1012 bzero(info, sizeof(*info));
1013 info->fd = pipefds[1];
1015 info->usrmsg_callback = hammer2_usrmsg_handler;
1016 info->node_handler = hammer2_node_handler;
1017 info->exit_callback = xdisk_disconnect;
1018 info->handle = xdisk;
1019 xdisk->servicing = 1;
1020 xdisk->servicefd = info->fd;
1021 info->label = strdup(xdisk->cl_label);
1022 pthread_create(&thread, NULL, dmsg_master_service, info);
1025 * We have to run the attach in its own pthread because it will
1026 * synchronously interact with the messaging subsystem over the
1027 * pipe. If we do it here we will deadlock.
1029 xaioc = malloc(sizeof(*xaioc));
1030 bzero(xaioc, sizeof(*xaioc));
1031 snprintf(xaioc->cl_label, sizeof(xaioc->cl_label),
1032 "%s", xdisk->cl_label);
1033 snprintf(xaioc->fs_label, sizeof(xaioc->fs_label),
1034 "X-%s", xdisk->fs_label);
1035 xaioc->bytes = xdisk->block.bytes;
1036 xaioc->blksize = xdisk->block.blksize;
1037 xaioc->fd = pipefds[0];
1039 pthread_create(&thread, NULL, xdisk_attach_tmpthread, xaioc);
1044 xdisk_attach_tmpthread(void *data)
1046 struct xdisk_attach_ioctl *xaioc = data;
1049 pthread_detach(pthread_self());
1051 fd = open("/dev/xdisk", O_RDWR, 0600);
1053 fprintf(stderr, "xdisk_reconnect: Unable to open /dev/xdisk\n");
1055 if (ioctl(fd, XDISKIOCATTACH, xaioc) < 0) {
1056 fprintf(stderr, "reconnect %s: xdisk attach failed\n",
1066 xdisk_disconnect(void *handle)
1068 struct service_node_opaque *info = handle;
1070 assert(info->servicing == 1);
1072 pthread_mutex_lock(&diskmtx);
1073 info->servicing = 0;
1074 if (info->attached == 0)
1076 pthread_mutex_unlock(&diskmtx);