lib/libdmsg/msg_lnk.c
void
dmsg_lnk_conn(dmsg_msg_t *msg)
{
	dmsg_state_t *state = msg->state;
	h2span_media_t *media;
	h2span_media_config_t *conf;
	h2span_conn_t *conn;
	h2span_relay_t *relay;
	char *alloc = NULL;
	int i;

	pthread_mutex_lock(&cluster_mtx);

	fprintf(stderr, "dmsg_lnk_conn: msg %p cmd %08x state %p txcmd %08x rxcmd %08x\n",
		msg, msg->any.head.cmd, state, state->txcmd, state->rxcmd);

	switch(msg->any.head.cmd & DMSGF_TRANSMASK) {
	case DMSG_LNK_CONN | DMSGF_CREATE:
	case DMSG_LNK_CONN | DMSGF_CREATE | DMSGF_DELETE:
		/*
		 * On transaction start we allocate a new h2span_conn and
		 * acknowledge the request, leaving the transaction open.
		 * We then relay priority-selected SPANs.
		 */
		fprintf(stderr, "LNK_CONN(%08x): %s/%s/%s\n",
			(uint32_t)msg->any.head.msgid,
			dmsg_uuid_to_str(&msg->any.lnk_conn.pfs_clid,
					    &alloc),
			msg->any.lnk_conn.cl_label,
			msg->any.lnk_conn.fs_label);
		free(alloc);

		conn = dmsg_alloc(sizeof(*conn));

		RB_INIT(&conn->tree);
		state->iocom->conn = conn;	/* XXX only one */
		conn->state = state;
		state->func = dmsg_lnk_conn;
		state->any.conn = conn;
		TAILQ_INSERT_TAIL(&connq, conn, entry);

		/*
		 * Set up media
		 */
		TAILQ_FOREACH(media, &mediaq, entry) {
			if (uuid_compare(&msg->any.lnk_conn.mediaid,
					 &media->mediaid, NULL) == 0) {
				break;
			}
		}
		if (media == NULL) {
			media = dmsg_alloc(sizeof(*media));
			media->mediaid = msg->any.lnk_conn.mediaid;
			TAILQ_INSERT_TAIL(&mediaq, media, entry);
		}
		conn->media = media;
		++media->refs;

		if ((msg->any.head.cmd & DMSGF_DELETE) == 0) {
			dmsg_msg_result(msg, 0);
			dmsg_iocom_signal(msg->iocom);
			break;
		}
		/* FALL THROUGH */
	case DMSG_LNK_CONN | DMSGF_DELETE:
	case DMSG_LNK_ERROR | DMSGF_DELETE:
deleteconn:
		/*
		 * On transaction terminate we clean out our h2span_conn
		 * and acknowledge the request, closing the transaction.
		 */
		fprintf(stderr, "LNK_CONN: Terminated\n");
		conn = state->any.conn;
		assert(conn);

		/*
		 * Clean out the media structure. If refs drops to zero we
		 * also clean out the media config threads.  These threads
		 * maintain span connections to other hammer2 service daemons.
		 */
		media = conn->media;
		if (--media->refs == 0) {
			fprintf(stderr, "Shutting down media spans\n");
			for (i = 0; i < DMSG_COPYID_COUNT; ++i) {
				conf = &media->config[i];

				if (conf->thread == NULL)
					continue;
				conf->ctl = H2CONFCTL_STOP;
				pthread_cond_signal(&conf->cond);
			}
			for (i = 0; i < DMSG_COPYID_COUNT; ++i) {
				conf = &media->config[i];

				if (conf->thread == NULL)
					continue;
				pthread_mutex_unlock(&cluster_mtx);
				pthread_join(conf->thread, NULL);
				pthread_mutex_lock(&cluster_mtx);
				conf->thread = NULL;
				pthread_cond_destroy(&conf->cond);
			}
			fprintf(stderr, "Media shutdown complete\n");
			TAILQ_REMOVE(&mediaq, media, entry);
			dmsg_free(media);
		}

		/*
		 * Clean out all relays.  This requires terminating each
		 * relay transaction.
		 */
		while ((relay = RB_ROOT(&conn->tree)) != NULL) {
			dmsg_relay_delete(relay);
		}

		/*
		 * Clean out conn
		 */
		conn->media = NULL;
		conn->state = NULL;
		msg->state->any.conn = NULL;
		msg->state->iocom->conn = NULL;
		TAILQ_REMOVE(&connq, conn, entry);
		dmsg_free(conn);

		dmsg_msg_reply(msg, 0);
		/* state invalid after reply */
		break;
	case DMSG_LNK_VOLCONF:
		/*
		 * One-way volume-configuration message is transmitted
		 * over the open LNK_CONN transaction.
		 */
		fprintf(stderr, "RECEIVED VOLCONF\n");
		if (msg->any.lnk_volconf.index < 0 ||
		    msg->any.lnk_volconf.index >= DMSG_COPYID_COUNT) {
			fprintf(stderr, "VOLCONF: ILLEGAL INDEX %d\n",
				msg->any.lnk_volconf.index);
			break;
		}
		if (msg->any.lnk_volconf.copy.path[sizeof(msg->any.lnk_volconf.copy.path) - 1] != 0 ||
		    msg->any.lnk_volconf.copy.path[0] == 0) {
			fprintf(stderr, "VOLCONF: ILLEGAL PATH %d\n",
				msg->any.lnk_volconf.index);
			break;
		}
		conn = msg->state->any.conn;
		if (conn == NULL) {
			fprintf(stderr, "VOLCONF: LNK_CONN is missing\n");
			break;
		}
		conf = &conn->media->config[msg->any.lnk_volconf.index];
		conf->copy_pend = msg->any.lnk_volconf.copy;
		conf->ctl |= H2CONFCTL_UPDATE;
		if (conf->thread == NULL) {
			fprintf(stderr, "VOLCONF THREAD STARTED\n");
			pthread_cond_init(&conf->cond, NULL);
			pthread_create(&conf->thread, NULL,
				       dmsg_volconf_thread, (void *)conf);
		}
		pthread_cond_signal(&conf->cond);
		break;
	default:
		/*
		 * Failsafe
		 */
		if (msg->any.head.cmd & DMSGF_DELETE)
			goto deleteconn;
		dmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
		break;
	}
	pthread_mutex_unlock(&cluster_mtx);
}
/*
 * LNK_CONN - iocom identify message reception.
 *	      (incoming iocom lock not held)
 *
 * Remote node identifies itself to us, sets up a SPAN filter, and gives us
 * the ok to start transmitting SPANs.
 */
