1 /*
2 * CDDL HEADER START
3 *
4 * The contents of this file are subject to the terms of the
5 * Common Development and Distribution License (the "License").
6 * You may not use this file except in compliance with the License.
7 *
8 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9 * or http://www.opensolaris.org/os/licensing.
10 * See the License for the specific language governing permissions
11 * and limitations under the License.
12 *
13 * When distributing Covered Code, include this CDDL HEADER in each
14 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15 * If applicable, add the following below this CDDL HEADER, with the
16 * fields enclosed by brackets "[]" replaced with your own identifying
17 * information: Portions Copyright [yyyy] [name of copyright owner]
18 *
19 * CDDL HEADER END
20 */
21
22 /*
23 * Copyright 2006 Sun Microsystems, Inc. All rights reserved.
24 * Use is subject to license terms.
25 */
26
27 #pragma ident "%Z%%M% %I% %E% SMI"
28
29 /*
30 * FMA Event Transport Module
31 *
32 * Plugin for sending/receiving FMA events to/from a remote endoint.
33 */
34
35 #include <netinet/in.h>
36 #include <errno.h>
37 #include <sys/fm/protocol.h>
38 #include <sys/sysmacros.h>
39 #include <pthread.h>
40 #include <strings.h>
41 #include <ctype.h>
42 #include <link.h>
43 #include <libnvpair.h>
44 #include "etm_xport_api.h"
45 #include "etm_proto.h"
46
47 /*
48 * ETM declarations
49 */
50
51 typedef enum etm_connection_status {
52 C_UNINITIALIZED = 0,
53 C_OPEN, /* Connection is open */
54 C_CLOSED, /* Connection is closed */
55 C_LIMBO, /* Bad value in header from peer */
56 C_TIMED_OUT /* Reconnection to peer timed out */
57 } etm_connstat_t;
58
59 typedef enum etm_fmd_queue_status {
60 Q_UNINITIALIZED = 100,
61 Q_INIT_PENDING, /* Queue initialization in progress */
62 Q_OPEN, /* Queue is open */
63 Q_SUSPENDED /* Queue is suspended */
64 } etm_qstat_t;
65
66 /* Per endpoint data */
67 typedef struct etm_endpoint_map {
68 uint8_t epm_ver; /* Protocol version being used */
69 char *epm_ep_str; /* Endpoint ID string */
70 int epm_xprtflags; /* FMD transport open flags */
71 etm_xport_hdl_t epm_tlhdl; /* Transport Layer instance handle */
72 pthread_mutex_t epm_lock; /* Protects remainder of struct */
73 pthread_cond_t epm_tx_cv; /* Cond var for send/transmit */
74 int epm_txbusy; /* Busy doing send/transmit */
75 fmd_xprt_t *epm_xprthdl; /* FMD transport handle */
76 etm_qstat_t epm_qstat; /* Status of fmd xprt queue */
77 nvlist_t *epm_ep_nvl; /* Endpoint ID nv_list */
78 etm_xport_conn_t epm_oconn; /* Connection for outgoing events */
79 etm_connstat_t epm_cstat; /* Status of connection */
80 id_t epm_timer_id; /* Timer id */
81 int epm_timer_in_use; /* Indicates if timer is in use */
82 hrtime_t epm_reconn_end; /* Reconnection end time */
83 struct etm_endpoint_map *epm_next;
84 } etm_epmap_t;
85
86 #define ETM_HDR_INVALID (ETM_HDR_TYPE_TOO_HIGH + 1)
87 #define ETM_HDR_BADVERSION (ETM_HDR_TYPE_TOO_HIGH + 2)
88 #define ETM_HDR_BADTYPE (ETM_HDR_TYPE_TOO_HIGH + 3)
89 #define ETM_EP_INST_MAX 4 /* Max chars in endpt instance */
90 #define ETM_CLIENT_XPRT_FLAGS FMD_XPRT_RDWR
91 #define ETM_SERVER_XPRT_FLAGS (FMD_XPRT_RDWR | FMD_XPRT_ACCEPT)
92
93 #define ALLOC_BUF(hdl, buf, size) \
94 buf = fmd_hdl_zalloc((hdl), (size), FMD_SLEEP);
95
96 #define FREE_BUF(hdl, buf, size) fmd_hdl_free((hdl), (buf), (size));
97
98 #define IS_CLIENT(mp) (((mp)->epm_xprtflags & FMD_XPRT_ACCEPT) ? 0 : 1)
99
100 #define INCRSTAT(x) { (void) pthread_mutex_lock(&Etm_mod_lock); \
101 (x)++; \
102 (void) pthread_mutex_unlock(&Etm_mod_lock); \
103 }
104
105 #define DECRSTAT(x) { (void) pthread_mutex_lock(&Etm_mod_lock); \
106 (x)--; \
107 (void) pthread_mutex_unlock(&Etm_mod_lock); \
108 }
109
110 #define ADDSTAT(x, y) { (void) pthread_mutex_lock(&Etm_mod_lock); \
111 (x) += (y); \
112 (void) pthread_mutex_unlock(&Etm_mod_lock); \
113 }
114
115 /*
116 * Global variables
117 */
118 static pthread_mutex_t Etm_mod_lock = PTHREAD_MUTEX_INITIALIZER;
119 /* Protects globals */
120 static hrtime_t Reconn_interval; /* Time between reconnection attempts */
121 static hrtime_t Reconn_timeout; /* Time allowed for reconnection */
122 static hrtime_t Rw_timeout; /* Time allowed for I/O operation */
123 static int Etm_dump = 0; /* Enables hex dump for debug */
124 static int Etm_exit = 0; /* Flag for exit */
125 static etm_epmap_t *Epmap_head = NULL; /* Head of list of epmap structs */
126
127 /* Module statistics */
128 static struct etm_stats {
129 /* read counters */
130 fmd_stat_t read_ack;
131 fmd_stat_t read_bytes;
132 fmd_stat_t read_msg;
133 fmd_stat_t post_filter;
134 /* write counters */
135 fmd_stat_t write_ack;
136 fmd_stat_t write_bytes;
137 fmd_stat_t write_msg;
138 fmd_stat_t send_filter;
139 /* error counters */
140 fmd_stat_t error_protocol;
141 fmd_stat_t error_drop_read;
142 fmd_stat_t error_read;
143 fmd_stat_t error_read_badhdr;
144 fmd_stat_t error_write;
145 fmd_stat_t error_send_filter;
146 fmd_stat_t error_post_filter;
147 /* misc */
148 fmd_stat_t peer_count;
149
150 } Etm_stats = {
151 /* read counters */
152 { "read_ack", FMD_TYPE_UINT64, "ACKs read" },
153 { "read_bytes", FMD_TYPE_UINT64, "Bytes read" },
154 { "read_msg", FMD_TYPE_UINT64, "Messages read" },
155 { "post_filter", FMD_TYPE_UINT64, "Drops by post_filter" },
156 /* write counters */
157 { "write_ack", FMD_TYPE_UINT64, "ACKs sent" },
158 { "write_bytes", FMD_TYPE_UINT64, "Bytes sent" },
159 { "write_msg", FMD_TYPE_UINT64, "Messages sent" },
160 { "send_filter", FMD_TYPE_UINT64, "Drops by send_filter" },
161 /* ETM error counters */
162 { "error_protocol", FMD_TYPE_UINT64, "ETM protocol errors" },
163 { "error_drop_read", FMD_TYPE_UINT64, "Dropped read messages" },
164 { "error_read", FMD_TYPE_UINT64, "Read I/O errors" },
165 { "error_read_badhdr", FMD_TYPE_UINT64, "Bad headers read" },
166 { "error_write", FMD_TYPE_UINT64, "Write I/O errors" },
167 { "error_send_filter", FMD_TYPE_UINT64, "Send filter errors" },
168 { "error_post_filter", FMD_TYPE_UINT64, "Post filter errors" },
169 /* ETM Misc */
170 { "peer_count", FMD_TYPE_UINT64, "Number of peers initialized" },
171 };
172
173 /*
174 * ETM Private functions
175 */
176
177 /*
178 * Hex dump for debug.
179 */
180 static void
etm_hex_dump(fmd_hdl_t * hdl,void * buf,size_t buflen,int direction)181 etm_hex_dump(fmd_hdl_t *hdl, void *buf, size_t buflen, int direction)
182 {
183 int i, j, k;
184 int16_t *c;
185
186 if (Etm_dump == 0)
187 return;
188
189 j = buflen / 16; /* Number of complete 8-column rows */
190 k = buflen % 16; /* Is there a last (non-8-column) row? */
191
192 if (direction)
193 fmd_hdl_debug(hdl, "--- WRITE Message Dump ---");
194 else
195 fmd_hdl_debug(hdl, "--- READ Message Dump ---");
196
197 fmd_hdl_debug(hdl, " Displaying %d bytes", buflen);
198
199 /* Dump the complete 8-column rows */
200 for (i = 0; i < j; i++) {
201 c = (int16_t *)buf + (i * 8);
202 fmd_hdl_debug(hdl, "%3d: %4x %4x %4x %4x %4x %4x %4x %4x", i,
203 *(c+0), *(c+1), *(c+2), *(c+3),
204 *(c+4), *(c+5), *(c+6), *(c+7));
205 }
206
207 /* Dump the last (incomplete) row */
208 c = (int16_t *)buf + (i * 8);
209 switch (k) {
210 case 4:
211 fmd_hdl_debug(hdl, "%3d: %4x %4x", i, *(c+0), *(c+1));
212 break;
213 case 8:
214 fmd_hdl_debug(hdl, "%3d: %4x %4x %4x %4x", i, *(c+0), *(c+1),
215 *(c+2), *(c+3));
216 break;
217 case 12:
218 fmd_hdl_debug(hdl, "%3d: %4x %4x %4x %4x %4x %4x", i, *(c+0),
219 *(c+1), *(c+2), *(c+3), *(c+4), *(c+5));
220 break;
221 }
222
223 fmd_hdl_debug(hdl, "--- End Dump ---");
224 }
225
226 /*
227 * Provide the length of a message based on the data in the given ETM header.
228 */
229 static size_t
etm_get_msglen(void * buf)230 etm_get_msglen(void *buf)
231 {
232 etm_proto_hdr_t *hp = (etm_proto_hdr_t *)buf;
233
234 return (ntohl(hp->hdr_msglen));
235 }
236
237 /*
238 * Check the contents of the ETM header for errors.
239 * Return the header type (hdr_type).
240 */
241 static int
etm_check_hdr(fmd_hdl_t * hdl,etm_epmap_t * mp,void * buf)242 etm_check_hdr(fmd_hdl_t *hdl, etm_epmap_t *mp, void *buf)
243 {
244 etm_proto_hdr_t *hp = (etm_proto_hdr_t *)buf;
245
246 if (bcmp(hp->hdr_delim, ETM_DELIM, ETM_DELIMLEN) != 0) {
247 fmd_hdl_debug(hdl, "Bad delimiter in ETM header from %s "
248 ": 0x%x\n", mp->epm_ep_str, hp->hdr_delim);
249 return (ETM_HDR_INVALID);
250 }
251
252 if ((hp->hdr_type == ETM_HDR_C_HELLO) ||
253 (hp->hdr_type == ETM_HDR_S_HELLO)) {
254 /* Until version is negotiated, other fields may be wrong */
255 return (hp->hdr_type);
256 }
257
258 if (hp->hdr_ver != mp->epm_ver) {
259 fmd_hdl_debug(hdl, "Bad version in ETM header from %s : 0x%x\n",
260 mp->epm_ep_str, hp->hdr_ver);
261 return (ETM_HDR_BADVERSION);
262 }
263
264 if ((hp->hdr_type == ETM_HDR_TYPE_TOO_LOW) ||
265 (hp->hdr_type >= ETM_HDR_TYPE_TOO_HIGH)) {
266 fmd_hdl_debug(hdl, "Bad type in ETM header from %s : 0x%x\n",
267 mp->epm_ep_str, hp->hdr_type);
268 return (ETM_HDR_BADTYPE);
269 }
270
271 return (hp->hdr_type);
272 }
273
274 /*
275 * Create an ETM header of a given type in the given buffer.
276 * Return length of header.
277 */
278 static size_t
etm_create_hdr(void * buf,uint8_t ver,uint8_t type,uint32_t msglen)279 etm_create_hdr(void *buf, uint8_t ver, uint8_t type, uint32_t msglen)
280 {
281 etm_proto_hdr_t *hp = (etm_proto_hdr_t *)buf;
282
283 bcopy(ETM_DELIM, hp->hdr_delim, ETM_DELIMLEN);
284 hp->hdr_ver = ver;
285 hp->hdr_type = type;
286 hp->hdr_msglen = htonl(msglen);
287
288 return (ETM_HDRLEN);
289 }
290
291 /*
292 * Convert message bytes to nvlist and post to fmd.
293 * Return zero for success, non-zero for failure.
294 *
295 * Note : nvl is free'd by fmd.
296 */
297 static int
etm_post_msg(fmd_hdl_t * hdl,etm_epmap_t * mp,void * buf,size_t buflen)298 etm_post_msg(fmd_hdl_t *hdl, etm_epmap_t *mp, void *buf, size_t buflen)
299 {
300 nvlist_t *nvl;
301 int rv;
302
303 if (nvlist_unpack((char *)buf, buflen, &nvl, 0)) {
304 fmd_hdl_error(hdl, "failed to unpack message");
305 return (1);
306 }
307
308 rv = etm_xport_post_filter(hdl, nvl, mp->epm_ep_str);
309 if (rv == ETM_XPORT_FILTER_DROP) {
310 fmd_hdl_debug(hdl, "post_filter dropped event");
311 INCRSTAT(Etm_stats.post_filter.fmds_value.ui64);
312 nvlist_free(nvl);
313 return (0);
314 } else if (rv == ETM_XPORT_FILTER_ERROR) {
315 fmd_hdl_debug(hdl, "post_filter error : %s", strerror(errno));
316 INCRSTAT(Etm_stats.error_post_filter.fmds_value.ui64);
317 /* Still post event */
318 }
319
320 (void) pthread_mutex_lock(&mp->epm_lock);
321 (void) pthread_mutex_lock(&Etm_mod_lock);
322 if (!Etm_exit) {
323 (void) pthread_mutex_unlock(&Etm_mod_lock);
324 if (mp->epm_qstat == Q_OPEN) {
325 fmd_xprt_post(hdl, mp->epm_xprthdl, nvl, 0);
326 rv = 0;
327 } else if (mp->epm_qstat == Q_SUSPENDED) {
328 fmd_xprt_resume(hdl, mp->epm_xprthdl);
329 if (mp->epm_timer_in_use) {
330 fmd_timer_remove(hdl, mp->epm_timer_id);
331 mp->epm_timer_in_use = 0;
332 }
333 mp->epm_qstat = Q_OPEN;
334 fmd_hdl_debug(hdl, "queue resumed for %s",
335 mp->epm_ep_str);
336 fmd_xprt_post(hdl, mp->epm_xprthdl, nvl, 0);
337 rv = 0;
338 } else {
339 fmd_hdl_debug(hdl, "unable to post message, qstat = %d",
340 mp->epm_qstat);
341 nvlist_free(nvl);
342 /* Remote peer will attempt to resend event */
343 rv = 2;
344 }
345 } else {
346 (void) pthread_mutex_unlock(&Etm_mod_lock);
347 fmd_hdl_debug(hdl, "unable to post message, module exiting");
348 nvlist_free(nvl);
349 /* Remote peer will attempt to resend event */
350 rv = 3;
351 }
352
353 (void) pthread_mutex_unlock(&mp->epm_lock);
354
355 return (rv);
356 }
357
358 /*
359 * Handle the startup handshake to the server. The client always initiates
360 * the startup handshake. In the following sequence, we are the client and
361 * the remote endpoint is the server.
362 *
363 * Client sends C_HELLO and transitions to Q_INIT_PENDING state.
364 * Server sends S_HELLO and transitions to Q_INIT_PENDING state.
365 * Client sends ACK and transitions to Q_OPEN state.
366 * Server receives ACK and transitions to Q_OPEN state.
367 *
368 * Return 0 for success, nonzero for failure.
369 */
370 static int
etm_handle_startup(fmd_hdl_t * hdl,etm_epmap_t * mp)371 etm_handle_startup(fmd_hdl_t *hdl, etm_epmap_t *mp)
372 {
373 etm_proto_hdr_t *hp;
374 size_t hdrlen = ETM_HDRLEN;
375 int hdrstat;
376 char hbuf[ETM_HDRLEN];
377
378 if ((mp->epm_oconn = etm_xport_open(hdl, mp->epm_tlhdl)) == NULL)
379 return (1);
380
381 mp->epm_cstat = C_OPEN;
382
383 hdrlen = etm_create_hdr(hbuf, mp->epm_ver, ETM_HDR_C_HELLO, 0);
384
385 if ((etm_xport_write(hdl, mp->epm_oconn, Rw_timeout, hbuf,
386 hdrlen)) != hdrlen) {
387 fmd_hdl_error(hdl, "Failed to write C_HELLO to %s",
388 mp->epm_ep_str);
389 return (2);
390 }
391
392 mp->epm_qstat = Q_INIT_PENDING;
393
394 if ((etm_xport_read(hdl, mp->epm_oconn, Rw_timeout, hbuf,
395 hdrlen)) != hdrlen) {
396 fmd_hdl_error(hdl, "Failed to read S_HELLO from %s",
397 mp->epm_ep_str);
398 return (3);
399 }
400
401 hdrstat = etm_check_hdr(hdl, mp, hbuf);
402
403 if (hdrstat != ETM_HDR_S_HELLO) {
404 fmd_hdl_error(hdl, "Protocol error, did not receive S_HELLO "
405 "from %s", mp->epm_ep_str);
406 return (4);
407 }
408
409 /*
410 * Get version from the server.
411 * Currently, only one version is supported.
412 */
413 hp = (etm_proto_hdr_t *)(void *)hbuf;
414 if (hp->hdr_ver != ETM_PROTO_V1) {
415 fmd_hdl_error(hdl, "Unable to use same version as %s : %d",
416 mp->epm_ep_str, hp->hdr_ver);
417 return (5);
418 }
419 mp->epm_ver = hp->hdr_ver;
420
421 hdrlen = etm_create_hdr(hbuf, mp->epm_ver, ETM_HDR_ACK, 0);
422
423 if ((etm_xport_write(hdl, mp->epm_oconn, Rw_timeout, hbuf,
424 hdrlen)) != hdrlen) {
425 fmd_hdl_error(hdl, "Failed to write ACK for S_HELLO to %s",
426 mp->epm_ep_str);
427 return (6);
428 }
429
430 /*
431 * Call fmd_xprt_open and fmd_xprt_setspecific with
432 * Etm_mod_lock held to avoid race with etm_send thread.
433 */
434 (void) pthread_mutex_lock(&Etm_mod_lock);
435 if ((mp->epm_xprthdl = fmd_xprt_open(hdl, mp->epm_xprtflags,
436 mp->epm_ep_nvl, NULL)) == NULL) {
437 fmd_hdl_abort(hdl, "Failed to init xprthdl for %s",
438 mp->epm_ep_str);
439 }
440 fmd_xprt_setspecific(hdl, mp->epm_xprthdl, mp);
441 (void) pthread_mutex_unlock(&Etm_mod_lock);
442
443 mp->epm_qstat = Q_OPEN;
444 fmd_hdl_debug(hdl, "queue open for %s", mp->epm_ep_str);
445
446 return (0);
447 }
448
449 /*
450 * Open a connection to the peer, send a SHUTDOWN message,
451 * and close the connection.
452 */
453 static void
etm_send_shutdown(fmd_hdl_t * hdl,etm_epmap_t * mp)454 etm_send_shutdown(fmd_hdl_t *hdl, etm_epmap_t *mp)
455 {
456 size_t hdrlen = ETM_HDRLEN;
457 char hbuf[ETM_HDRLEN];
458
459 if ((mp->epm_oconn = etm_xport_open(hdl, mp->epm_tlhdl)) == NULL)
460 return;
461
462 hdrlen = etm_create_hdr(hbuf, mp->epm_ver, ETM_HDR_SHUTDOWN, 0);
463
464 (void) etm_xport_write(hdl, mp->epm_oconn, Rw_timeout, hbuf, hdrlen);
465
466 (void) etm_xport_close(hdl, mp->epm_oconn);
467 mp->epm_oconn = NULL;
468 }
469
470 /*
471 * Alloc a nvlist and add a string for the endpoint.
472 * Return zero for success, non-zero for failure.
473 */
474 static int
etm_get_ep_nvl(fmd_hdl_t * hdl,etm_epmap_t * mp)475 etm_get_ep_nvl(fmd_hdl_t *hdl, etm_epmap_t *mp)
476 {
477 /*
478 * Cannot use nvlist_xalloc(3NVPAIR) due to a recursive mutex situation
479 * in fmd when this nvlist_t is free'd.
480 */
481 (void) nvlist_alloc(&mp->epm_ep_nvl, NV_UNIQUE_NAME, 0);
482
483 if (nvlist_add_string(mp->epm_ep_nvl, "domain-id", mp->epm_ep_str)) {
484 fmd_hdl_error(hdl, "failed to add domain-id string to nvlist "
485 "for %s", mp->epm_ep_str);
486 nvlist_free(mp->epm_ep_nvl);
487 return (1);
488 }
489
490 return (0);
491 }
492
493 /*
494 * Free the nvlist for the endpoint_id string.
495 */
496 /*ARGSUSED*/
497 static void
etm_free_ep_nvl(fmd_hdl_t * hdl,etm_epmap_t * mp)498 etm_free_ep_nvl(fmd_hdl_t *hdl, etm_epmap_t *mp)
499 {
500 nvlist_free(mp->epm_ep_nvl);
501 }
502
503 /*
504 * Check for a duplicate endpoint/peer string.
505 */
506 /*ARGSUSED*/
507 static int
etm_check_dup_ep_str(fmd_hdl_t * hdl,char * epname)508 etm_check_dup_ep_str(fmd_hdl_t *hdl, char *epname)
509 {
510 etm_epmap_t *mp;
511
512 for (mp = Epmap_head; mp != NULL; mp = mp->epm_next)
513 if (strcmp(epname, mp->epm_ep_str) == 0)
514 return (1);
515
516 return (0);
517 }
518
519 /*
520 * Attempt to re-open a connection with the remote endpoint.
521 */
522 static void
etm_reconnect(fmd_hdl_t * hdl,etm_epmap_t * mp)523 etm_reconnect(fmd_hdl_t *hdl, etm_epmap_t *mp)
524 {
525 if ((mp->epm_reconn_end > 0) && (mp->epm_cstat == C_UNINITIALIZED)) {
526 if (gethrtime() < mp->epm_reconn_end) {
527 if ((mp->epm_oconn = etm_xport_open(hdl,
528 mp->epm_tlhdl)) == NULL) {
529 fmd_hdl_debug(hdl, "reconnect failed for %s",
530 mp->epm_ep_str);
531 mp->epm_timer_id = fmd_timer_install(hdl, mp,
532 NULL, Reconn_interval);
533 mp->epm_timer_in_use = 1;
534 } else {
535 fmd_hdl_debug(hdl, "reconnect success for %s",
536 mp->epm_ep_str);
537 mp->epm_reconn_end = 0;
538 mp->epm_cstat = C_OPEN;
539 }
540 } else {
541 fmd_hdl_error(hdl, "Reconnect timed out for %s\n",
542 mp->epm_ep_str);
543 mp->epm_reconn_end = 0;
544 mp->epm_cstat = C_TIMED_OUT;
545 }
546 }
547
548 if (mp->epm_cstat == C_OPEN) {
549 fmd_xprt_resume(hdl, mp->epm_xprthdl);
550 mp->epm_qstat = Q_OPEN;
551 fmd_hdl_debug(hdl, "queue resumed for %s", mp->epm_ep_str);
552 }
553 }
554
555 /*
556 * Suspend a given connection and setup for reconnection retries.
557 * Assume caller holds lock on epm_lock.
558 */
559 static void
etm_suspend_reconnect(fmd_hdl_t * hdl,etm_epmap_t * mp)560 etm_suspend_reconnect(fmd_hdl_t *hdl, etm_epmap_t *mp)
561 {
562 (void) pthread_mutex_lock(&Etm_mod_lock);
563 if (Etm_exit) {
564 (void) pthread_mutex_unlock(&Etm_mod_lock);
565 return;
566 }
567 (void) pthread_mutex_unlock(&Etm_mod_lock);
568
569 if (mp->epm_oconn != NULL) {
570 (void) etm_xport_close(hdl, mp->epm_oconn);
571 mp->epm_oconn = NULL;
572 }
573
574 mp->epm_reconn_end = gethrtime() + Reconn_timeout;
575 mp->epm_cstat = C_UNINITIALIZED;
576
577 if (mp->epm_xprthdl != NULL) {
578 fmd_xprt_suspend(hdl, mp->epm_xprthdl);
579 mp->epm_qstat = Q_SUSPENDED;
580 fmd_hdl_debug(hdl, "queue suspended for %s", mp->epm_ep_str);
581
582 if (mp->epm_timer_in_use == 0) {
583 mp->epm_timer_id = fmd_timer_install(hdl, mp, NULL,
584 Reconn_interval);
585 mp->epm_timer_in_use = 1;
586 }
587 }
588 }
589
590 /*
591 * Reinitialize the connection. The old fmd_xprt_t handle must be
592 * removed/closed first.
593 * Assume caller holds lock on epm_lock.
594 */
595 static void
etm_reinit(fmd_hdl_t * hdl,etm_epmap_t * mp)596 etm_reinit(fmd_hdl_t *hdl, etm_epmap_t *mp)
597 {
598 /*
599 * To avoid a deadlock, wait for etm_send to finish before
600 * calling fmd_xprt_close()
601 */
602 while (mp->epm_txbusy)
603 (void) pthread_cond_wait(&mp->epm_tx_cv, &mp->epm_lock);
604
605 if (mp->epm_xprthdl != NULL) {
606 fmd_xprt_close(hdl, mp->epm_xprthdl);
607 fmd_hdl_debug(hdl, "queue closed for %s", mp->epm_ep_str);
608 mp->epm_xprthdl = NULL;
609 /* mp->epm_ep_nvl is free'd in fmd_xprt_close */
610 mp->epm_ep_nvl = NULL;
611 }
612
613 if (mp->epm_timer_in_use) {
614 fmd_timer_remove(hdl, mp->epm_timer_id);
615 mp->epm_timer_in_use = 0;
616 }
617
618 if (mp->epm_oconn != NULL) {
619 (void) etm_xport_close(hdl, mp->epm_oconn);
620 mp->epm_oconn = NULL;
621 }
622
623 mp->epm_cstat = C_UNINITIALIZED;
624 mp->epm_qstat = Q_UNINITIALIZED;
625 }
626
627 /*
628 * Receive data from ETM transport layer.
629 * Note : This is not the fmdo_recv entry point.
630 *
631 */
632 static int
etm_recv(fmd_hdl_t * hdl,etm_xport_conn_t conn,etm_epmap_t * mp)633 etm_recv(fmd_hdl_t *hdl, etm_xport_conn_t conn, etm_epmap_t *mp)
634 {
635 size_t buflen, hdrlen;
636 void *buf;
637 char hbuf[ETM_HDRLEN];
638 int hdrstat, rv;
639
640 hdrlen = ETM_HDRLEN;
641
642 if ((etm_xport_read(hdl, conn, Rw_timeout, hbuf, hdrlen)) != hdrlen) {
643 fmd_hdl_debug(hdl, "failed to read header from %s",
644 mp->epm_ep_str);
645 INCRSTAT(Etm_stats.error_read.fmds_value.ui64);
646 return (EIO);
647 }
648
649 hdrstat = etm_check_hdr(hdl, mp, hbuf);
650
651 switch (hdrstat) {
652 case ETM_HDR_INVALID:
653 (void) pthread_mutex_lock(&mp->epm_lock);
654 if (mp->epm_cstat == C_OPEN)
655 mp->epm_cstat = C_CLOSED;
656 (void) pthread_mutex_unlock(&mp->epm_lock);
657
658 INCRSTAT(Etm_stats.error_read_badhdr.fmds_value.ui64);
659 rv = ECANCELED;
660 break;
661
662 case ETM_HDR_BADTYPE:
663 case ETM_HDR_BADVERSION:
664 hdrlen = etm_create_hdr(hbuf, mp->epm_ver, ETM_HDR_NAK, 0);
665
666 if ((etm_xport_write(hdl, conn, Rw_timeout, hbuf,
667 hdrlen)) != hdrlen) {
668 fmd_hdl_debug(hdl, "failed to write NAK to %s",
669 mp->epm_ep_str);
670 INCRSTAT(Etm_stats.error_write.fmds_value.ui64);
671 return (EIO);
672 }
673
674 (void) pthread_mutex_lock(&mp->epm_lock);
675 mp->epm_cstat = C_LIMBO;
676 (void) pthread_mutex_unlock(&mp->epm_lock);
677
678 INCRSTAT(Etm_stats.error_read_badhdr.fmds_value.ui64);
679 rv = ENOTSUP;
680 break;
681
682 case ETM_HDR_C_HELLO:
683 /* Client is initiating a startup handshake */
684 (void) pthread_mutex_lock(&mp->epm_lock);
685 etm_reinit(hdl, mp);
686 mp->epm_qstat = Q_INIT_PENDING;
687 (void) pthread_mutex_unlock(&mp->epm_lock);
688
689 hdrlen = etm_create_hdr(hbuf, mp->epm_ver, ETM_HDR_S_HELLO, 0);
690
691 if ((etm_xport_write(hdl, conn, Rw_timeout, hbuf,
692 hdrlen)) != hdrlen) {
693 fmd_hdl_debug(hdl, "failed to write S_HELLO to %s",
694 mp->epm_ep_str);
695 INCRSTAT(Etm_stats.error_write.fmds_value.ui64);
696 return (EIO);
697 }
698
699 rv = 0;
700 break;
701
702 case ETM_HDR_ACK:
703 (void) pthread_mutex_lock(&mp->epm_lock);
704 if (mp->epm_qstat == Q_INIT_PENDING) {
705 /* This is client's ACK from startup handshake */
706 /* mp->epm_ep_nvl is free'd in fmd_xprt_close */
707 if (mp->epm_ep_nvl == NULL)
708 (void) etm_get_ep_nvl(hdl, mp);
709
710 /*
711 * Call fmd_xprt_open and fmd_xprt_setspecific with
712 * Etm_mod_lock held to avoid race with etm_send thread.
713 */
714 (void) pthread_mutex_lock(&Etm_mod_lock);
715 if ((mp->epm_xprthdl = fmd_xprt_open(hdl,
716 mp->epm_xprtflags, mp->epm_ep_nvl, NULL)) == NULL) {
717 fmd_hdl_abort(hdl, "Failed to init xprthdl "
718 "for %s", mp->epm_ep_str);
719 }
720 fmd_xprt_setspecific(hdl, mp->epm_xprthdl, mp);
721 (void) pthread_mutex_unlock(&Etm_mod_lock);
722
723 mp->epm_qstat = Q_OPEN;
724 (void) pthread_mutex_unlock(&mp->epm_lock);
725 fmd_hdl_debug(hdl, "queue open for %s",
726 mp->epm_ep_str);
727 } else {
728 (void) pthread_mutex_unlock(&mp->epm_lock);
729 fmd_hdl_debug(hdl, "protocol error, not expecting ACK "
730 "from %s\n", mp->epm_ep_str);
731 INCRSTAT(Etm_stats.error_protocol.fmds_value.ui64);
732 }
733
734 rv = 0;
735 break;
736
737 case ETM_HDR_SHUTDOWN:
738 fmd_hdl_debug(hdl, "received shutdown from %s",
739 mp->epm_ep_str);
740
741 (void) pthread_mutex_lock(&mp->epm_lock);
742
743 etm_reinit(hdl, mp);
744
745 if (IS_CLIENT(mp)) {
746 /*
747 * A server shutdown is considered to be temporary.
748 * Prepare for reconnection.
749 */
750 mp->epm_timer_id = fmd_timer_install(hdl, mp, NULL,
751 Reconn_interval);
752
753 mp->epm_timer_in_use = 1;
754 }
755
756 (void) pthread_mutex_unlock(&mp->epm_lock);
757
758 rv = ECANCELED;
759 break;
760
761 case ETM_HDR_MSG:
762 (void) pthread_mutex_lock(&mp->epm_lock);
763 if (mp->epm_qstat == Q_UNINITIALIZED) {
764 /* Peer (client) is unaware that we've restarted */
765 (void) pthread_mutex_unlock(&mp->epm_lock);
766 hdrlen = etm_create_hdr(hbuf, mp->epm_ver,
767 ETM_HDR_S_RESTART, 0);
768
769 if ((etm_xport_write(hdl, conn, Rw_timeout, hbuf,
770 hdrlen)) != hdrlen) {
771 fmd_hdl_debug(hdl, "failed to write S_RESTART "
772 "to %s", mp->epm_ep_str);
773 INCRSTAT(Etm_stats.error_write.fmds_value.ui64);
774 return (EIO);
775 }
776
777 return (ECANCELED);
778 }
779 (void) pthread_mutex_unlock(&mp->epm_lock);
780
781 buflen = etm_get_msglen(hbuf);
782 ALLOC_BUF(hdl, buf, buflen);
783
784 if (etm_xport_read(hdl, conn, Rw_timeout, buf,
785 buflen) != buflen) {
786 fmd_hdl_debug(hdl, "failed to read message from %s",
787 mp->epm_ep_str);
788 FREE_BUF(hdl, buf, buflen);
789 INCRSTAT(Etm_stats.error_read.fmds_value.ui64);
790 return (EIO);
791 }
792
793 INCRSTAT(Etm_stats.read_msg.fmds_value.ui64);
794 ADDSTAT(Etm_stats.read_bytes.fmds_value.ui64, buflen);
795
796 etm_hex_dump(hdl, buf, buflen, 0);
797
798 if (etm_post_msg(hdl, mp, buf, buflen)) {
799 INCRSTAT(Etm_stats.error_drop_read.fmds_value.ui64);
800 FREE_BUF(hdl, buf, buflen);
801 return (EIO);
802 }
803
804 FREE_BUF(hdl, buf, buflen);
805
806 hdrlen = etm_create_hdr(hbuf, mp->epm_ver, ETM_HDR_ACK, 0);
807
808 if ((etm_xport_write(hdl, conn, Rw_timeout, hbuf,
809 hdrlen)) != hdrlen) {
810 fmd_hdl_debug(hdl, "failed to write ACK to %s",
811 mp->epm_ep_str);
812 INCRSTAT(Etm_stats.error_write.fmds_value.ui64);
813 return (EIO);
814 }
815
816 INCRSTAT(Etm_stats.write_ack.fmds_value.ui64);
817
818 /*
819 * If we got this far and the current state of the
820 * outbound/sending connection is TIMED_OUT or
821 * LIMBO, then we should reinitialize it.
822 */
823 (void) pthread_mutex_lock(&mp->epm_lock);
824 if (mp->epm_cstat == C_TIMED_OUT ||
825 mp->epm_cstat == C_LIMBO) {
826 if (mp->epm_oconn != NULL) {
827 (void) etm_xport_close(hdl, mp->epm_oconn);
828 mp->epm_oconn = NULL;
829 }
830 mp->epm_cstat = C_UNINITIALIZED;
831 fmd_xprt_resume(hdl, mp->epm_xprthdl);
832 if (mp->epm_timer_in_use) {
833 fmd_timer_remove(hdl, mp->epm_timer_id);
834 mp->epm_timer_in_use = 0;
835 }
836 mp->epm_qstat = Q_OPEN;
837 fmd_hdl_debug(hdl, "queue resumed for %s",
838 mp->epm_ep_str);
839 }
840 (void) pthread_mutex_unlock(&mp->epm_lock);
841
842 rv = 0;
843 break;
844
845 default:
846 fmd_hdl_debug(hdl, "protocol error, unexpected header "
847 "from %s : %d", mp->epm_ep_str, hdrstat);
848 INCRSTAT(Etm_stats.error_protocol.fmds_value.ui64);
849 rv = 0;
850 }
851
852 return (rv);
853 }
854
855 /*
856 * ETM transport layer callback function.
857 * The transport layer calls this function to :
858 * (a) pass an incoming message (flag == ETM_CBFLAG_RECV)
859 * (b) tell us to reinitialize the connection (flag == ETM_CBFLAG_REINIT)
860 */
861 static int
etm_cb_func(fmd_hdl_t * hdl,etm_xport_conn_t conn,etm_cb_flag_t flag,void * arg)862 etm_cb_func(fmd_hdl_t *hdl, etm_xport_conn_t conn, etm_cb_flag_t flag,
863 void *arg)
864 {
865 etm_epmap_t *mp = (etm_epmap_t *)arg;
866 int rv = 0;
867
868 (void) pthread_mutex_lock(&Etm_mod_lock);
869 if (Etm_exit) {
870 (void) pthread_mutex_unlock(&Etm_mod_lock);
871 return (ECANCELED);
872 }
873 (void) pthread_mutex_unlock(&Etm_mod_lock);
874
875 switch (flag) {
876 case ETM_CBFLAG_RECV:
877 rv = etm_recv(hdl, conn, mp);
878 break;
879 case ETM_CBFLAG_REINIT:
880 (void) pthread_mutex_lock(&mp->epm_lock);
881 etm_reinit(hdl, mp);
882 etm_send_shutdown(hdl, mp);
883 (void) pthread_mutex_unlock(&mp->epm_lock);
884 /*
885 * Return ECANCELED so the transport layer will close the
886 * server connection. The transport layer is responsible for
887 * reestablishing this connection (should a connection request
888 * arrive from the peer).
889 */
890 rv = ECANCELED;
891 break;
892 default:
893 fmd_hdl_debug(hdl, "Unknown callback flag : 0x%x", flag);
894 rv = ENOTSUP;
895 }
896
897 return (rv);
898 }
899
900 /*
901 * Allocate and initialize an etm_epmap_t struct for the given endpoint
902 * name string.
903 */
904 static void
etm_init_epmap(fmd_hdl_t * hdl,char * epname,int flags)905 etm_init_epmap(fmd_hdl_t *hdl, char *epname, int flags)
906 {
907 etm_epmap_t *newmap;
908
909 if (etm_check_dup_ep_str(hdl, epname)) {
910 fmd_hdl_debug(hdl, "skipping duplicate peer : %s", epname);
911 return;
912 }
913
914 newmap = fmd_hdl_zalloc(hdl, sizeof (etm_epmap_t), FMD_SLEEP);
915 newmap->epm_ep_str = fmd_hdl_strdup(hdl, epname, FMD_SLEEP);
916 newmap->epm_xprtflags = flags;
917 newmap->epm_cstat = C_UNINITIALIZED;
918 newmap->epm_qstat = Q_UNINITIALIZED;
919 newmap->epm_ver = ETM_PROTO_V1; /* Currently support one proto ver */
920 newmap->epm_txbusy = 0;
921
922 (void) pthread_mutex_init(&newmap->epm_lock, NULL);
923 (void) pthread_cond_init(&newmap->epm_tx_cv, NULL);
924
925 if (etm_get_ep_nvl(hdl, newmap)) {
926 fmd_hdl_strfree(hdl, newmap->epm_ep_str);
927 fmd_hdl_free(hdl, newmap, sizeof (etm_epmap_t));
928 return;
929 }
930
931 (void) pthread_mutex_lock(&newmap->epm_lock);
932
933 if ((newmap->epm_tlhdl = etm_xport_init(hdl, newmap->epm_ep_str,
934 etm_cb_func, newmap)) == NULL) {
935 fmd_hdl_debug(hdl, "failed to init tlhdl for %s\n",
936 newmap->epm_ep_str);
937 etm_free_ep_nvl(hdl, newmap);
938 (void) pthread_mutex_unlock(&newmap->epm_lock);
939 (void) pthread_mutex_destroy(&newmap->epm_lock);
940 fmd_hdl_strfree(hdl, newmap->epm_ep_str);
941 fmd_hdl_free(hdl, newmap, sizeof (etm_epmap_t));
942 return;
943 }
944
945 if (IS_CLIENT(newmap)) {
946 if (etm_handle_startup(hdl, newmap)) {
947 /*
948 * For whatever reason, we could not complete the
949 * startup handshake with the server. Set the timer
950 * and try again.
951 */
952 if (newmap->epm_oconn != NULL) {
953 (void) etm_xport_close(hdl, newmap->epm_oconn);
954 newmap->epm_oconn = NULL;
955 }
956 newmap->epm_cstat = C_UNINITIALIZED;
957 newmap->epm_qstat = Q_UNINITIALIZED;
958 newmap->epm_timer_id = fmd_timer_install(hdl, newmap,
959 NULL, Reconn_interval);
960 newmap->epm_timer_in_use = 1;
961 }
962 } else {
963 /*
964 * We may be restarting after a crash. If so, the client
965 * may be unaware of this.
966 */
967 etm_send_shutdown(hdl, newmap);
968 }
969
970 /* Add this transport instance handle to the list */
971 newmap->epm_next = Epmap_head;
972 Epmap_head = newmap;
973
974 (void) pthread_mutex_unlock(&newmap->epm_lock);
975
976 INCRSTAT(Etm_stats.peer_count.fmds_value.ui64);
977 }
978
979 /*
980 * Parse the given property list string and call etm_init_epmap
981 * for each endpoint.
982 */
983 static void
etm_create_epmaps(fmd_hdl_t * hdl,char * eplist,int flags)984 etm_create_epmaps(fmd_hdl_t *hdl, char *eplist, int flags)
985 {
986 char *epstr, *ep, *prefix, *lasts, *numstr;
987 char epname[MAXPATHLEN];
988 size_t slen, nlen;
989 int beg, end, i;
990
991 if (eplist == NULL)
992 return;
993 /*
994 * Create a copy of eplist for parsing.
995 * strtok/strtok_r(3C) will insert null chars to the string.
996 * Therefore, fmd_hdl_strdup/fmd_hdl_strfree cannot be used.
997 */
998 slen = strlen(eplist);
999 epstr = fmd_hdl_zalloc(hdl, slen + 1, FMD_SLEEP);
1000 (void) strcpy(epstr, eplist);
1001
1002 /*
1003 * The following are supported for the "client_list" and
1004 * "server_list" properties :
1005 *
1006 * A space-separated list of endpoints.
1007 * "dev:///dom0 dev:///dom1 dev:///dom2"
1008 *
1009 * An array syntax for a range of instances.
1010 * "dev:///dom[0:2]"
1011 *
1012 * A combination of both.
1013 * "dev:///dom0 dev:///dom[1:2]"
1014 */
1015 ep = strtok_r(epstr, " ", &lasts);
1016 while (ep != NULL) {
1017 if (strchr(ep, '[') != NULL) {
1018 /*
1019 * This string is using array syntax.
1020 * Check the string for correct syntax.
1021 */
1022 if ((strchr(ep, ':') == NULL) ||
1023 (strchr(ep, ']') == NULL)) {
1024 fmd_hdl_error(hdl, "Syntax error in property "
1025 "that includes : %s\n", ep);
1026 ep = strtok_r(NULL, " ", &lasts);
1027 continue;
1028 }
1029
1030 /* expand the array syntax */
1031 prefix = strtok(ep, "[");
1032
1033 numstr = strtok(NULL, ":");
1034 if ((numstr == NULL) || (!isdigit(*numstr))) {
1035 fmd_hdl_error(hdl, "Syntax error in property "
1036 "that includes : %s[\n", prefix);
1037 ep = strtok_r(NULL, " ", &lasts);
1038 continue;
1039 }
1040 beg = atoi(numstr);
1041
1042 numstr = strtok(NULL, "]");
1043 if ((numstr == NULL) || (!isdigit(*numstr))) {
1044 fmd_hdl_error(hdl, "Syntax error in property "
1045 "that includes : %s[\n", prefix);
1046 ep = strtok_r(NULL, " ", &lasts);
1047 continue;
1048 }
1049 end = atoi(numstr);
1050
1051 nlen = strlen(prefix) + ETM_EP_INST_MAX;
1052
1053 if (nlen > MAXPATHLEN) {
1054 fmd_hdl_error(hdl, "Endpoint prop string "
1055 "exceeds MAXPATHLEN\n");
1056 ep = strtok_r(NULL, " ", &lasts);
1057 continue;
1058 }
1059
1060 for (i = beg; i <= end; i++) {
1061 bzero(epname, MAXPATHLEN);
1062 (void) snprintf(epname, nlen, "%s%d",
1063 prefix, i);
1064 etm_init_epmap(hdl, epname, flags);
1065 }
1066 } else {
1067 etm_init_epmap(hdl, ep, flags);
1068 }
1069
1070 ep = strtok_r(NULL, " ", &lasts);
1071 }
1072
1073 fmd_hdl_free(hdl, epstr, slen + 1);
1074 }
1075
1076 /*
1077 * Free the transport infrastructure for an endpoint.
1078 */
1079 static void
etm_free_epmap(fmd_hdl_t * hdl,etm_epmap_t * mp)1080 etm_free_epmap(fmd_hdl_t *hdl, etm_epmap_t *mp)
1081 {
1082 size_t hdrlen;
1083 char hbuf[ETM_HDRLEN];
1084
1085 (void) pthread_mutex_lock(&mp->epm_lock);
1086
1087 /*
1088 * If an etm_send thread is in progress, wait for it to finish.
1089 * The etm_recv thread is managed by the transport layer and will
1090 * be destroyed with etm_xport_fini().
1091 */
1092 while (mp->epm_txbusy)
1093 (void) pthread_cond_wait(&mp->epm_tx_cv, &mp->epm_lock);
1094
1095 if (mp->epm_timer_in_use)
1096 fmd_timer_remove(hdl, mp->epm_timer_id);
1097
1098 if (mp->epm_oconn != NULL) {
1099 hdrlen = etm_create_hdr(hbuf, mp->epm_ver,
1100 ETM_HDR_SHUTDOWN, 0);
1101 (void) etm_xport_write(hdl, mp->epm_oconn, Rw_timeout, hbuf,
1102 hdrlen);
1103 (void) etm_xport_close(hdl, mp->epm_oconn);
1104 mp->epm_oconn = NULL;
1105 }
1106
1107 if (mp->epm_xprthdl != NULL) {
1108 fmd_xprt_close(hdl, mp->epm_xprthdl);
1109 /* mp->epm_ep_nvl is free'd in fmd_xprt_close */
1110 mp->epm_ep_nvl = NULL;
1111 }
1112
1113 if (mp->epm_ep_nvl != NULL)
1114 etm_free_ep_nvl(hdl, mp);
1115
1116 if (mp->epm_tlhdl != NULL)
1117 (void) etm_xport_fini(hdl, mp->epm_tlhdl);
1118
1119 (void) pthread_mutex_unlock(&mp->epm_lock);
1120 (void) pthread_mutex_destroy(&mp->epm_lock);
1121 fmd_hdl_strfree(hdl, mp->epm_ep_str);
1122 fmd_hdl_free(hdl, mp, sizeof (etm_epmap_t));
1123 DECRSTAT(Etm_stats.peer_count.fmds_value.ui64);
1124 }
1125
1126 /*
1127 * FMD entry points
1128 */
1129
1130 /*
1131 * FMD fmdo_send entry point.
1132 * Send an event to the remote endpoint and receive an ACK.
1133 */
1134 static int
etm_send(fmd_hdl_t * hdl,fmd_xprt_t * xprthdl,fmd_event_t * ep,nvlist_t * nvl)1135 etm_send(fmd_hdl_t *hdl, fmd_xprt_t *xprthdl, fmd_event_t *ep, nvlist_t *nvl)
1136 {
1137 etm_epmap_t *mp;
1138 nvlist_t *msgnvl;
1139 int hdrstat, rv, cnt = 0;
1140 char *buf, *nvbuf, *class;
1141 size_t nvsize, buflen, hdrlen;
1142 struct timespec tms;
1143
1144 (void) pthread_mutex_lock(&Etm_mod_lock);
1145 if (Etm_exit) {
1146 (void) pthread_mutex_unlock(&Etm_mod_lock);
1147 return (FMD_SEND_RETRY);
1148 }
1149 (void) pthread_mutex_unlock(&Etm_mod_lock);
1150
1151 mp = fmd_xprt_getspecific(hdl, xprthdl);
1152
1153 for (;;) {
1154 if (pthread_mutex_trylock(&mp->epm_lock) == 0) {
1155 break;
1156 } else {
1157 /*
1158 * Another thread may be (1) trying to close this
1159 * fmd_xprt_t, or (2) posting an event to it.
1160 * If (1), don't want to spend too much time here.
1161 * If (2), allow it to finish and release epm_lock.
1162 */
1163 if (cnt++ < 10) {
1164 tms.tv_sec = 0;
1165 tms.tv_nsec = (cnt * 10000);
1166 (void) nanosleep(&tms, NULL);
1167
1168 } else {
1169 return (FMD_SEND_RETRY);
1170 }
1171 }
1172 }
1173
1174 mp->epm_txbusy++;
1175
1176 if (mp->epm_qstat == Q_UNINITIALIZED) {
1177 mp->epm_txbusy--;
1178 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1179 (void) pthread_mutex_unlock(&mp->epm_lock);
1180 return (FMD_SEND_FAILED);
1181 }
1182
1183 if (mp->epm_cstat == C_CLOSED) {
1184 etm_suspend_reconnect(hdl, mp);
1185 mp->epm_txbusy--;
1186 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1187 (void) pthread_mutex_unlock(&mp->epm_lock);
1188 return (FMD_SEND_RETRY);
1189 }
1190
1191 if (mp->epm_cstat == C_LIMBO) {
1192 if (mp->epm_oconn != NULL) {
1193 (void) etm_xport_close(hdl, mp->epm_oconn);
1194 mp->epm_oconn = NULL;
1195 }
1196
1197 fmd_xprt_suspend(hdl, xprthdl);
1198 mp->epm_qstat = Q_SUSPENDED;
1199 mp->epm_txbusy--;
1200 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1201 (void) pthread_mutex_unlock(&mp->epm_lock);
1202 fmd_hdl_debug(hdl, "queue suspended for %s", mp->epm_ep_str);
1203 return (FMD_SEND_RETRY);
1204 }
1205
1206 if (mp->epm_oconn == NULL) {
1207 if ((mp->epm_oconn = etm_xport_open(hdl, mp->epm_tlhdl))
1208 == NULL) {
1209 etm_suspend_reconnect(hdl, mp);
1210 mp->epm_txbusy--;
1211 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1212 (void) pthread_mutex_unlock(&mp->epm_lock);
1213 return (FMD_SEND_RETRY);
1214 } else {
1215 mp->epm_cstat = C_OPEN;
1216 }
1217 }
1218
1219 if (nvlist_lookup_string(nvl, FM_CLASS, &class) != 0)
1220 fmd_hdl_abort(hdl, "No class string in nvlist");
1221
1222 msgnvl = fmd_xprt_translate(hdl, xprthdl, ep);
1223 if (msgnvl == NULL) {
1224 mp->epm_txbusy--;
1225 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1226 (void) pthread_mutex_unlock(&mp->epm_lock);
1227 fmd_hdl_error(hdl, "Failed to translate event %p\n",
1228 (void *) ep);
1229 return (FMD_SEND_FAILED);
1230 }
1231
1232 rv = etm_xport_send_filter(hdl, msgnvl, mp->epm_ep_str);
1233 if (rv == ETM_XPORT_FILTER_DROP) {
1234 mp->epm_txbusy--;
1235 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1236 (void) pthread_mutex_unlock(&mp->epm_lock);
1237 fmd_hdl_debug(hdl, "send_filter dropped event");
1238 nvlist_free(msgnvl);
1239 INCRSTAT(Etm_stats.send_filter.fmds_value.ui64);
1240 return (FMD_SEND_SUCCESS);
1241 } else if (rv == ETM_XPORT_FILTER_ERROR) {
1242 fmd_hdl_debug(hdl, "send_filter error : %s", strerror(errno));
1243 INCRSTAT(Etm_stats.error_send_filter.fmds_value.ui64);
1244 /* Still send event */
1245 }
1246
1247 (void) pthread_mutex_unlock(&mp->epm_lock);
1248
1249 (void) nvlist_size(msgnvl, &nvsize, NV_ENCODE_XDR);
1250
1251 hdrlen = ETM_HDRLEN;
1252 buflen = nvsize + hdrlen;
1253
1254 ALLOC_BUF(hdl, buf, buflen);
1255
1256 nvbuf = buf + hdrlen;
1257
1258 (void) etm_create_hdr(buf, mp->epm_ver, ETM_HDR_MSG, nvsize);
1259
1260 if (rv = nvlist_pack(msgnvl, &nvbuf, &nvsize, NV_ENCODE_XDR, 0)) {
1261 (void) pthread_mutex_lock(&mp->epm_lock);
1262 mp->epm_txbusy--;
1263 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1264 (void) pthread_mutex_unlock(&mp->epm_lock);
1265 fmd_hdl_error(hdl, "Failed to pack event : %s\n", strerror(rv));
1266 nvlist_free(msgnvl);
1267 FREE_BUF(hdl, buf, buflen);
1268 return (FMD_SEND_FAILED);
1269 }
1270
1271 nvlist_free(msgnvl);
1272
1273 if (etm_xport_write(hdl, mp->epm_oconn, Rw_timeout, buf,
1274 buflen) != buflen) {
1275 fmd_hdl_debug(hdl, "failed to send message to %s",
1276 mp->epm_ep_str);
1277 (void) pthread_mutex_lock(&mp->epm_lock);
1278 etm_suspend_reconnect(hdl, mp);
1279 mp->epm_txbusy--;
1280 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1281 (void) pthread_mutex_unlock(&mp->epm_lock);
1282 FREE_BUF(hdl, buf, buflen);
1283 INCRSTAT(Etm_stats.error_write.fmds_value.ui64);
1284 return (FMD_SEND_RETRY);
1285 }
1286
1287 INCRSTAT(Etm_stats.write_msg.fmds_value.ui64);
1288 ADDSTAT(Etm_stats.write_bytes.fmds_value.ui64, nvsize);
1289
1290 etm_hex_dump(hdl, nvbuf, nvsize, 1);
1291
1292 if (etm_xport_read(hdl, mp->epm_oconn, Rw_timeout, buf,
1293 hdrlen) != hdrlen) {
1294 fmd_hdl_debug(hdl, "failed to read ACK from %s",
1295 mp->epm_ep_str);
1296 (void) pthread_mutex_lock(&mp->epm_lock);
1297 etm_suspend_reconnect(hdl, mp);
1298 mp->epm_txbusy--;
1299 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1300 (void) pthread_mutex_unlock(&mp->epm_lock);
1301 FREE_BUF(hdl, buf, buflen);
1302 INCRSTAT(Etm_stats.error_read.fmds_value.ui64);
1303 return (FMD_SEND_RETRY);
1304 }
1305
1306 hdrstat = etm_check_hdr(hdl, mp, buf);
1307 FREE_BUF(hdl, buf, buflen);
1308
1309 if (hdrstat == ETM_HDR_ACK) {
1310 INCRSTAT(Etm_stats.read_ack.fmds_value.ui64);
1311 } else {
1312 (void) pthread_mutex_lock(&mp->epm_lock);
1313
1314 (void) etm_xport_close(hdl, mp->epm_oconn);
1315 mp->epm_oconn = NULL;
1316
1317 if (hdrstat == ETM_HDR_NAK) {
1318 /* Peer received a bad value in the header */
1319 if (mp->epm_xprthdl != NULL) {
1320 mp->epm_cstat = C_LIMBO;
1321 fmd_xprt_suspend(hdl, xprthdl);
1322 mp->epm_qstat = Q_SUSPENDED;
1323 fmd_hdl_debug(hdl, "received NAK, queue "
1324 "suspended for %s", mp->epm_ep_str);
1325 }
1326
1327 rv = FMD_SEND_RETRY;
1328
1329 } else if (hdrstat == ETM_HDR_S_RESTART) {
1330 /* Server has restarted */
1331 mp->epm_cstat = C_CLOSED;
1332 mp->epm_qstat = Q_UNINITIALIZED;
1333 fmd_hdl_debug(hdl, "server %s restarted",
1334 mp->epm_ep_str);
1335 /*
1336 * Cannot call fmd_xprt_close here, so we'll do it
1337 * on the timeout thread.
1338 */
1339 if (mp->epm_timer_in_use == 0) {
1340 mp->epm_timer_id = fmd_timer_install(
1341 hdl, mp, NULL, 0);
1342 mp->epm_timer_in_use = 1;
1343 }
1344
1345 /*
1346 * fault.* or list.* events will be replayed if a
1347 * transport is opened with the same auth.
1348 * Other events will be discarded.
1349 */
1350 rv = FMD_SEND_FAILED;
1351
1352 } else {
1353 mp->epm_cstat = C_CLOSED;
1354 fmd_hdl_debug(hdl, "bad ACK from %s", mp->epm_ep_str);
1355
1356 rv = FMD_SEND_RETRY;
1357 }
1358
1359 mp->epm_txbusy--;
1360
1361 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1362 (void) pthread_mutex_unlock(&mp->epm_lock);
1363
1364 INCRSTAT(Etm_stats.error_read_badhdr.fmds_value.ui64);
1365
1366 return (rv);
1367 }
1368
1369 (void) pthread_mutex_lock(&mp->epm_lock);
1370 mp->epm_txbusy--;
1371 (void) pthread_cond_broadcast(&mp->epm_tx_cv);
1372 (void) pthread_mutex_unlock(&mp->epm_lock);
1373
1374 return (FMD_SEND_SUCCESS);
1375 }
1376
1377 /*
1378 * FMD fmdo_timeout entry point..
1379 */
1380 /*ARGSUSED*/
1381 static void
etm_timeout(fmd_hdl_t * hdl,id_t id,void * data)1382 etm_timeout(fmd_hdl_t *hdl, id_t id, void *data)
1383 {
1384 etm_epmap_t *mp = (etm_epmap_t *)data;
1385
1386 (void) pthread_mutex_lock(&mp->epm_lock);
1387
1388 mp->epm_timer_in_use = 0;
1389
1390 if (mp->epm_qstat == Q_UNINITIALIZED) {
1391 /* Server has shutdown and we (client) need to reconnect */
1392 if (mp->epm_xprthdl != NULL) {
1393 fmd_xprt_close(hdl, mp->epm_xprthdl);
1394 fmd_hdl_debug(hdl, "queue closed for %s",
1395 mp->epm_ep_str);
1396 mp->epm_xprthdl = NULL;
1397 /* mp->epm_ep_nvl is free'd in fmd_xprt_close */
1398 mp->epm_ep_nvl = NULL;
1399 }
1400
1401 if (mp->epm_ep_nvl == NULL)
1402 (void) etm_get_ep_nvl(hdl, mp);
1403
1404 if (etm_handle_startup(hdl, mp)) {
1405 if (mp->epm_oconn != NULL) {
1406 (void) etm_xport_close(hdl, mp->epm_oconn);
1407 mp->epm_oconn = NULL;
1408 }
1409 mp->epm_cstat = C_UNINITIALIZED;
1410 mp->epm_qstat = Q_UNINITIALIZED;
1411 mp->epm_timer_id = fmd_timer_install(hdl, mp, NULL,
1412 Reconn_interval);
1413 mp->epm_timer_in_use = 1;
1414 }
1415 } else {
1416 etm_reconnect(hdl, mp);
1417 }
1418
1419 (void) pthread_mutex_unlock(&mp->epm_lock);
1420 }
1421
1422 /*
1423 * FMD Module declarations
1424 */
1425 static const fmd_hdl_ops_t etm_ops = {
1426 NULL, /* fmdo_recv */
1427 etm_timeout, /* fmdo_timeout */
1428 NULL, /* fmdo_close */
1429 NULL, /* fmdo_stats */
1430 NULL, /* fmdo_gc */
1431 etm_send, /* fmdo_send */
1432 };
1433
1434 static const fmd_prop_t etm_props[] = {
1435 { "client_list", FMD_TYPE_STRING, NULL },
1436 { "server_list", FMD_TYPE_STRING, NULL },
1437 { "reconnect_interval", FMD_TYPE_UINT64, "10000000000" },
1438 { "reconnect_timeout", FMD_TYPE_UINT64, "300000000000" },
1439 { "rw_timeout", FMD_TYPE_UINT64, "2000000000" },
1440 { "filter_path", FMD_TYPE_STRING, NULL },
1441 { NULL, 0, NULL }
1442 };
1443
1444 static const fmd_hdl_info_t etm_info = {
1445 "Event Transport Module", "2.0", &etm_ops, etm_props
1446 };
1447
1448 /*
1449 * Initialize the transport for use by ETM.
1450 */
1451 void
_fmd_init(fmd_hdl_t * hdl)1452 _fmd_init(fmd_hdl_t *hdl)
1453 {
1454 char *propstr;
1455
1456 if (fmd_hdl_register(hdl, FMD_API_VERSION, &etm_info) != 0) {
1457 return; /* invalid data in configuration file */
1458 }
1459
1460 /* Create global stats */
1461 (void) fmd_stat_create(hdl, FMD_STAT_NOALLOC,
1462 sizeof (Etm_stats) / sizeof (fmd_stat_t), (fmd_stat_t *)&Etm_stats);
1463
1464 /* Get module properties */
1465 Reconn_timeout = fmd_prop_get_int64(hdl, "reconnect_timeout");
1466 Reconn_interval = fmd_prop_get_int64(hdl, "reconnect_interval");
1467 Rw_timeout = fmd_prop_get_int64(hdl, "rw_timeout");
1468
1469 propstr = fmd_prop_get_string(hdl, "client_list");
1470 etm_create_epmaps(hdl, propstr, ETM_SERVER_XPRT_FLAGS);
1471 fmd_prop_free_string(hdl, propstr);
1472
1473 propstr = fmd_prop_get_string(hdl, "server_list");
1474 etm_create_epmaps(hdl, propstr, ETM_CLIENT_XPRT_FLAGS);
1475 fmd_prop_free_string(hdl, propstr);
1476
1477 if (Etm_stats.peer_count.fmds_value.ui64 == 0) {
1478 fmd_hdl_debug(hdl, "Failed to init any endpoint\n");
1479 fmd_hdl_unregister(hdl);
1480 return;
1481 }
1482 }
1483
1484 /*
1485 * Teardown the transport
1486 */
1487 void
_fmd_fini(fmd_hdl_t * hdl)1488 _fmd_fini(fmd_hdl_t *hdl)
1489 {
1490 etm_epmap_t *mp, *next;
1491
1492 (void) pthread_mutex_lock(&Etm_mod_lock);
1493 Etm_exit = 1;
1494 (void) pthread_mutex_unlock(&Etm_mod_lock);
1495
1496 mp = Epmap_head;
1497
1498 while (mp) {
1499 next = mp->epm_next;
1500 etm_free_epmap(hdl, mp);
1501 mp = next;
1502 }
1503
1504 fmd_hdl_unregister(hdl);
1505 }
1506