iocom identify message reception.
(incoming iocom lock not held)

Remote node identifies itself to us, sets up a SPAN filter, and gives us
the ok to start transmitting SPANs.
iocom identify message reception . ( incoming iocom lock not held ) Remote node identifies itself to us sets up a SPAN filter and gives us the ok to start transmitting SPANs .
void
dmsg_lnk_conn(dmsg_msg_t *msg)
{
	dmsg_state_t *state = msg->state;
	h2span_media_t *media;
	h2span_media_config_t *conf;
	h2span_conn_t *conn;
	h2span_relay_t *relay;
	char *alloc = NULL;
	int i;
	pthread_mutex_lock(&cluster_mtx);
	fprintf(stderr, "dmsg_lnk_conn: msg %p cmd %08x state %p txcmd %08x rxcmd %08x\n",
		msg, msg->any.head.cmd, state, state->txcmd, state->rxcmd);
	switch(msg->any.head.cmd & DMSGF_TRANSMASK) {
	case DMSG_LNK_CONN | DMSGF_CREATE:
	case DMSG_LNK_CONN | DMSGF_CREATE | DMSGF_DELETE:
		fprintf(stderr, "LNK_CONN(%08x): %s/%s/%s\n",
			(uint32_t)msg->any.head.msgid,
			dmsg_uuid_to_str(&msg->any.lnk_conn.pfs_clid,
					    &alloc),
			msg->any.lnk_conn.cl_label,
			msg->any.lnk_conn.fs_label);
		free(alloc);
		conn = dmsg_alloc(sizeof(*conn));
		RB_INIT(&conn->tree);
		state->iocom->conn = conn;	
		conn->state = state;
		state->func = dmsg_lnk_conn;
		state->any.conn = conn;
		TAILQ_INSERT_TAIL(&connq, conn, entry);
		TAILQ_FOREACH(media, &mediaq, entry) {
			if (uuid_compare(&msg->any.lnk_conn.mediaid,
					 &media->mediaid, NULL) == 0) {
				break;
			}
		}
		if (media == NULL) {
			media = dmsg_alloc(sizeof(*media));
			media->mediaid = msg->any.lnk_conn.mediaid;
			TAILQ_INSERT_TAIL(&mediaq, media, entry);
		}
		conn->media = media;
		++media->refs;
		if ((msg->any.head.cmd & DMSGF_DELETE) == 0) {
			dmsg_msg_result(msg, 0);
			dmsg_iocom_signal(msg->iocom);
			break;
		}
	case DMSG_LNK_CONN | DMSGF_DELETE:
	case DMSG_LNK_ERROR | DMSGF_DELETE:
deleteconn:
		fprintf(stderr, "LNK_CONN: Terminated\n");
		conn = state->any.conn;
		assert(conn);
		media = conn->media;
		if (--media->refs == 0) {
			fprintf(stderr, "Shutting down media spans\n");
			for (i = 0; i < DMSG_COPYID_COUNT; ++i) {
				conf = &media->config[i];
				if (conf->thread == NULL)
					continue;
				conf->ctl = H2CONFCTL_STOP;
				pthread_cond_signal(&conf->cond);
			}
			for (i = 0; i < DMSG_COPYID_COUNT; ++i) {
				conf = &media->config[i];
				if (conf->thread == NULL)
					continue;
				pthread_mutex_unlock(&cluster_mtx);
				pthread_join(conf->thread, NULL);
				pthread_mutex_lock(&cluster_mtx);
				conf->thread = NULL;
				pthread_cond_destroy(&conf->cond);
			}
			fprintf(stderr, "Media shutdown complete\n");
			TAILQ_REMOVE(&mediaq, media, entry);
			dmsg_free(media);
		}
		while ((relay = RB_ROOT(&conn->tree)) != NULL) {
			dmsg_relay_delete(relay);
		}
		conn->media = NULL;
		conn->state = NULL;
		msg->state->any.conn = NULL;
		msg->state->iocom->conn = NULL;
		TAILQ_REMOVE(&connq, conn, entry);
		dmsg_free(conn);
		dmsg_msg_reply(msg, 0);
		break;
	case DMSG_LNK_VOLCONF:
		fprintf(stderr, "RECEIVED VOLCONF\n");
		if (msg->any.lnk_volconf.index < 0 ||
		    msg->any.lnk_volconf.index >= DMSG_COPYID_COUNT) {
			fprintf(stderr, "VOLCONF: ILLEGAL INDEX %d\n",
				msg->any.lnk_volconf.index);
			break;
		}
		if (msg->any.lnk_volconf.copy.path[sizeof(msg->any.lnk_volconf.copy.path) - 1] != 0 ||
		    msg->any.lnk_volconf.copy.path[0] == 0) {
			fprintf(stderr, "VOLCONF: ILLEGAL PATH %d\n",
				msg->any.lnk_volconf.index);
			break;
		}
		conn = msg->state->any.conn;
		if (conn == NULL) {
			fprintf(stderr, "VOLCONF: LNK_CONN is missing\n");
			break;
		}
		conf = &conn->media->config[msg->any.lnk_volconf.index];
		conf->copy_pend = msg->any.lnk_volconf.copy;
		conf->ctl |= H2CONFCTL_UPDATE;
		if (conf->thread == NULL) {
			fprintf(stderr, "VOLCONF THREAD STARTED\n");
			pthread_cond_init(&conf->cond, NULL);
			pthread_create(&conf->thread, NULL,
				       dmsg_volconf_thread, (void *)conf);
		}
		pthread_cond_signal(&conf->cond);
		break;
	default:
		if (msg->any.head.cmd & DMSGF_DELETE)
			goto deleteconn;
		dmsg_msg_reply(msg, DMSG_ERR_NOSUPP);
		break;
	}
	pthread_mutex_unlock(&cluster_mtx);
}
void dmsg_lnk_conn ( dmsg_msg_t * msg ) { dmsg_state_t * state = msg -> state ; h2span_media_t * media ; h2span_media_config_t * conf ; h2span_conn_t * conn ; h2span_relay_t * relay ; char * alloc = NULL ; int i ; pthread_mutex_lock ( & cluster_mtx ) ; fprintf ( stderr , " \n " , msg , msg -> any . head . cmd , state , state -> txcmd , state -> rxcmd ) ; switch ( msg -> any . head . cmd & DMSGF_TRANSMASK ) { case DMSG_LNK_CONN | DMSGF_CREATE : case DMSG_LNK_CONN | DMSGF_CREATE | DMSGF_DELETE : fprintf ( stderr , " \n " , ( uint32_t ) msg -> any . head . msgid , dmsg_uuid_to_str ( & msg -> any . lnk_conn . pfs_clid , & alloc ) , msg -> any . lnk_conn . cl_label , msg -> any . lnk_conn . fs_label ) ; free ( alloc ) ; conn = dmsg_alloc ( sizeof ( * conn ) ) ; RB_INIT ( & conn -> tree ) ; state -> iocom -> conn = conn ; conn -> state = state ; state -> func = dmsg_lnk_conn ; state -> any . conn = conn ; TAILQ_INSERT_TAIL ( & connq , conn , entry ) ; TAILQ_FOREACH ( media , & mediaq , entry ) { if ( uuid_compare ( & msg -> any . lnk_conn . mediaid , & media -> mediaid , NULL ) == 0 ) { break ; } } if ( media == NULL ) { media = dmsg_alloc ( sizeof ( * media ) ) ; media -> mediaid = msg -> any . lnk_conn . mediaid ; TAILQ_INSERT_TAIL ( & mediaq , media , entry ) ; } conn -> media = media ; ++ media -> refs ; if ( ( msg -> any . head . cmd & DMSGF_DELETE ) == 0 ) { dmsg_msg_result ( msg , 0 ) ; dmsg_iocom_signal ( msg -> iocom ) ; break ; } case DMSG_LNK_CONN | DMSGF_DELETE : case DMSG_LNK_ERROR | DMSGF_DELETE : deleteconn : fprintf ( stderr , " \n " ) ; conn = state -> any . conn ; assert ( conn ) ; media = conn -> media ; if ( -- media -> refs == 0 ) { fprintf ( stderr , " \n " ) ; for ( i = 0 ; i < DMSG_COPYID_COUNT ; ++ i ) { conf = & media -> config [ i ] ; if ( conf -> thread == NULL ) continue ; conf -> ctl = H2CONFCTL_STOP ; pthread_cond_signal ( & conf -> cond ) ; } for ( i = 0 ; i < DMSG_COPYID_COUNT ; ++ i ) { conf = & media -> config [ i ] ; if ( conf -> thread == NULL ) continue ; pthread_mutex_unlock ( & cluster_mtx ) ; pthread_join ( conf -> thread , NULL ) ; pthread_mutex_lock ( & cluster_mtx ) ; conf -> thread = NULL ; pthread_cond_destroy ( & conf -> cond ) ; } fprintf ( stderr , " \n " ) ; TAILQ_REMOVE ( & mediaq , media , entry ) ; dmsg_free ( media ) ; } while ( ( relay = RB_ROOT ( & conn -> tree ) ) != NULL ) { dmsg_relay_delete ( relay ) ; } conn -> media = NULL ; conn -> state = NULL ; msg -> state -> any . conn = NULL ; msg -> state -> iocom -> conn = NULL ; TAILQ_REMOVE ( & connq , conn , entry ) ; dmsg_free ( conn ) ; dmsg_msg_reply ( msg , 0 ) ; break ; case DMSG_LNK_VOLCONF : fprintf ( stderr , " \n " ) ; if ( msg -> any . lnk_volconf . index < 0 || msg -> any . lnk_volconf . index >= DMSG_COPYID_COUNT ) { fprintf ( stderr , " \n " , msg -> any . lnk_volconf . index ) ; break ; } if ( msg -> any . lnk_volconf . copy . path [ sizeof ( msg -> any . lnk_volconf . copy . path ) - 1 ] != 0 || msg -> any . lnk_volconf . copy . path [ 0 ] == 0 ) { fprintf ( stderr , " \n " , msg -> any . lnk_volconf . index ) ; break ; } conn = msg -> state -> any . conn ; if ( conn == NULL ) { fprintf ( stderr , " \n " ) ; break ; } conf = & conn -> media -> config [ msg -> any . lnk_volconf . index ] ; conf -> copy_pend = msg -> any . lnk_volconf . copy ; conf -> ctl |= H2CONFCTL_UPDATE ; if ( conf -> thread == NULL ) { fprintf ( stderr , " \n " ) ; pthread_cond_init ( & conf -> cond , NULL ) ; pthread_create ( & conf -> thread , NULL , dmsg_volconf_thread , ( void * ) conf ) ; } pthread_cond_signal ( & conf -> cond ) ; break ; default : if ( msg -> any . head . cmd & DMSGF_DELETE ) goto deleteconn ; dmsg_msg_reply ( msg , DMSG_ERR_NOSUPP ) ; break ; } pthread_mutex_unlock ( & cluster_mtx ) ; }
LNK_CONN - iocom identify message reception.
LNK_CONN - iocom identify message reception .
/*
		 * On transaction start we allocate a new h2span_conn and
		 * acknowledge the request, leaving the transaction open.
		 * We then relay priority-selected SPANs.
		 */ /* XXX only one */ /*
		 * Set up media
		 */ /* FALL THROUGH */ /*
		 * On transaction terminate we clean out our h2span_conn
		 * and acknowledge the request, closing the transaction.
		 */ /*
		 * Clean out the media structure. If refs drops to zero we
		 * also clean out the media config threads.  These threads
		 * maintain span connections to other hammer2 service daemons.
		 */ /*
		 * Clean out all relays.  This requires terminating each
		 * relay transaction.
		 */ /*
		 * Clean out conn
		 */ /* state invalid after reply */ /*
		 * One-way volume-configuration message is transmitted
		 * over the open LNK_CONN transaction.
		 */ /*
		 * Failsafe
		 */