xref: /titanic_51/usr/src/uts/sun4v/io/ds_common.c (revision df3f3af853032c14214c08bb7f1b9c376dea8218)
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 2009 Sun Microsystems, Inc.  All rights reserved.
24  * Use is subject to license terms.
25  */
26 
27 /*
28  * Domain Services Module Common Code.
29  *
30  * This module is intended to be used by both Solaris and the VBSC
31  * module.
32  */
33 
34 #include <sys/modctl.h>
35 #include <sys/ksynch.h>
36 #include <sys/taskq.h>
37 #include <sys/disp.h>
38 #include <sys/cmn_err.h>
39 #include <sys/note.h>
40 #include <sys/mach_descrip.h>
41 #include <sys/mdesc.h>
42 #include <sys/ldc.h>
43 #include <sys/ds.h>
44 #include <sys/ds_impl.h>
45 
46 #ifndef MIN
47 #define	MIN(a, b)	((a) < (b) ? (a) : (b))
48 #endif
49 
50 #define	DS_DECODE_BUF_LEN		30
51 
52 /*
53  * All DS ports in the system
54  *
55  * The list of DS ports is read in from the MD when the DS module is
56  * initialized and is never modified. This eliminates the need for
57  * locking to access the port array itself. Access to the individual
58  * ports are synchronized at the port level.
59  */
60 ds_port_t	ds_ports[DS_MAX_PORTS];
61 ds_portset_t	ds_allports;	/* all DS ports in the system */
62 ds_portset_t	ds_nullport;	/* allows test against null portset */
63 
64 /* DS SP port id */
65 uint64_t ds_sp_port_id = DS_PORTID_INVALID;
66 
67 /*
68  * Table of registered services
69  *
70  * Locking: Accesses to the table of services are synchronized using
71  *   a mutex lock. The reader lock must be held when looking up service
72  *   information in the table. The writer lock must be held when any
73  *   service information is being modified.
74  */
75 ds_svcs_t	ds_svcs;
76 
77 /*
78  * Flag to prevent callbacks while in the middle of DS teardown.
79  */
80 boolean_t ds_enabled = B_FALSE;	/* enable/disable taskq processing */
81 
82 /*
83  * Retry count and delay for LDC reads and writes
84  */
85 #ifndef DS_DEFAULT_RETRIES
86 #define	DS_DEFAULT_RETRIES	10000	/* number of times to retry */
87 #endif
88 #ifndef DS_DEFAULT_DELAY
89 #define	DS_DEFAULT_DELAY	1000	/* usecs to wait between retries */
90 #endif
91 
92 static int ds_retries = DS_DEFAULT_RETRIES;
93 static clock_t ds_delay = DS_DEFAULT_DELAY;
94 
95 /*
96  * Supported versions of the DS message protocol
97  *
98  * The version array must be sorted in order from the highest
99  * supported version to the lowest. Support for a particular
100  * <major>.<minor> version implies all lower minor versions of
101  * that same major version are supported as well.
102  */
103 static ds_ver_t ds_vers[] = { { 1, 0 } };
104 
105 #define	DS_NUM_VER	(sizeof (ds_vers) / sizeof (ds_vers[0]))
106 
107 
108 /* incoming message handling functions */
109 typedef void (*ds_msg_handler_t)(ds_port_t *port, caddr_t buf, size_t len);
110 static void ds_handle_init_req(ds_port_t *port, caddr_t buf, size_t len);
111 static void ds_handle_init_ack(ds_port_t *port, caddr_t buf, size_t len);
112 static void ds_handle_init_nack(ds_port_t *port, caddr_t buf, size_t len);
113 static void ds_handle_reg_req(ds_port_t *port, caddr_t buf, size_t len);
114 static void ds_handle_reg_ack(ds_port_t *port, caddr_t buf, size_t len);
115 static void ds_handle_reg_nack(ds_port_t *port, caddr_t buf, size_t len);
116 static void ds_handle_unreg_req(ds_port_t *port, caddr_t buf, size_t len);
117 static void ds_handle_unreg_ack(ds_port_t *port, caddr_t buf, size_t len);
118 static void ds_handle_unreg_nack(ds_port_t *port, caddr_t buf, size_t len);
119 static void ds_handle_data(ds_port_t *port, caddr_t buf, size_t len);
120 static void ds_handle_nack(ds_port_t *port, caddr_t buf, size_t len);
121 
122 /*
123  * DS Message Handler Dispatch Table
124  *
125  * A table used to dispatch all incoming messages. This table
126  * contains handlers for all the fixed message types, as well as
127  * the the messages defined in the 1.0 version of the DS protocol.
128  * The handlers are indexed based on the DS header msg_type values
129  */
130 static const ds_msg_handler_t ds_msg_handlers[] = {
131 	ds_handle_init_req,		/* DS_INIT_REQ */
132 	ds_handle_init_ack,		/* DS_INIT_ACK */
133 	ds_handle_init_nack,		/* DS_INIT_NACK */
134 	ds_handle_reg_req,		/* DS_REG_REQ */
135 	ds_handle_reg_ack,		/* DS_REG_ACK */
136 	ds_handle_reg_nack,		/* DS_REG_NACK */
137 	ds_handle_unreg_req,		/* DS_UNREG */
138 	ds_handle_unreg_ack,		/* DS_UNREG_ACK */
139 	ds_handle_unreg_nack,		/* DS_UNREG_NACK */
140 	ds_handle_data,			/* DS_DATA */
141 	ds_handle_nack			/* DS_NACK */
142 };
143 
144 
145 
146 /* initialization functions */
147 static int ds_ldc_init(ds_port_t *port);
148 
149 /* event processing functions */
150 static uint_t ds_ldc_cb(uint64_t event, caddr_t arg);
151 static int ds_recv_msg(ds_port_t *port, caddr_t msgp, size_t *sizep);
152 static void ds_handle_up_event(ds_port_t *port);
153 static void ds_handle_down_reset_events(ds_port_t *port);
154 static void ds_handle_recv(void *arg);
155 static void ds_dispatch_event(void *arg);
156 
157 /* message sending functions */
158 static int ds_send_msg(ds_port_t *port, caddr_t msg, size_t msglen);
159 static int ds_send_reg_req(ds_svc_t *svc, ds_port_t *port);
160 static void ds_send_unreg_nack(ds_port_t *port, ds_svc_hdl_t bad_hdl);
161 static void ds_send_data_nack(ds_port_t *port, ds_svc_hdl_t bad_hdl);
162 
163 /* walker functions */
164 static int ds_svc_isfree(ds_svc_t *svc, void *arg);
165 static int ds_svc_unregister(ds_svc_t *svc, void *arg);
166 static int ds_svc_port_up(ds_svc_t *svc, void *arg);
167 
168 /* service utilities */
169 static void ds_reset_svc(ds_svc_t *svc, ds_port_t *port);
170 static int ds_svc_register_onport(ds_svc_t *svc, ds_port_t *port);
171 static int ds_svc_register_onport_walker(ds_svc_t *svc, void *arg);
172 static void ds_set_port_ready(ds_port_t *port, uint16_t major, uint16_t minor);
173 
174 /* port utilities */
175 static void ds_port_reset(ds_port_t *port);
176 static ldc_status_t ds_update_ldc_state(ds_port_t *port);
177 
178 /* misc utilities */
179 static void min_max_versions(int num_versions, ds_ver_t *sup_versionsp,
180     uint16_t *min_major, uint16_t *max_major);
181 
182 /* debug */
183 static char *decode_ldc_events(uint64_t event, char *buf);
184 
185 /* loopback */
186 static void ds_loopback_register(ds_svc_hdl_t hdl);
187 static void ds_loopback_unregister(ds_svc_hdl_t hdl);
188 static void ds_loopback_send(ds_svc_hdl_t hdl, void *buf, size_t buflen);
189 static int ds_loopback_set_svc(ds_svc_t *svc, ds_capability_t *cap,
190     ds_svc_hdl_t *lb_hdlp);
191 
192 /* client handling */
193 static int i_ds_hdl_lookup(char *service, uint_t is_client, ds_svc_hdl_t *hdlp,
194     uint_t maxhdls);
195 static ds_svc_t *ds_find_clnt_svc_by_hdl_port(ds_svc_hdl_t hdl,
196     ds_port_t *port);
197 static ds_svc_t *ds_find_svc_by_id_port(char *svc_id, int is_client,
198     ds_port_t *port);
199 static ds_svc_t *ds_svc_clone(ds_svc_t *svc);
200 static void ds_check_for_dup_services(ds_svc_t *svc);
201 static void ds_delete_svc_entry(ds_svc_t *svc);
202 
203 char *
204 ds_strdup(char *str)
205 {
206 	char *newstr;
207 
208 	newstr = DS_MALLOC(strlen(str) + 1);
209 	(void) strcpy(newstr, str);
210 	return (newstr);
211 }
212 
213 void
214 ds_common_init(void)
215 {
216 	/* Validate version table */
217 	ASSERT(ds_vers_isvalid(ds_vers, DS_NUM_VER) == DS_VERS_OK);
218 
219 	/* Initialize services table */
220 	ds_init_svcs_tbl(DS_MAXSVCS_INIT);
221 
222 	/* enable callback processing */
223 	ds_enabled = B_TRUE;
224 }
225 
226 /* BEGIN LDC SUPPORT FUNCTIONS */
227 
228 static char *
229 decode_ldc_events(uint64_t event, char *buf)
230 {
231 	buf[0] = 0;
232 	if (event & LDC_EVT_DOWN)	(void) strcat(buf, " DOWN");
233 	if (event & LDC_EVT_RESET)	(void) strcat(buf, " RESET");
234 	if (event & LDC_EVT_UP)		(void) strcat(buf, " UP");
235 	if (event & LDC_EVT_READ)	(void) strcat(buf, " READ");
236 	if (event & LDC_EVT_WRITE)	(void) strcat(buf, " WRITE");
237 	return (buf);
238 }
239 
240 static ldc_status_t
241 ds_update_ldc_state(ds_port_t *port)
242 {
243 	ldc_status_t	ldc_state;
244 	int		rv;
245 	char		ebuf[DS_EBUFSIZE];
246 
247 	ASSERT(MUTEX_HELD(&port->lock));
248 
249 	/*
250 	 * Read status and update ldc state info in port structure.
251 	 */
252 	if ((rv = ldc_status(port->ldc.hdl, &ldc_state)) != 0) {
253 		cmn_err(CE_WARN, "ds@%lx: %s: ldc_status error: %s" DS_EOL,
254 		    PORTID(port), __func__, ds_errno_to_str(rv, ebuf));
255 		ldc_state = port->ldc.state;
256 	} else {
257 		port->ldc.state = ldc_state;
258 	}
259 
260 	return (ldc_state);
261 }
262 
263 static void
264 ds_handle_down_reset_events(ds_port_t *port)
265 {
266 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s: entered" DS_EOL, PORTID(port),
267 	    __func__);
268 
269 	mutex_enter(&ds_svcs.lock);
270 	mutex_enter(&port->lock);
271 
272 	ds_sys_drain_events(port);
273 
274 	(void) ds_update_ldc_state(port);
275 
276 	/* reset the port state */
277 	ds_port_reset(port);
278 
279 	/* acknowledge the reset */
280 	(void) ldc_up(port->ldc.hdl);
281 
282 	mutex_exit(&port->lock);
283 	mutex_exit(&ds_svcs.lock);
284 
285 	ds_handle_up_event(port);
286 
287 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s: exit" DS_EOL, PORTID(port), __func__);
288 }
289 
290 static void
291 ds_handle_up_event(ds_port_t *port)
292 {
293 	ldc_status_t	ldc_state;
294 
295 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s: entered" DS_EOL, PORTID(port),
296 	    __func__);
297 
298 	mutex_enter(&port->lock);
299 
300 	ldc_state = ds_update_ldc_state(port);
301 
302 	mutex_exit(&port->lock);
303 
304 	if ((ldc_state == LDC_UP) && IS_DS_PORT(port)) {
305 		/*
306 		 * Initiate the handshake.
307 		 */
308 		ds_send_init_req(port);
309 	}
310 
311 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s: exit" DS_EOL, PORTID(port), __func__);
312 }
313 
314 static uint_t
315 ds_ldc_cb(uint64_t event, caddr_t arg)
316 {
317 	ds_port_t	*port = (ds_port_t *)arg;
318 	char		evstring[DS_DECODE_BUF_LEN];
319 
320 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s: %s event (%llx) received" DS_EOL,
321 	    PORTID(port), __func__, decode_ldc_events(event, evstring),
322 	    (u_longlong_t)event);
323 
324 	if (!ds_enabled) {
325 		DS_DBG_LDC(CE_NOTE, "ds@%lx: %s: callback handling is disabled"
326 		    DS_EOL, PORTID(port), __func__);
327 		return (LDC_SUCCESS);
328 	}
329 
330 	if (event & (LDC_EVT_DOWN | LDC_EVT_RESET)) {
331 		ds_handle_down_reset_events(port);
332 		goto done;
333 	}
334 
335 	if (event & LDC_EVT_UP) {
336 		ds_handle_up_event(port);
337 	}
338 
339 	if (event & LDC_EVT_READ) {
340 		if (port->ldc.state != LDC_UP) {
341 			cmn_err(CE_WARN, "ds@%lx: %s: LDC READ event while "
342 			    "port not up" DS_EOL, PORTID(port), __func__);
343 			goto done;
344 		}
345 
346 		if (ds_sys_dispatch_func(ds_handle_recv, port)) {
347 			cmn_err(CE_WARN, "ds@%lx: error initiating LDC READ "
348 			    " event", PORTID(port));
349 		}
350 	}
351 
352 	if (event & LDC_EVT_WRITE) {
353 		DS_DBG_LDC(CE_NOTE, "ds@%lx: %s: LDC WRITE event received, "
354 		    "not supported" DS_EOL, PORTID(port), __func__);
355 	}
356 
357 	if (event & ~(LDC_EVT_UP | LDC_EVT_READ)) {
358 		cmn_err(CE_WARN, "ds@%lx: %s: Unexpected LDC event received: "
359 		    "0x%llx" DS_EOL, PORTID(port), __func__,
360 		    (u_longlong_t)event);
361 	}
362 done:
363 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s: exit" DS_EOL, PORTID(port), __func__);
364 
365 	return (LDC_SUCCESS);
366 }
367 
368 static int
369 ds_ldc_init(ds_port_t *port)
370 {
371 	int		rv;
372 	ldc_attr_t	ldc_attr;
373 	caddr_t		ldc_cb_arg = (caddr_t)port;
374 	char		ebuf[DS_EBUFSIZE];
375 
376 	ASSERT(MUTEX_HELD(&port->lock));
377 
378 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s: ldc_id=%lld" DS_EOL,
379 	    PORTID(port), __func__, (u_longlong_t)port->ldc.id);
380 
381 	ldc_attr.devclass = LDC_DEV_GENERIC;
382 	ldc_attr.instance = 0;
383 	ldc_attr.mode = LDC_MODE_RELIABLE;
384 	ldc_attr.mtu = DS_STREAM_MTU;
385 
386 	if ((rv = ldc_init(port->ldc.id, &ldc_attr, &port->ldc.hdl)) != 0) {
387 		cmn_err(CE_WARN, "ds@%lx: %s: ldc_id: %lx, ldc_init error: %s"
388 		    DS_EOL, PORTID(port), __func__, port->ldc.id,
389 		    ds_errno_to_str(rv, ebuf));
390 		return (rv);
391 	}
392 
393 	rv = ldc_reg_callback(port->ldc.hdl, ds_ldc_cb, ldc_cb_arg);
394 	if (rv != 0) {
395 		cmn_err(CE_WARN, "ds@%lx: %s: ldc_reg_callback error: %s"
396 		    DS_EOL, PORTID(port), __func__, ds_errno_to_str(rv, ebuf));
397 		return (rv);
398 	}
399 
400 	ds_sys_ldc_init(port);
401 	return (0);
402 }
403 
404 int
405 ds_ldc_fini(ds_port_t *port)
406 {
407 	int	rv;
408 	char	ebuf[DS_EBUFSIZE];
409 
410 	ASSERT(port->state >= DS_PORT_LDC_INIT);
411 	ASSERT(MUTEX_HELD(&port->lock));
412 
413 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s: ldc_id=%ld" DS_EOL, PORTID(port),
414 	    __func__, port->ldc.id);
415 
416 	if ((rv = ldc_close(port->ldc.hdl)) != 0) {
417 		cmn_err(CE_WARN, "ds@%lx: %s: ldc_close error: %s" DS_EOL,
418 		    PORTID(port), __func__, ds_errno_to_str(rv, ebuf));
419 		return (rv);
420 	}
421 
422 	if ((rv = ldc_unreg_callback(port->ldc.hdl)) != 0) {
423 		cmn_err(CE_WARN, "ds@%lx: %s: ldc_unreg_callback error: %s"
424 		    DS_EOL, PORTID(port), __func__, ds_errno_to_str(rv, ebuf));
425 		return (rv);
426 	}
427 
428 	if ((rv = ldc_fini(port->ldc.hdl)) != 0) {
429 		cmn_err(CE_WARN, "ds@%lx: %s: ldc_fini error: %s" DS_EOL,
430 		    PORTID(port), __func__, ds_errno_to_str(rv, ebuf));
431 		return (rv);
432 	}
433 
434 	port->ldc.id = (uint64_t)-1;
435 	port->ldc.hdl = NULL;
436 	port->ldc.state = 0;
437 
438 	return (rv);
439 }
440 
441 /*
442  * Attempt to read a specified number of bytes from a particular LDC.
443  * Returns zero for success or the return code from the LDC read on
444  * failure. The actual number of bytes read from the LDC is returned
445  * in the size parameter.
446  */
447 static int
448 ds_recv_msg(ds_port_t *port, caddr_t msgp, size_t *sizep)
449 {
450 	int	rv = 0;
451 	size_t	bytes_req = *sizep;
452 	size_t	bytes_left = bytes_req;
453 	size_t	nbytes;
454 	int	retry_count = 0;
455 	char	ebuf[DS_EBUFSIZE];
456 
457 	ASSERT(MUTEX_HELD(&port->rcv_lock));
458 
459 	*sizep = 0;
460 
461 	DS_DBG_LDC(CE_NOTE, "ds@%lx: attempting to read %ld bytes" DS_EOL,
462 	    PORTID(port), bytes_req);
463 
464 	while (bytes_left > 0) {
465 
466 		nbytes = bytes_left;
467 
468 		mutex_enter(&port->lock);
469 		if (port->ldc.state == LDC_UP) {
470 			rv = ldc_read(port->ldc.hdl, msgp, &nbytes);
471 		} else
472 			rv = ENXIO;
473 		mutex_exit(&port->lock);
474 		if (rv != 0) {
475 			if (rv == ECONNRESET) {
476 				break;
477 			} else if (rv != EAGAIN) {
478 				cmn_err(CE_NOTE, "ds@%lx: %s: %s" DS_EOL,
479 				    PORTID(port), __func__,
480 				    ds_errno_to_str(rv, ebuf));
481 				break;
482 			}
483 		} else {
484 			if (nbytes != 0) {
485 				DS_DBG_LDC(CE_NOTE, "ds@%lx: "
486 				    "read %ld bytes, %d retries" DS_EOL,
487 				    PORTID(port), nbytes, retry_count);
488 
489 				*sizep += nbytes;
490 				msgp += nbytes;
491 				bytes_left -= nbytes;
492 
493 				/* reset counter on a successful read */
494 				retry_count = 0;
495 				continue;
496 			}
497 
498 			/*
499 			 * No data was read. Check if this is the
500 			 * first attempt. If so, just return since
501 			 * nothing has been read yet.
502 			 */
503 			if (bytes_left == bytes_req) {
504 				DS_DBG_LDC(CE_NOTE, "ds@%lx: read zero bytes, "
505 				    " no data available" DS_EOL, PORTID(port));
506 				break;
507 			}
508 		}
509 
510 		/*
511 		 * A retry is necessary because the read returned
512 		 * EAGAIN, or a zero length read occurred after
513 		 * reading a partial message.
514 		 */
515 		if (retry_count++ >= ds_retries) {
516 			DS_DBG_LDC(CE_NOTE, "ds@%lx: timed out waiting for "
517 			    "message" DS_EOL, PORTID(port));
518 			break;
519 		}
520 
521 		drv_usecwait(ds_delay);
522 	}
523 
524 	return (rv);
525 }
526 
527 static void
528 ds_handle_recv(void *arg)
529 {
530 	ds_port_t	*port = (ds_port_t *)arg;
531 	char		*hbuf;
532 	size_t		msglen;
533 	size_t		read_size;
534 	boolean_t	hasdata;
535 	ds_hdr_t	hdr;
536 	uint8_t		*msg;
537 	char		*currp;
538 	int		rv;
539 	ds_event_t	*devent;
540 
541 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s..." DS_EOL, PORTID(port), __func__);
542 
543 	/*
544 	 * Read messages from the channel until there are none
545 	 * pending. Valid messages are dispatched to be handled
546 	 * by a separate thread while any malformed messages are
547 	 * dropped.
548 	 */
549 
550 	mutex_enter(&port->rcv_lock);
551 
552 	for (;;) {
553 		mutex_enter(&port->lock);
554 		if (port->ldc.state == LDC_UP) {
555 			rv = ldc_chkq(port->ldc.hdl, &hasdata);
556 		} else
557 			rv = ENXIO;
558 		mutex_exit(&port->lock);
559 		if (rv != 0 || !hasdata)
560 			break;
561 
562 		DS_DBG(CE_NOTE, "ds@%lx: %s: reading next message" DS_EOL,
563 		    PORTID(port), __func__);
564 
565 		/*
566 		 * Read in the next message.
567 		 */
568 		hbuf = (char *)&hdr;
569 		bzero(hbuf, DS_HDR_SZ);
570 		read_size = DS_HDR_SZ;
571 		currp = hbuf;
572 
573 		/* read in the message header */
574 		if ((rv = ds_recv_msg(port, currp, &read_size)) != 0) {
575 			break;
576 		}
577 
578 		if (read_size < DS_HDR_SZ) {
579 			/*
580 			 * A zero length read is a valid signal that
581 			 * there is no data left on the channel.
582 			 */
583 			if (read_size != 0) {
584 				cmn_err(CE_WARN, "ds@%lx: invalid message "
585 				    "length, received %ld bytes, expected %ld"
586 				    DS_EOL, PORTID(port), read_size, DS_HDR_SZ);
587 			}
588 			continue;
589 		}
590 
591 		/* get payload size and allocate a buffer */
592 		read_size = ((ds_hdr_t *)hbuf)->payload_len;
593 		msglen = DS_HDR_SZ + read_size;
594 		msg = DS_MALLOC(msglen);
595 		if (!msg) {
596 			cmn_err(CE_WARN, "Memory allocation failed attempting "
597 			    " to allocate %d bytes." DS_EOL, (int)msglen);
598 			continue;
599 		}
600 
601 		DS_DBG(CE_NOTE, "ds@%lx: %s: message payload len %d" DS_EOL,
602 		    PORTID(port), __func__, (int)read_size);
603 
604 		/* move message header into buffer */
605 		(void) memcpy(msg, hbuf, DS_HDR_SZ);
606 		currp = (char *)(msg) + DS_HDR_SZ;
607 
608 		/* read in the message body */
609 		if ((rv = ds_recv_msg(port, currp, &read_size)) != 0) {
610 			DS_FREE(msg, msglen);
611 			break;
612 		}
613 
614 		/* validate the size of the message */
615 		if ((DS_HDR_SZ + read_size) != msglen) {
616 			cmn_err(CE_WARN, "ds@%lx: %s: invalid message length, "
617 			    "received %ld bytes, expected %ld" DS_EOL,
618 			    PORTID(port), __func__, (DS_HDR_SZ + read_size),
619 			    msglen);
620 			DS_FREE(msg, msglen);
621 			continue;
622 		}
623 
624 		DS_DUMP_MSG(DS_DBG_FLAG_LDC, msg, msglen);
625 
626 		/*
627 		 * Send the message for processing, and store it
628 		 * in the log. The memory is deallocated only when
629 		 * the message is removed from the log.
630 		 */
631 
632 		devent = DS_MALLOC(sizeof (ds_event_t));
633 		devent->port = port;
634 		devent->buf = (char *)msg;
635 		devent->buflen = msglen;
636 
637 		/* log the message */
638 		(void) ds_log_add_msg(DS_LOG_IN(port->id), msg, msglen);
639 
640 		if (ds_sys_dispatch_func(ds_dispatch_event, devent)) {
641 			cmn_err(CE_WARN, "ds@%lx: error initiating "
642 			    "event handler", PORTID(port));
643 			DS_FREE(devent, sizeof (ds_event_t));
644 		}
645 	}
646 
647 	mutex_exit(&port->rcv_lock);
648 
649 	/* handle connection reset errors returned from ds_recv_msg */
650 	if (rv == ECONNRESET) {
651 		ds_handle_down_reset_events(port);
652 	}
653 
654 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s done" DS_EOL, PORTID(port), __func__);
655 }
656 
657 static void
658 ds_dispatch_event(void *arg)
659 {
660 	ds_event_t	*event = (ds_event_t *)arg;
661 	ds_hdr_t	*hdr;
662 	ds_port_t	*port;
663 
664 	port = event->port;
665 
666 	hdr = (ds_hdr_t *)event->buf;
667 
668 	if (DS_MSG_TYPE_VALID(hdr->msg_type)) {
669 		DS_DBG(CE_NOTE, "ds@%lx: dispatch_event: msg_type=%d" DS_EOL,
670 		    PORTID(port), hdr->msg_type);
671 
672 		(*ds_msg_handlers[hdr->msg_type])(port, event->buf,
673 		    event->buflen);
674 	} else {
675 		cmn_err(CE_WARN, "ds@%lx: dispatch_event: invalid msg "
676 		    "type (%d)" DS_EOL, PORTID(port), hdr->msg_type);
677 	}
678 
679 	DS_FREE(event->buf, event->buflen);
680 	DS_FREE(event, sizeof (ds_event_t));
681 }
682 
683 int
684 ds_send_msg(ds_port_t *port, caddr_t msg, size_t msglen)
685 {
686 	int	rv;
687 	caddr_t	currp = msg;
688 	size_t	amt_left = msglen;
689 	int	loopcnt = 0;
690 
691 	DS_DBG_LDC(CE_NOTE, "ds@%lx: %s msglen: %ld" DS_EOL, PORTID(port),
692 	    __func__, msglen);
693 	DS_DUMP_MSG(DS_DBG_FLAG_LDC, msg, msglen);
694 
695 	(void) ds_log_add_msg(DS_LOG_OUT(port->id), (uint8_t *)msg, msglen);
696 
697 	/*
698 	 * Ensure that no other messages can be sent on this port by holding
699 	 * the tx_lock mutex in case the write doesn't get sent with one write.
700 	 * This guarantees that the message doesn't become fragmented.
701 	 */
702 	mutex_enter(&port->tx_lock);
703 
704 	do {
705 		mutex_enter(&port->lock);
706 		if (port->ldc.state == LDC_UP) {
707 			rv = ldc_write(port->ldc.hdl, currp, &msglen);
708 		} else
709 			rv = ENXIO;
710 		mutex_exit(&port->lock);
711 		if (rv != 0) {
712 			if (rv == ECONNRESET) {
713 				mutex_exit(&port->tx_lock);
714 				ds_handle_down_reset_events(port);
715 				return (rv);
716 			} else if ((rv == EWOULDBLOCK) &&
717 			    (loopcnt++ < ds_retries)) {
718 				drv_usecwait(ds_delay);
719 			} else {
720 				DS_DBG_PRCL(CE_NOTE, "ds@%lx: send_msg: "
721 				    "ldc_write failed (%d), %d bytes "
722 				    "remaining" DS_EOL, PORTID(port), rv,
723 				    (int)amt_left);
724 				goto error;
725 			}
726 		} else {
727 			amt_left -= msglen;
728 			currp += msglen;
729 			msglen = amt_left;
730 			loopcnt = 0;
731 		}
732 	} while (amt_left > 0);
733 error:
734 	mutex_exit(&port->tx_lock);
735 
736 	return (rv);
737 }
738 
739 /* END LDC SUPPORT FUNCTIONS */
740 
741 
742 /* BEGIN DS PROTOCOL SUPPORT FUNCTIONS */
743 
744 static void
745 ds_handle_init_req(ds_port_t *port, caddr_t buf, size_t len)
746 {
747 	ds_hdr_t	*hdr;
748 	ds_init_ack_t	*ack;
749 	ds_init_nack_t	*nack;
750 	char		*msg;
751 	size_t		msglen;
752 	ds_init_req_t	*req;
753 	size_t		explen = DS_MSG_LEN(ds_init_req_t);
754 	uint16_t	new_major;
755 	uint16_t	new_minor;
756 	boolean_t	match;
757 
758 	/* sanity check the incoming message */
759 	if (len != explen) {
760 		cmn_err(CE_WARN, "ds@%lx: <init_req: invalid message "
761 		    "length (%ld), expected %ld" DS_EOL, PORTID(port), len,
762 		    explen);
763 		return;
764 	}
765 
766 	req = (ds_init_req_t *)(buf + DS_HDR_SZ);
767 
768 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <init_req: ver=%d.%d" DS_EOL,
769 	    PORTID(port), req->major_vers, req->minor_vers);
770 
771 	match = negotiate_version(DS_NUM_VER, &ds_vers[0],
772 	    req->major_vers, &new_major, &new_minor);
773 
774 	/*
775 	 * Check version info. ACK only if the major numbers exactly
776 	 * match. The service entity can retry with a new minor
777 	 * based on the response sent as part of the NACK.
778 	 */
779 	if (match) {
780 		msglen = DS_MSG_LEN(ds_init_ack_t);
781 		msg = DS_MALLOC(msglen);
782 
783 		hdr = (ds_hdr_t *)msg;
784 		hdr->msg_type = DS_INIT_ACK;
785 		hdr->payload_len = sizeof (ds_init_ack_t);
786 
787 		ack = (ds_init_ack_t *)(msg + DS_HDR_SZ);
788 		ack->minor_vers = MIN(new_minor, req->minor_vers);
789 
790 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: init_ack>: minor=0x%04X" DS_EOL,
791 		    PORTID(port), MIN(new_minor, req->minor_vers));
792 	} else {
793 		msglen = DS_MSG_LEN(ds_init_nack_t);
794 		msg = DS_MALLOC(msglen);
795 
796 		hdr = (ds_hdr_t *)msg;
797 		hdr->msg_type = DS_INIT_NACK;
798 		hdr->payload_len = sizeof (ds_init_nack_t);
799 
800 		nack = (ds_init_nack_t *)(msg + DS_HDR_SZ);
801 		nack->major_vers = new_major;
802 
803 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: init_nack>: major=0x%04X" DS_EOL,
804 		    PORTID(port), new_major);
805 	}
806 
807 	/*
808 	 * Send the response
809 	 */
810 	(void) ds_send_msg(port, msg, msglen);
811 	DS_FREE(msg, msglen);
812 
813 	if (match) {
814 		ds_set_port_ready(port, req->major_vers, ack->minor_vers);
815 	}
816 }
817 
818 static void
819 ds_handle_init_ack(ds_port_t *port, caddr_t buf, size_t len)
820 {
821 	ds_init_ack_t	*ack;
822 	ds_ver_t	*ver;
823 	uint16_t	major;
824 	uint16_t	minor;
825 	size_t		explen = DS_MSG_LEN(ds_init_ack_t);
826 
827 	/* sanity check the incoming message */
828 	if (len != explen) {
829 		cmn_err(CE_WARN, "ds@%lx: <init_ack: invalid message "
830 		    "length (%ld), expected %ld" DS_EOL, PORTID(port), len,
831 		    explen);
832 		return;
833 	}
834 
835 	ack = (ds_init_ack_t *)(buf + DS_HDR_SZ);
836 
837 	mutex_enter(&port->lock);
838 
839 	if (port->state == DS_PORT_READY) {
840 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <init_ack: port ready" DS_EOL,
841 		    PORTID(port));
842 		mutex_exit(&port->lock);
843 		return;
844 	}
845 
846 	if (port->state != DS_PORT_INIT_REQ) {
847 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <init_ack: invalid state: %d"
848 		    DS_EOL, PORTID(port), port->state);
849 		mutex_exit(&port->lock);
850 		return;
851 	}
852 
853 	ver = &(ds_vers[port->ver_idx]);
854 	major = ver->major;
855 	minor = MIN(ver->minor, ack->minor_vers);
856 	mutex_exit(&port->lock);
857 
858 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <init_ack: port ready v%d.%d" DS_EOL,
859 	    PORTID(port), major, minor);
860 
861 	ds_set_port_ready(port, major, minor);
862 }
863 
864 static void
865 ds_handle_init_nack(ds_port_t *port, caddr_t buf, size_t len)
866 {
867 	int		idx;
868 	ds_init_nack_t	*nack;
869 	ds_ver_t	*ver;
870 	size_t		explen = DS_MSG_LEN(ds_init_nack_t);
871 
872 	/* sanity check the incoming message */
873 	if (len != explen) {
874 		DS_DBG_PRCL(CE_WARN, "ds@%lx: <init_nack: invalid message "
875 		    "length (%ld), expected %ld" DS_EOL, PORTID(port), len,
876 		    explen);
877 		return;
878 	}
879 
880 	nack = (ds_init_nack_t *)(buf + DS_HDR_SZ);
881 
882 	mutex_enter(&port->lock);
883 
884 	if (port->state != DS_PORT_INIT_REQ) {
885 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <init_nack: invalid state: %d"
886 		    DS_EOL, PORTID(port), port->state);
887 		mutex_exit(&port->lock);
888 		return;
889 	}
890 
891 	ver = &(ds_vers[port->ver_idx]);
892 
893 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <init_nack: req=v%d.%d, nack=v%d.x"
894 	    DS_EOL, PORTID(port), ver->major, ver->minor, nack->major_vers);
895 
896 	if (nack->major_vers == 0) {
897 		/* no supported protocol version */
898 		DS_DBG_PRCL(CE_WARN, "ds@%lx: <init_nack: DS not supported"
899 		    DS_EOL, PORTID(port));
900 		mutex_exit(&port->lock);
901 		return;
902 	}
903 
904 	/*
905 	 * Walk the version list, looking for a major version
906 	 * that is as close to the requested major version as
907 	 * possible.
908 	 */
909 	for (idx = port->ver_idx; idx < DS_NUM_VER; idx++) {
910 		if (ds_vers[idx].major <= nack->major_vers) {
911 			/* found a version to try */
912 			goto done;
913 		}
914 	}
915 
916 	if (idx == DS_NUM_VER) {
917 		/* no supported version */
918 		DS_DBG_PRCL(CE_WARN, "ds@%lx: <init_nack: DS v%d.x not "
919 		    "supported" DS_EOL, PORTID(port), nack->major_vers);
920 
921 		mutex_exit(&port->lock);
922 		return;
923 	}
924 
925 done:
926 	/* start the handshake again */
927 	port->ver_idx = idx;
928 	port->state = DS_PORT_LDC_INIT;
929 	mutex_exit(&port->lock);
930 
931 	ds_send_init_req(port);
932 
933 }
934 
935 static ds_svc_t *
936 ds_find_svc_by_id_port(char *svc_id, int is_client, ds_port_t *port)
937 {
938 	int		idx;
939 	ds_svc_t	*svc, *found_svc = 0;
940 	uint32_t	flag_match = is_client ? DSSF_ISCLIENT : 0;
941 
942 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
943 
944 	/* walk every table entry */
945 	for (idx = 0; idx < ds_svcs.maxsvcs; idx++) {
946 		svc = ds_svcs.tbl[idx];
947 		if (DS_SVC_ISFREE(svc))
948 			continue;
949 		if (strcmp(svc->cap.svc_id, svc_id) != 0)
950 			continue;
951 		if ((svc->flags & DSSF_ISCLIENT) != flag_match)
952 			continue;
953 		if (port != NULL && svc->port == port) {
954 			return (svc);
955 		} else if (svc->state == DS_SVC_INACTIVE) {
956 			found_svc = svc;
957 		} else if (!found_svc) {
958 			found_svc = svc;
959 		}
960 	}
961 
962 	return (found_svc);
963 }
964 
965 static void
966 ds_handle_reg_req(ds_port_t *port, caddr_t buf, size_t len)
967 {
968 	ds_reg_req_t	*req;
969 	ds_hdr_t	*hdr;
970 	ds_reg_ack_t	*ack;
971 	ds_reg_nack_t	*nack;
972 	char		*msg;
973 	size_t		msglen;
974 	size_t		explen = DS_MSG_LEN(ds_reg_req_t);
975 	ds_svc_t	*svc = NULL;
976 	ds_ver_t	version;
977 	uint16_t	new_major;
978 	uint16_t	new_minor;
979 	boolean_t	match;
980 
981 	/* sanity check the incoming message */
982 	if (len < explen) {
983 		cmn_err(CE_WARN, "ds@%lx: <reg_req: invalid message "
984 		    "length (%ld), expected at least %ld" DS_EOL,
985 		    PORTID(port), len, explen);
986 		return;
987 	}
988 
989 	req = (ds_reg_req_t *)(buf + DS_HDR_SZ);
990 
991 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_req: '%s' ver=%d.%d, hdl=0x%llx"
992 	    DS_EOL, PORTID(port), req->svc_id, req->major_vers, req->minor_vers,
993 	    (u_longlong_t)req->svc_handle);
994 
995 	mutex_enter(&ds_svcs.lock);
996 	svc = ds_find_svc_by_id_port(req->svc_id,
997 	    DS_HDL_ISCLIENT(req->svc_handle) == 0, port);
998 	if (svc == NULL) {
999 
1000 do_reg_nack:
1001 		mutex_exit(&ds_svcs.lock);
1002 
1003 		msglen = DS_MSG_LEN(ds_reg_nack_t);
1004 		msg = DS_MALLOC(msglen);
1005 
1006 		hdr = (ds_hdr_t *)msg;
1007 		hdr->msg_type = DS_REG_NACK;
1008 		hdr->payload_len = sizeof (ds_reg_nack_t);
1009 
1010 		nack = (ds_reg_nack_t *)(msg + DS_HDR_SZ);
1011 		nack->svc_handle = req->svc_handle;
1012 		nack->result = DS_REG_VER_NACK;
1013 		nack->major_vers = 0;
1014 
1015 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: reg_nack>: '%s'" DS_EOL,
1016 		    PORTID(port), req->svc_id);
1017 		/*
1018 		 * Send the response
1019 		 */
1020 		(void) ds_send_msg(port, msg, msglen);
1021 		DS_FREE(msg, msglen);
1022 		return;
1023 	}
1024 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_req: '%s' found, hdl: 0x%llx" DS_EOL,
1025 	    PORTID(port), req->svc_id, (u_longlong_t)svc->hdl);
1026 
1027 	/*
1028 	 * A client sends out a reg req in order to force service providers to
1029 	 * initiate a reg req from their end (limitation in the protocol).  We
1030 	 * expect the service provider to be in the inactive (DS_SVC_INACTIVE)
1031 	 * state.  If the service provider has already sent out a reg req (the
1032 	 * state is DS_SVC_REG_PENDING) or has already handshaken (the
1033 	 * state is DS_SVC_ACTIVE), then we can simply ignore this reg
1034 	 * req.  For any other state, we force an unregister before initiating
1035 	 * a reg req.
1036 	 */
1037 
1038 	if (DS_HDL_ISCLIENT(req->svc_handle)) {
1039 		switch (svc->state) {
1040 
1041 		case DS_SVC_REG_PENDING:
1042 		case DS_SVC_ACTIVE:
1043 			DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_req: '%s' pinging "
1044 			    "client, state (%x)" DS_EOL, PORTID(port),
1045 			    req->svc_id, svc->state);
1046 			mutex_exit(&ds_svcs.lock);
1047 			return;
1048 
1049 		case DS_SVC_INACTIVE:
1050 			DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_req: '%s' pinging "
1051 			    "client" DS_EOL, PORTID(port), req->svc_id);
1052 			break;
1053 
1054 		default:
1055 			DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_req: '%s' pinging "
1056 			    "client forced unreg, state (%x)" DS_EOL,
1057 			    PORTID(port), req->svc_id, svc->state);
1058 			(void) ds_svc_unregister(svc, port);
1059 			break;
1060 		}
1061 		(void) ds_svc_port_up(svc, port);
1062 		(void) ds_svc_register_onport(svc, port);
1063 		mutex_exit(&ds_svcs.lock);
1064 		return;
1065 	}
1066 
1067 	/*
1068 	 * Only remote service providers can initiate a registration.  The
1069 	 * local sevice from here must be a client service.
1070 	 */
1071 
1072 	match = negotiate_version(svc->cap.nvers, svc->cap.vers,
1073 	    req->major_vers, &new_major, &new_minor);
1074 
1075 	/*
1076 	 * Check version info. ACK only if the major numbers exactly
1077 	 * match. The service entity can retry with a new minor
1078 	 * based on the response sent as part of the NACK.
1079 	 */
1080 	if (match) {
1081 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_req: '%s' svc%d: state: %x "
1082 		    "svc_portid: %d" DS_EOL, PORTID(port), req->svc_id,
1083 		    (int)DS_HDL2IDX(svc->hdl), svc->state,
1084 		    (int)(svc->port == NULL ? -1 : PORTID(svc->port)));
1085 		/*
1086 		 * If the current local service is already in use and
1087 		 * it's not on this port, clone it.
1088 		 */
1089 		if (svc->state != DS_SVC_INACTIVE) {
1090 			if (svc->port != NULL && port == svc->port) {
1091 				/*
1092 				 * Someone probably dropped an unreg req
1093 				 * somewhere.  Force a local unreg.
1094 				 */
1095 				(void) ds_svc_unregister(svc, port);
1096 			} else if (!DS_HDL_ISCLIENT(svc->hdl)) {
1097 				/*
1098 				 * Can't clone a non-client (service provider)
1099 				 * handle.  This is because old in-kernel
1100 				 * service providers can't deal with multiple
1101 				 * handles.
1102 				 */
1103 				goto do_reg_nack;
1104 			} else {
1105 				svc = ds_svc_clone(svc);
1106 			}
1107 		}
1108 		svc->port = port;
1109 		svc->svc_hdl = req->svc_handle;
1110 		svc->state = DS_SVC_ACTIVE;
1111 
1112 		msglen = DS_MSG_LEN(ds_reg_ack_t);
1113 		msg = DS_MALLOC(msglen);
1114 
1115 		hdr = (ds_hdr_t *)msg;
1116 		hdr->msg_type = DS_REG_ACK;
1117 		hdr->payload_len = sizeof (ds_reg_ack_t);
1118 
1119 		ack = (ds_reg_ack_t *)(msg + DS_HDR_SZ);
1120 		ack->svc_handle = req->svc_handle;
1121 		ack->minor_vers = MIN(new_minor, req->minor_vers);
1122 
1123 
1124 		if (svc->ops.ds_reg_cb) {
1125 			/* Call the registration callback */
1126 			version.major = req->major_vers;
1127 			version.minor = ack->minor_vers;
1128 			(*svc->ops.ds_reg_cb)(svc->ops.cb_arg, &version,
1129 			    svc->hdl);
1130 		}
1131 		mutex_exit(&ds_svcs.lock);
1132 
1133 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: reg_ack>: '%s' minor=0x%04X"
1134 		    DS_EOL, PORTID(port), svc->cap.svc_id,
1135 		    MIN(new_minor, req->minor_vers));
1136 	} else {
1137 		mutex_exit(&ds_svcs.lock);
1138 
1139 		msglen = DS_MSG_LEN(ds_reg_nack_t);
1140 		msg = DS_MALLOC(msglen);
1141 
1142 		hdr = (ds_hdr_t *)msg;
1143 		hdr->msg_type = DS_REG_NACK;
1144 		hdr->payload_len = sizeof (ds_reg_nack_t);
1145 
1146 		nack = (ds_reg_nack_t *)(msg + DS_HDR_SZ);
1147 		nack->svc_handle = req->svc_handle;
1148 		nack->result = DS_REG_VER_NACK;
1149 		nack->major_vers = new_major;
1150 
1151 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: reg_nack>: '%s' major=0x%04X"
1152 		    DS_EOL, PORTID(port), svc->cap.svc_id, new_major);
1153 	}
1154 
1155 	/* send message */
1156 	(void) ds_send_msg(port, msg, msglen);
1157 	DS_FREE(msg, msglen);
1158 }
1159 
1160 static void
1161 ds_handle_reg_ack(ds_port_t *port, caddr_t buf, size_t len)
1162 {
1163 	ds_reg_ack_t	*ack;
1164 	ds_ver_t	*ver;
1165 	ds_ver_t	tmpver;
1166 	ds_svc_t	*svc;
1167 	size_t		explen = DS_MSG_LEN(ds_reg_ack_t);
1168 
1169 	/* sanity check the incoming message */
1170 	if (len != explen) {
1171 		cmn_err(CE_WARN, "ds@%lx: <reg_ack: invalid message "
1172 		    "length (%ld), expected %ld" DS_EOL, PORTID(port), len,
1173 		    explen);
1174 		return;
1175 	}
1176 
1177 	ack = (ds_reg_ack_t *)(buf + DS_HDR_SZ);
1178 
1179 	mutex_enter(&ds_svcs.lock);
1180 
1181 	/*
1182 	 * This searches for service based on how we generate handles
1183 	 * and so only works because this is a reg ack.
1184 	 */
1185 	if (DS_HDL_ISCLIENT(ack->svc_handle) ||
1186 	    (svc = ds_get_svc(ack->svc_handle)) == NULL) {
1187 		cmn_err(CE_WARN, "ds@%lx: <reg_ack: invalid handle 0x%llx"
1188 		    DS_EOL, PORTID(port), (u_longlong_t)ack->svc_handle);
1189 		goto done;
1190 	}
1191 
1192 	/* make sure the message makes sense */
1193 	if (svc->state != DS_SVC_REG_PENDING) {
1194 		cmn_err(CE_WARN, "ds@%lx: <reg_ack: invalid state (%d)" DS_EOL,
1195 		    PORTID(port), svc->state);
1196 		goto done;
1197 	}
1198 
1199 	ver = &(svc->cap.vers[svc->ver_idx]);
1200 
1201 	/* major version has been agreed upon */
1202 	svc->ver.major = ver->major;
1203 
1204 	if (ack->minor_vers >= ver->minor) {
1205 		/*
1206 		 * Use the minor version specified in the
1207 		 * original request.
1208 		 */
1209 		svc->ver.minor = ver->minor;
1210 	} else {
1211 		/*
1212 		 * Use the lower minor version returned in
1213 		 * the ack. By defninition, all lower minor
1214 		 * versions must be supported.
1215 		 */
1216 		svc->ver.minor = ack->minor_vers;
1217 	}
1218 
1219 	svc->state = DS_SVC_ACTIVE;
1220 	svc->port = port;
1221 
1222 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_ack: '%s' v%d.%d ready, hdl=0x%llx"
1223 	    DS_EOL, PORTID(port), svc->cap.svc_id, svc->ver.major,
1224 	    svc->ver.minor, (u_longlong_t)svc->hdl);
1225 
1226 	/* notify the client that registration is complete */
1227 	if (svc->ops.ds_reg_cb) {
1228 		/*
1229 		 * Use a temporary version structure so that
1230 		 * the copy in the svc structure cannot be
1231 		 * modified by the client.
1232 		 */
1233 		tmpver.major = svc->ver.major;
1234 		tmpver.minor = svc->ver.minor;
1235 
1236 		(*svc->ops.ds_reg_cb)(svc->ops.cb_arg, &tmpver, svc->hdl);
1237 	}
1238 
1239 done:
1240 	mutex_exit(&ds_svcs.lock);
1241 }
1242 
1243 static void
1244 ds_try_next_port(ds_svc_t *svc, int portid)
1245 {
1246 	ds_port_t *port;
1247 	ds_portset_t totry;
1248 	int i;
1249 
1250 	DS_DBG_LDC(CE_NOTE, "ds@%x %s" DS_EOL, portid, __func__);
1251 
1252 	/*
1253 	 * Get the ports that haven't been tried yet and are available to try.
1254 	 */
1255 	DS_PORTSET_DUP(totry, svc->avail);
1256 	for (i = 0; i < DS_MAX_PORTS; i++) {
1257 		if (DS_PORT_IN_SET(svc->tried, i))
1258 			DS_PORTSET_DEL(totry, i);
1259 	}
1260 
1261 	if (DS_PORTSET_ISNULL(totry))
1262 		return;
1263 
1264 	for (i = 0; i < DS_MAX_PORTS; i++, portid++) {
1265 		if (portid >= DS_MAX_PORTS) {
1266 			portid = 0;
1267 		}
1268 
1269 		/*
1270 		 * If the port is not in the available list,
1271 		 * it is not a candidate for registration.
1272 		 */
1273 		if (!DS_PORT_IN_SET(totry, portid)) {
1274 			continue;
1275 		}
1276 
1277 		port = &ds_ports[portid];
1278 		DS_DBG_LDC(CE_NOTE, "ds@%x: %s trying ldc.id: %d" DS_EOL,
1279 		    portid, __func__, (uint_t)(port->ldc.id));
1280 		if (ds_send_reg_req(svc, port) == 0) {
1281 			DS_DBG_LDC(CE_NOTE, "ds@%x: %s reg msg send OK" DS_EOL,
1282 			    portid, __func__);
1283 			/* register sent successfully */
1284 			break;
1285 		}
1286 		DS_DBG_LDC(CE_NOTE, "ds@%x: %s reg msg send FAIL" DS_EOL,
1287 		    portid, __func__);
1288 
1289 		/* reset the service to try the next port */
1290 		ds_reset_svc(svc, port);
1291 	}
1292 }
1293 
1294 static void
1295 ds_handle_reg_nack(ds_port_t *port, caddr_t buf, size_t len)
1296 {
1297 	ds_reg_nack_t	*nack;
1298 	ds_svc_t	*svc;
1299 	int		idx;
1300 	size_t		explen = DS_MSG_LEN(ds_reg_nack_t);
1301 
1302 	/* sanity check the incoming message */
1303 	if (len != explen) {
1304 		cmn_err(CE_WARN, "ds@%lx: <reg_nack: invalid message "
1305 		    "length (%ld), expected %ld" DS_EOL, PORTID(port), len,
1306 		    explen);
1307 		return;
1308 	}
1309 
1310 	nack = (ds_reg_nack_t *)(buf + DS_HDR_SZ);
1311 
1312 	mutex_enter(&ds_svcs.lock);
1313 
1314 	/*
1315 	 * We expect a reg_nack for a client ping.
1316 	 */
1317 	if (DS_HDL_ISCLIENT(nack->svc_handle)) {
1318 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_nack: ping hdl: 0x%llx"
1319 		    DS_EOL, PORTID(port), (u_longlong_t)nack->svc_handle);
1320 		goto done;
1321 	}
1322 
1323 	/*
1324 	 * This searches for service based on how we generate handles
1325 	 * and so only works because this is a reg nack.
1326 	 */
1327 	if ((svc = ds_get_svc(nack->svc_handle)) == NULL) {
1328 		cmn_err(CE_WARN, "ds@%lx: <reg_nack: invalid handle 0x%llx"
1329 		    DS_EOL, PORTID(port), (u_longlong_t)nack->svc_handle);
1330 		goto done;
1331 	}
1332 
1333 	/* make sure the message makes sense */
1334 	if (svc->state != DS_SVC_REG_PENDING) {
1335 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_nack: '%s' handle: 0x%llx "
1336 		    "invalid state (%d)" DS_EOL, PORTID(port), svc->cap.svc_id,
1337 		    (u_longlong_t)nack->svc_handle, svc->state);
1338 		goto done;
1339 	}
1340 
1341 	if (nack->result == DS_REG_DUP) {
1342 		cmn_err(CE_WARN, "ds@%lx: <reg_nack: duplicate registration "
1343 		    " for %s" DS_EOL, PORTID(port), svc->cap.svc_id);
1344 		ds_reset_svc(svc, port);
1345 		goto done;
1346 	}
1347 
1348 	/*
1349 	 * A major version of zero indicates that the
1350 	 * service is not supported at all.
1351 	 */
1352 	if (nack->major_vers == 0) {
1353 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_nack: '%s' not supported"
1354 		    DS_EOL, PORTID(port), svc->cap.svc_id);
1355 		ds_reset_svc(svc, port);
1356 		if ((svc->flags & DSSF_ISCLIENT) == 0)
1357 			ds_try_next_port(svc, PORTID(port) + 1);
1358 		goto done;
1359 	}
1360 
1361 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_nack: '%s' hdl=0x%llx, nack=%d.x"
1362 	    DS_EOL, PORTID(port), svc->cap.svc_id,
1363 	    (u_longlong_t)nack->svc_handle, nack->major_vers);
1364 
1365 	/*
1366 	 * Walk the version list for the service, looking for
1367 	 * a major version that is as close to the requested
1368 	 * major version as possible.
1369 	 */
1370 	for (idx = svc->ver_idx; idx < svc->cap.nvers; idx++) {
1371 		if (svc->cap.vers[idx].major <= nack->major_vers) {
1372 			/* found a version to try */
1373 			break;
1374 		}
1375 	}
1376 
1377 	if (idx == svc->cap.nvers) {
1378 		/* no supported version */
1379 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <reg_nack: %s v%d.x not supported"
1380 		    DS_EOL, PORTID(port), svc->cap.svc_id, nack->major_vers);
1381 		ds_reset_svc(svc, port);
1382 		if ((svc->flags & DSSF_ISCLIENT) == 0)
1383 			ds_try_next_port(svc, PORTID(port) + 1);
1384 		goto done;
1385 	}
1386 
1387 	/* start the handshake again */
1388 	svc->state = DS_SVC_INACTIVE;
1389 	svc->ver_idx = idx;
1390 
1391 	(void) ds_svc_register(svc, NULL);
1392 
1393 done:
1394 	mutex_exit(&ds_svcs.lock);
1395 }
1396 
1397 static void
1398 ds_handle_unreg_req(ds_port_t *port, caddr_t buf, size_t len)
1399 {
1400 	ds_hdr_t	*hdr;
1401 	ds_unreg_req_t	*req;
1402 	ds_unreg_ack_t	*ack;
1403 	ds_svc_t	*svc;
1404 	char		*msg;
1405 	size_t		msglen;
1406 	size_t		explen = DS_MSG_LEN(ds_unreg_req_t);
1407 	boolean_t	is_up;
1408 
1409 	/* sanity check the incoming message */
1410 	if (len != explen) {
1411 		cmn_err(CE_WARN, "ds@%lx: <unreg_req: invalid message "
1412 		    "length (%ld), expected %ld" DS_EOL, PORTID(port), len,
1413 		    explen);
1414 		return;
1415 	}
1416 
1417 	req = (ds_unreg_req_t *)(buf + DS_HDR_SZ);
1418 
1419 	mutex_enter(&ds_svcs.lock);
1420 
1421 	/* lookup appropriate client or service */
1422 	if (DS_HDL_ISCLIENT(req->svc_handle) ||
1423 	    ((svc = ds_find_clnt_svc_by_hdl_port(req->svc_handle, port))
1424 	    == NULL && ((svc = ds_get_svc(req->svc_handle)) == NULL ||
1425 	    svc->port != port))) {
1426 		mutex_exit(&ds_svcs.lock);
1427 		mutex_enter(&port->lock);
1428 		is_up = (port->ldc.state == LDC_UP);
1429 		mutex_exit(&port->lock);
1430 		if (!is_up)
1431 			return;
1432 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <unreg_req: invalid handle 0x%llx"
1433 		    DS_EOL, PORTID(port), (u_longlong_t)req->svc_handle);
1434 		ds_send_unreg_nack(port, req->svc_handle);
1435 		return;
1436 	}
1437 
1438 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <unreg_req: '%s' handle 0x%llx" DS_EOL,
1439 	    PORTID(port), svc->cap.svc_id, (u_longlong_t)req->svc_handle);
1440 
1441 	(void) ds_svc_unregister(svc, svc->port);
1442 
1443 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: unreg_ack>: '%s' hdl=0x%llx" DS_EOL,
1444 	    PORTID(port), svc->cap.svc_id, (u_longlong_t)req->svc_handle);
1445 
1446 	ds_check_for_dup_services(svc);
1447 
1448 	mutex_exit(&ds_svcs.lock);
1449 
1450 	msglen = DS_HDR_SZ + sizeof (ds_unreg_ack_t);
1451 	msg = DS_MALLOC(msglen);
1452 
1453 	hdr = (ds_hdr_t *)msg;
1454 	hdr->msg_type = DS_UNREG_ACK;
1455 	hdr->payload_len = sizeof (ds_unreg_ack_t);
1456 
1457 	ack = (ds_unreg_ack_t *)(msg + DS_HDR_SZ);
1458 	ack->svc_handle = req->svc_handle;
1459 
1460 	/* send message */
1461 	(void) ds_send_msg(port, msg, msglen);
1462 	DS_FREE(msg, msglen);
1463 
1464 }
1465 
1466 static void
1467 ds_handle_unreg_ack(ds_port_t *port, caddr_t buf, size_t len)
1468 {
1469 	ds_unreg_ack_t	*ack;
1470 	size_t		explen = DS_MSG_LEN(ds_unreg_ack_t);
1471 
1472 	/* sanity check the incoming message */
1473 	if (len != explen) {
1474 		cmn_err(CE_WARN, "ds@%lx: <unreg_ack: invalid message "
1475 		    "length (%ld), expected %ld" DS_EOL, PORTID(port), len,
1476 		    explen);
1477 		return;
1478 	}
1479 
1480 	ack = (ds_unreg_ack_t *)(buf + DS_HDR_SZ);
1481 
1482 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <unreg_ack: hdl=0x%llx" DS_EOL,
1483 	    PORTID(port), (u_longlong_t)ack->svc_handle);
1484 
1485 #ifdef DEBUG
1486 	mutex_enter(&ds_svcs.lock);
1487 
1488 	/*
1489 	 * Since the unregister request was initiated locally,
1490 	 * the service structure has already been torn down.
1491 	 * Just perform a sanity check to make sure the message
1492 	 * is appropriate.
1493 	 */
1494 	if (ds_get_svc(ack->svc_handle) != NULL) {
1495 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <unreg_ack: handle 0x%llx in use"
1496 		    DS_EOL, PORTID(port), (u_longlong_t)ack->svc_handle);
1497 	}
1498 
1499 	mutex_exit(&ds_svcs.lock);
1500 #endif	/* DEBUG */
1501 }
1502 
1503 static void
1504 ds_handle_unreg_nack(ds_port_t *port, caddr_t buf, size_t len)
1505 {
1506 	ds_unreg_nack_t	*nack;
1507 	size_t		explen = DS_MSG_LEN(ds_unreg_nack_t);
1508 
1509 	/* sanity check the incoming message */
1510 	if (len != explen) {
1511 		cmn_err(CE_WARN, "ds@%lx: <unreg_nack: invalid message "
1512 		    "length (%ld), expected %ld" DS_EOL, PORTID(port), len,
1513 		    explen);
1514 		return;
1515 	}
1516 
1517 	nack = (ds_unreg_nack_t *)(buf + DS_HDR_SZ);
1518 
1519 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <unreg_nack: hdl=0x%llx" DS_EOL,
1520 	    PORTID(port), (u_longlong_t)nack->svc_handle);
1521 
1522 #ifdef DEBUG
1523 	mutex_enter(&ds_svcs.lock);
1524 
1525 	/*
1526 	 * Since the unregister request was initiated locally,
1527 	 * the service structure has already been torn down.
1528 	 * Just perform a sanity check to make sure the message
1529 	 * is appropriate.
1530 	 */
1531 	if (ds_get_svc(nack->svc_handle) != NULL) {
1532 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: <unreg_nack: handle 0x%llx in use"
1533 		    DS_EOL, PORTID(port), (u_longlong_t)nack->svc_handle);
1534 	}
1535 
1536 	mutex_exit(&ds_svcs.lock);
1537 #endif	/* DEBUG */
1538 }
1539 
1540 static void
1541 ds_handle_data(ds_port_t *port, caddr_t buf, size_t len)
1542 {
1543 	ds_data_handle_t	*data;
1544 	ds_svc_t		*svc;
1545 	char			*msg;
1546 	int			msgsz;
1547 	int			hdrsz;
1548 	size_t			explen = DS_MSG_LEN(ds_data_handle_t);
1549 
1550 	/* sanity check the incoming message */
1551 	if (len < explen) {
1552 		cmn_err(CE_WARN, "ds@%lx: <data: invalid message length "
1553 		    "(%ld), expected at least %ld" DS_EOL, PORTID(port), len,
1554 		    explen);
1555 		return;
1556 	}
1557 
1558 	data = (ds_data_handle_t *)(buf + DS_HDR_SZ);
1559 
1560 	hdrsz = DS_HDR_SZ + sizeof (ds_data_handle_t);
1561 	msgsz = len - hdrsz;
1562 
1563 	/* strip off the header for the client */
1564 	msg = (msgsz) ? (buf + hdrsz) : NULL;
1565 
1566 	mutex_enter(&ds_svcs.lock);
1567 
1568 	if ((svc = ds_find_clnt_svc_by_hdl_port(data->svc_handle, port))
1569 	    == NULL) {
1570 		if ((svc = ds_get_svc(data->svc_handle)) == NULL) {
1571 			mutex_exit(&ds_svcs.lock);
1572 			cmn_err(CE_WARN, "ds@%lx: <data: invalid handle 0x%llx"
1573 			    DS_EOL, PORTID(port),
1574 			    (u_longlong_t)data->svc_handle);
1575 			ds_send_data_nack(port, data->svc_handle);
1576 			return;
1577 		}
1578 	}
1579 
1580 	mutex_exit(&ds_svcs.lock);
1581 
1582 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: <data: '%s' hdl=0x%llx" DS_EOL,
1583 	    PORTID(port), svc->cap.svc_id, (u_longlong_t)svc->hdl);
1584 	DS_DUMP_MSG(DS_DBG_FLAG_PRCL, msg, msgsz);
1585 
1586 	/* dispatch this message to the client */
1587 	(*svc->ops.ds_data_cb)(svc->ops.cb_arg, msg, msgsz);
1588 }
1589 
1590 static void
1591 ds_handle_nack(ds_port_t *port, caddr_t buf, size_t len)
1592 {
1593 	ds_svc_t	*svc;
1594 	ds_data_nack_t	*nack;
1595 	size_t		explen = DS_MSG_LEN(ds_data_nack_t);
1596 
1597 	/* sanity check the incoming message */
1598 	if (len != explen) {
1599 		cmn_err(CE_WARN, "ds@%lx: <data_nack: invalid message "
1600 		    "length (%ld), expected %ld" DS_EOL, PORTID(port), len,
1601 		    explen);
1602 		return;
1603 	}
1604 
1605 	nack = (ds_data_nack_t *)(buf + DS_HDR_SZ);
1606 
1607 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: data_nack: hdl=0x%llx, result=0x%llx"
1608 	    DS_EOL, PORTID(port), (u_longlong_t)nack->svc_handle,
1609 	    (u_longlong_t)nack->result);
1610 
1611 	if (nack->result == DS_INV_HDL) {
1612 
1613 		mutex_enter(&ds_svcs.lock);
1614 
1615 		if ((svc = ds_find_clnt_svc_by_hdl_port(nack->svc_handle,
1616 		    port)) == NULL) {
1617 			if ((svc = ds_get_svc(nack->svc_handle)) == NULL) {
1618 				mutex_exit(&ds_svcs.lock);
1619 				return;
1620 			}
1621 		}
1622 
1623 		cmn_err(CE_WARN, "ds@%lx: <data_nack: handle 0x%llx reported "
1624 		    " as invalid" DS_EOL, PORTID(port),
1625 		    (u_longlong_t)nack->svc_handle);
1626 
1627 		(void) ds_svc_unregister(svc, svc->port);
1628 
1629 		mutex_exit(&ds_svcs.lock);
1630 	}
1631 }
1632 
1633 /* Initialize the port */
1634 void
1635 ds_send_init_req(ds_port_t *port)
1636 {
1637 	ds_hdr_t	*hdr;
1638 	ds_init_req_t	*init_req;
1639 	size_t		msglen;
1640 	ds_ver_t	*vers = &ds_vers[port->ver_idx];
1641 
1642 	mutex_enter(&port->lock);
1643 	if (port->state != DS_PORT_LDC_INIT) {
1644 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: init_req>: invalid state: %d"
1645 		    DS_EOL, PORTID(port), port->state);
1646 		mutex_exit(&port->lock);
1647 		return;
1648 	}
1649 	mutex_exit(&port->lock);
1650 
1651 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: init_req>: req=v%d.%d" DS_EOL,
1652 	    PORTID(port), vers->major, vers->minor);
1653 
1654 	msglen = DS_HDR_SZ + sizeof (ds_init_req_t);
1655 	hdr = DS_MALLOC(msglen);
1656 
1657 	hdr->msg_type = DS_INIT_REQ;
1658 	hdr->payload_len = sizeof (ds_init_req_t);
1659 
1660 	init_req = (ds_init_req_t *)((caddr_t)hdr + DS_HDR_SZ);
1661 	init_req->major_vers = vers->major;
1662 	init_req->minor_vers = vers->minor;
1663 
1664 	if (ds_send_msg(port, (caddr_t)hdr, msglen) == 0) {
1665 		/*
1666 		 * We've left the port state unlocked over the malloc/send,
1667 		 * make sure no one has changed the state under us before
1668 		 * we update the state.
1669 		 */
1670 		mutex_enter(&port->lock);
1671 		if (port->state == DS_PORT_LDC_INIT)
1672 			port->state = DS_PORT_INIT_REQ;
1673 		mutex_exit(&port->lock);
1674 	}
1675 	DS_FREE(hdr, msglen);
1676 }
1677 
1678 static int
1679 ds_send_reg_req(ds_svc_t *svc, ds_port_t *port)
1680 {
1681 	ds_ver_t	*ver;
1682 	ds_hdr_t	*hdr;
1683 	caddr_t		msg;
1684 	size_t		msglen;
1685 	ds_reg_req_t	*req;
1686 	size_t		idlen;
1687 	int		rv;
1688 
1689 	if ((svc->state != DS_SVC_INACTIVE) &&
1690 	    ((svc->flags & DSSF_ISCLIENT) == 0)) {
1691 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: reg_req>: invalid svc state (%d) "
1692 		    "for svc '%s'" DS_EOL, PORTID(port), svc->state,
1693 		    svc->cap.svc_id);
1694 		return (-1);
1695 	}
1696 
1697 	mutex_enter(&port->lock);
1698 
1699 	/* check on the LDC to Zeus */
1700 	if (port->ldc.state != LDC_UP) {
1701 		/* can not send message */
1702 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: reg_req>: channel %ld is not up"
1703 		    DS_EOL, PORTID(port), port->ldc.id);
1704 		mutex_exit(&port->lock);
1705 		return (-1);
1706 	}
1707 
1708 	/* make sure port is ready */
1709 	if (port->state != DS_PORT_READY) {
1710 		/* can not send message */
1711 		DS_DBG_PRCL(CE_NOTE, "ds@%lx: reg_req>: port is not ready"
1712 		    DS_EOL, PORTID(port));
1713 		mutex_exit(&port->lock);
1714 		return (-1);
1715 	}
1716 
1717 	mutex_exit(&port->lock);
1718 
1719 	/* allocate the message buffer */
1720 	idlen = strlen(svc->cap.svc_id);
1721 	msglen = DS_HDR_SZ + sizeof (ds_reg_req_t) + idlen;
1722 	msg = DS_MALLOC(msglen);
1723 
1724 	/* copy in the header data */
1725 	hdr = (ds_hdr_t *)msg;
1726 	hdr->msg_type = DS_REG_REQ;
1727 	hdr->payload_len = sizeof (ds_reg_req_t) + idlen;
1728 
1729 	req = (ds_reg_req_t *)(msg + DS_HDR_SZ);
1730 	req->svc_handle = svc->hdl;
1731 	ver = &(svc->cap.vers[svc->ver_idx]);
1732 	req->major_vers = ver->major;
1733 	req->minor_vers = ver->minor;
1734 
1735 	/* copy in the service id */
1736 	(void) memcpy(req->svc_id, svc->cap.svc_id, idlen + 1);
1737 
1738 	/* send the message */
1739 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: reg_req>: '%s' ver=%d.%d, hdl=0x%llx"
1740 	    DS_EOL, PORTID(port), svc->cap.svc_id, ver->major, ver->minor,
1741 	    (u_longlong_t)svc->hdl);
1742 
1743 	if ((rv = ds_send_msg(port, msg, msglen)) != 0) {
1744 		svc->port = port;
1745 		rv = -1;
1746 	} else if ((svc->flags & DSSF_ISCLIENT) == 0) {
1747 		svc->state = DS_SVC_REG_PENDING;
1748 	}
1749 	DS_FREE(msg, msglen);
1750 
1751 	return (rv);
1752 }
1753 
1754 /*
1755  * Keep around in case we want this later
1756  */
1757 int
1758 ds_send_unreg_req(ds_svc_t *svc)
1759 {
1760 	caddr_t		msg;
1761 	size_t		msglen;
1762 	ds_hdr_t	*hdr;
1763 	ds_unreg_req_t	*req;
1764 	ds_port_t	*port = svc->port;
1765 	int		rv;
1766 
1767 	if (port == NULL) {
1768 		DS_DBG(CE_NOTE, "send_unreg_req: service '%s' not "
1769 		    "associated with a port" DS_EOL, svc->cap.svc_id);
1770 		return (-1);
1771 	}
1772 
1773 	mutex_enter(&port->lock);
1774 
1775 	/* check on the LDC to Zeus */
1776 	if (port->ldc.state != LDC_UP) {
1777 		/* can not send message */
1778 		cmn_err(CE_WARN, "ds@%lx: unreg_req>: channel %ld is not up"
1779 		    DS_EOL, PORTID(port), port->ldc.id);
1780 		mutex_exit(&port->lock);
1781 		return (-1);
1782 	}
1783 
1784 	/* make sure port is ready */
1785 	if (port->state != DS_PORT_READY) {
1786 		/* can not send message */
1787 		cmn_err(CE_WARN, "ds@%lx: unreg_req>: port is not ready" DS_EOL,
1788 		    PORTID(port));
1789 		mutex_exit(&port->lock);
1790 		return (-1);
1791 	}
1792 
1793 	mutex_exit(&port->lock);
1794 
1795 	msglen = DS_HDR_SZ + sizeof (ds_unreg_req_t);
1796 	msg = DS_MALLOC(msglen);
1797 
1798 	/* copy in the header data */
1799 	hdr = (ds_hdr_t *)msg;
1800 	hdr->msg_type = DS_UNREG;
1801 	hdr->payload_len = sizeof (ds_unreg_req_t);
1802 
1803 	req = (ds_unreg_req_t *)(msg + DS_HDR_SZ);
1804 	if (svc->flags & DSSF_ISCLIENT) {
1805 		req->svc_handle = svc->svc_hdl;
1806 	} else {
1807 		req->svc_handle = svc->hdl;
1808 	}
1809 
1810 	/* send the message */
1811 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: unreg_req>: '%s' hdl=0x%llx" DS_EOL,
1812 	    PORTID(port), (svc->cap.svc_id) ? svc->cap.svc_id : "NULL",
1813 	    (u_longlong_t)svc->hdl);
1814 
1815 	if ((rv = ds_send_msg(port, msg, msglen)) != 0) {
1816 		rv = -1;
1817 	}
1818 	DS_FREE(msg, msglen);
1819 
1820 	return (rv);
1821 }
1822 
1823 static void
1824 ds_send_unreg_nack(ds_port_t *port, ds_svc_hdl_t bad_hdl)
1825 {
1826 	caddr_t		msg;
1827 	size_t		msglen;
1828 	ds_hdr_t	*hdr;
1829 	ds_unreg_nack_t	*nack;
1830 
1831 	mutex_enter(&port->lock);
1832 
1833 	/* check on the LDC to Zeus */
1834 	if (port->ldc.state != LDC_UP) {
1835 		/* can not send message */
1836 		cmn_err(CE_WARN, "ds@%lx: unreg_nack>: channel %ld is not up"
1837 		    DS_EOL, PORTID(port), port->ldc.id);
1838 		mutex_exit(&port->lock);
1839 		return;
1840 	}
1841 
1842 	/* make sure port is ready */
1843 	if (port->state != DS_PORT_READY) {
1844 		/* can not send message */
1845 		cmn_err(CE_WARN, "ds@%lx: unreg_nack>: port is not ready"
1846 		    DS_EOL, PORTID(port));
1847 		mutex_exit(&port->lock);
1848 		return;
1849 	}
1850 
1851 	mutex_exit(&port->lock);
1852 
1853 	msglen = DS_HDR_SZ + sizeof (ds_unreg_nack_t);
1854 	msg = DS_MALLOC(msglen);
1855 
1856 	/* copy in the header data */
1857 	hdr = (ds_hdr_t *)msg;
1858 	hdr->msg_type = DS_UNREG_NACK;
1859 	hdr->payload_len = sizeof (ds_unreg_nack_t);
1860 
1861 	nack = (ds_unreg_nack_t *)(msg + DS_HDR_SZ);
1862 	nack->svc_handle = bad_hdl;
1863 
1864 	/* send the message */
1865 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: unreg_nack>: hdl=0x%llx" DS_EOL,
1866 	    PORTID(port), (u_longlong_t)bad_hdl);
1867 
1868 	(void) ds_send_msg(port, msg, msglen);
1869 	DS_FREE(msg, msglen);
1870 }
1871 
1872 static void
1873 ds_send_data_nack(ds_port_t *port, ds_svc_hdl_t bad_hdl)
1874 {
1875 	caddr_t		msg;
1876 	size_t		msglen;
1877 	ds_hdr_t	*hdr;
1878 	ds_data_nack_t	*nack;
1879 
1880 	mutex_enter(&port->lock);
1881 
1882 	/* check on the LDC to Zeus */
1883 	if (port->ldc.state != LDC_UP) {
1884 		/* can not send message */
1885 		cmn_err(CE_WARN, "ds@%lx: data_nack>: channel %ld is not up"
1886 		    DS_EOL, PORTID(port), port->ldc.id);
1887 		mutex_exit(&port->lock);
1888 		return;
1889 	}
1890 
1891 	/* make sure port is ready */
1892 	if (port->state != DS_PORT_READY) {
1893 		/* can not send message */
1894 		cmn_err(CE_WARN, "ds@%lx: data_nack>: port is not ready" DS_EOL,
1895 		    PORTID(port));
1896 		mutex_exit(&port->lock);
1897 		return;
1898 	}
1899 
1900 	mutex_exit(&port->lock);
1901 
1902 	msglen = DS_HDR_SZ + sizeof (ds_data_nack_t);
1903 	msg = DS_MALLOC(msglen);
1904 
1905 	/* copy in the header data */
1906 	hdr = (ds_hdr_t *)msg;
1907 	hdr->msg_type = DS_NACK;
1908 	hdr->payload_len = sizeof (ds_data_nack_t);
1909 
1910 	nack = (ds_data_nack_t *)(msg + DS_HDR_SZ);
1911 	nack->svc_handle = bad_hdl;
1912 	nack->result = DS_INV_HDL;
1913 
1914 	/* send the message */
1915 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: data_nack>: hdl=0x%llx" DS_EOL,
1916 	    PORTID(port), (u_longlong_t)bad_hdl);
1917 
1918 	(void) ds_send_msg(port, msg, msglen);
1919 	DS_FREE(msg, msglen);
1920 }
1921 
1922 /* END DS PROTOCOL SUPPORT FUNCTIONS */
1923 
1924 #ifdef DEBUG
1925 
1926 #define	BYTESPERLINE	8
1927 #define	LINEWIDTH	((BYTESPERLINE * 3) + (BYTESPERLINE + 2) + 1)
1928 #define	ASCIIOFFSET	((BYTESPERLINE * 3) + 2)
1929 #define	ISPRINT(c)	((c >= ' ') && (c <= '~'))
1930 
1931 /*
1932  * Output a buffer formatted with a set number of bytes on
1933  * each line. Append each line with the ASCII equivalent of
1934  * each byte if it falls within the printable ASCII range,
1935  * and '.' otherwise.
1936  */
1937 void
1938 ds_dump_msg(void *vbuf, size_t len)
1939 {
1940 	int	i, j;
1941 	char	*curr;
1942 	char	*aoff;
1943 	char	line[LINEWIDTH];
1944 	uint8_t	*buf = vbuf;
1945 
1946 	if (len > 128)
1947 		len = 128;
1948 
1949 	/* walk the buffer one line at a time */
1950 	for (i = 0; i < len; i += BYTESPERLINE) {
1951 
1952 		bzero(line, LINEWIDTH);
1953 
1954 		curr = line;
1955 		aoff = line + ASCIIOFFSET;
1956 
1957 		/*
1958 		 * Walk the bytes in the current line, storing
1959 		 * the hex value for the byte as well as the
1960 		 * ASCII representation in a temporary buffer.
1961 		 * All ASCII values are placed at the end of
1962 		 * the line.
1963 		 */
1964 		for (j = 0; (j < BYTESPERLINE) && ((i + j) < len); j++) {
1965 			(void) sprintf(curr, " %02x", buf[i + j]);
1966 			*aoff = (ISPRINT(buf[i + j])) ? buf[i + j] : '.';
1967 			curr += 3;
1968 			aoff++;
1969 		}
1970 
1971 		/*
1972 		 * Fill in to the start of the ASCII translation
1973 		 * with spaces. This will only be necessary if
1974 		 * this is the last line and there are not enough
1975 		 * bytes to fill the whole line.
1976 		 */
1977 		while (curr != (line + ASCIIOFFSET))
1978 			*curr++ = ' ';
1979 
1980 		cmn_err(CE_NOTE, "%s" DS_EOL, line);
1981 	}
1982 }
1983 #endif /* DEBUG */
1984 
1985 
1986 /*
1987  * Walk the table of registered services, executing the specified callback
1988  * function for each service on a port. A non-zero return value from the
1989  * callback is used to terminate the walk, not to indicate an error. Returns
1990  * the index of the last service visited.
1991  */
1992 int
1993 ds_walk_svcs(svc_cb_t svc_cb, void *arg)
1994 {
1995 	int		idx;
1996 	ds_svc_t	*svc;
1997 
1998 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
1999 
2000 	/* walk every table entry */
2001 	for (idx = 0; idx < ds_svcs.maxsvcs; idx++) {
2002 		svc = ds_svcs.tbl[idx];
2003 
2004 		/* execute the callback */
2005 		if ((*svc_cb)(svc, arg) != 0)
2006 			break;
2007 	}
2008 
2009 	return (idx);
2010 }
2011 
2012 static int
2013 ds_svc_isfree(ds_svc_t *svc, void *arg)
2014 {
2015 	_NOTE(ARGUNUSED(arg))
2016 
2017 	/*
2018 	 * Looking for a free service. This may be a NULL entry
2019 	 * in the table, or an unused structure that could be
2020 	 * reused.
2021 	 */
2022 
2023 	if (DS_SVC_ISFREE(svc)) {
2024 		/* yes, it is free */
2025 		return (1);
2026 	}
2027 
2028 	/* not a candidate */
2029 	return (0);
2030 }
2031 
2032 int
2033 ds_svc_ismatch(ds_svc_t *svc, void *arg)
2034 {
2035 	if (DS_SVC_ISFREE(svc)) {
2036 		return (0);
2037 	}
2038 
2039 	if (strcmp(svc->cap.svc_id, arg) == 0 &&
2040 	    (svc->flags & DSSF_ISCLIENT) == 0) {
2041 		/* found a match */
2042 		return (1);
2043 	}
2044 
2045 	return (0);
2046 }
2047 
2048 int
2049 ds_svc_clnt_ismatch(ds_svc_t *svc, void *arg)
2050 {
2051 	if (DS_SVC_ISFREE(svc)) {
2052 		return (0);
2053 	}
2054 
2055 	if (strcmp(svc->cap.svc_id, arg) == 0 &&
2056 	    (svc->flags & DSSF_ISCLIENT) != 0) {
2057 		/* found a match */
2058 		return (1);
2059 	}
2060 
2061 	return (0);
2062 }
2063 
2064 int
2065 ds_svc_free(ds_svc_t *svc, void *arg)
2066 {
2067 	_NOTE(ARGUNUSED(arg))
2068 
2069 	if (svc == NULL) {
2070 		return (0);
2071 	}
2072 
2073 	if (svc->cap.svc_id) {
2074 		DS_FREE(svc->cap.svc_id, strlen(svc->cap.svc_id) + 1);
2075 		svc->cap.svc_id = NULL;
2076 	}
2077 
2078 	if (svc->cap.vers) {
2079 		DS_FREE(svc->cap.vers, svc->cap.nvers * sizeof (ds_ver_t));
2080 		svc->cap.vers = NULL;
2081 	}
2082 
2083 	DS_FREE(svc, sizeof (ds_svc_t));
2084 
2085 	return (0);
2086 }
2087 
2088 static void
2089 ds_set_svc_port_tried(char *svc_id, ds_port_t *port)
2090 {
2091 	int		idx;
2092 	ds_svc_t	*svc;
2093 
2094 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
2095 
2096 	/* walk every table entry */
2097 	for (idx = 0; idx < ds_svcs.maxsvcs; idx++) {
2098 		svc = ds_svcs.tbl[idx];
2099 		if (!DS_SVC_ISFREE(svc) && (svc->flags & DSSF_ISCLIENT) != 0 &&
2100 		    strcmp(svc_id, svc->cap.svc_id) == 0)
2101 			DS_PORTSET_ADD(svc->tried, PORTID(port));
2102 	}
2103 }
2104 
2105 static int
2106 ds_svc_register_onport(ds_svc_t *svc, ds_port_t *port)
2107 {
2108 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
2109 
2110 	if (DS_SVC_ISFREE(svc))
2111 		return (0);
2112 
2113 	if (!DS_PORT_IN_SET(svc->avail, PORTID(port)))
2114 		return (0);
2115 
2116 	if (DS_PORT_IN_SET(svc->tried, PORTID(port)))
2117 		return (0);
2118 
2119 	if ((svc->flags & DSSF_ISCLIENT) == 0) {
2120 		DS_PORTSET_ADD(svc->tried, PORTID(port));
2121 		if (svc->state != DS_SVC_INACTIVE)
2122 			return (0);
2123 	} else {
2124 		ds_set_svc_port_tried(svc->cap.svc_id, port);
2125 
2126 		/*
2127 		 * Never send a client reg req to the SP.
2128 		 */
2129 		if (PORTID(port) == ds_sp_port_id) {
2130 			return (0);
2131 		}
2132 	}
2133 
2134 	if (ds_send_reg_req(svc, port) == 0) {
2135 		/* register sent successfully */
2136 		return (1);
2137 	}
2138 
2139 	if ((svc->flags & DSSF_ISCLIENT) == 0) {
2140 		/* reset the service */
2141 		ds_reset_svc(svc, port);
2142 	}
2143 	return (0);
2144 }
2145 
2146 static int
2147 ds_svc_register_onport_walker(ds_svc_t *svc, void *arg)
2148 {
2149 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
2150 
2151 	if (DS_SVC_ISFREE(svc))
2152 		return (0);
2153 
2154 	(void) ds_svc_register_onport(svc, arg);
2155 	return (0);
2156 }
2157 
2158 int
2159 ds_svc_register(ds_svc_t *svc, void *arg)
2160 {
2161 	_NOTE(ARGUNUSED(arg))
2162 	ds_portset_t ports;
2163 	ds_port_t *port;
2164 	int	idx;
2165 
2166 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
2167 
2168 	if (DS_SVC_ISFREE(svc))
2169 		return (0);
2170 
2171 	DS_PORTSET_DUP(ports, svc->avail);
2172 	if (svc->flags & DSSF_ISCLIENT) {
2173 		for (idx = 0; idx < DS_MAX_PORTS; idx++) {
2174 			if (DS_PORT_IN_SET(svc->tried, idx))
2175 				DS_PORTSET_DEL(ports, idx);
2176 		}
2177 	} else if (svc->state != DS_SVC_INACTIVE)
2178 		return (0);
2179 
2180 	if (DS_PORTSET_ISNULL(ports))
2181 		return (0);
2182 
2183 	/*
2184 	 * Attempt to register the service. Start with the lowest
2185 	 * numbered port and continue until a registration message
2186 	 * is sent successfully, or there are no ports left to try.
2187 	 */
2188 	for (idx = 0; idx < DS_MAX_PORTS; idx++) {
2189 
2190 		/*
2191 		 * If the port is not in the available list,
2192 		 * it is not a candidate for registration.
2193 		 */
2194 		if (!DS_PORT_IN_SET(ports, idx)) {
2195 			continue;
2196 		}
2197 
2198 		port = &ds_ports[idx];
2199 		if (ds_svc_register_onport(svc, port)) {
2200 			if ((svc->flags & DSSF_ISCLIENT) == 0)
2201 				break;
2202 		}
2203 	}
2204 
2205 	return (0);
2206 }
2207 
2208 static int
2209 ds_svc_unregister(ds_svc_t *svc, void *arg)
2210 {
2211 	ds_port_t *port = (ds_port_t *)arg;
2212 	ds_svc_hdl_t hdl;
2213 
2214 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
2215 
2216 	if (DS_SVC_ISFREE(svc)) {
2217 		return (0);
2218 	}
2219 
2220 	/* make sure the service is using this port */
2221 	if (svc->port != port) {
2222 		return (0);
2223 	}
2224 
2225 	if (port) {
2226 		DS_DBG(CE_NOTE, "ds@%lx: svc_unreg: id='%s', ver=%d.%d, "
2227 		    " hdl=0x%09lx" DS_EOL, PORTID(port), svc->cap.svc_id,
2228 		    svc->ver.major, svc->ver.minor, svc->hdl);
2229 	} else {
2230 		DS_DBG(CE_NOTE, "port=NULL: svc_unreg: id='%s', ver=%d.%d, "
2231 		    " hdl=0x%09lx" DS_EOL, svc->cap.svc_id, svc->ver.major,
2232 		    svc->ver.minor, svc->hdl);
2233 	}
2234 
2235 	/* reset the service structure */
2236 	ds_reset_svc(svc, port);
2237 
2238 	/* call the client unregister callback */
2239 	if (svc->ops.ds_unreg_cb) {
2240 		(*svc->ops.ds_unreg_cb)(svc->ops.cb_arg);
2241 	}
2242 
2243 	/* increment the count in the handle to prevent reuse */
2244 	hdl = DS_ALLOC_HDL(DS_HDL2IDX(svc->hdl), DS_HDL2COUNT(svc->hdl));
2245 	if (DS_HDL_ISCLIENT(svc->hdl)) {
2246 		DS_HDL_SET_ISCLIENT(hdl);
2247 	}
2248 	svc->hdl = hdl;
2249 
2250 	if (svc->state != DS_SVC_UNREG_PENDING) {
2251 		/* try to initiate a new registration */
2252 		(void) ds_svc_register(svc, NULL);
2253 	}
2254 
2255 	return (0);
2256 }
2257 
2258 static int
2259 ds_svc_port_up(ds_svc_t *svc, void *arg)
2260 {
2261 	ds_port_t *port = (ds_port_t *)arg;
2262 
2263 	if (DS_SVC_ISFREE(svc)) {
2264 		/* nothing to do */
2265 		return (0);
2266 	}
2267 
2268 	DS_PORTSET_ADD(svc->avail, port->id);
2269 	DS_PORTSET_DEL(svc->tried, port->id);
2270 
2271 	return (0);
2272 }
2273 
2274 static void
2275 ds_set_port_ready(ds_port_t *port, uint16_t major, uint16_t minor)
2276 {
2277 	boolean_t was_ready;
2278 
2279 	mutex_enter(&port->lock);
2280 	was_ready = (port->state == DS_PORT_READY);
2281 	if (!was_ready) {
2282 		port->state = DS_PORT_READY;
2283 		port->ver.major = major;
2284 		port->ver.minor = minor;
2285 	}
2286 	mutex_exit(&port->lock);
2287 
2288 	if (!was_ready) {
2289 
2290 		/*
2291 		 * The port came up, so update all the services
2292 		 * with this information. Follow that up with an
2293 		 * attempt to register any service that is not
2294 		 * already registered.
2295 		 */
2296 		mutex_enter(&ds_svcs.lock);
2297 
2298 		(void) ds_walk_svcs(ds_svc_port_up, port);
2299 		(void) ds_walk_svcs(ds_svc_register_onport_walker, port);
2300 
2301 		mutex_exit(&ds_svcs.lock);
2302 	}
2303 }
2304 
2305 ds_svc_t *
2306 ds_alloc_svc(void)
2307 {
2308 	int		idx;
2309 	uint_t		newmaxsvcs;
2310 	ds_svc_t	**newtbl;
2311 	ds_svc_t	*newsvc;
2312 
2313 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
2314 
2315 	idx = ds_walk_svcs(ds_svc_isfree, NULL);
2316 
2317 	if (idx != ds_svcs.maxsvcs) {
2318 		goto found;
2319 	}
2320 
2321 	/*
2322 	 * There was no free space in the table. Grow
2323 	 * the table to double its current size.
2324 	 */
2325 	newmaxsvcs = ds_svcs.maxsvcs * 2;
2326 	newtbl = DS_MALLOC(newmaxsvcs * sizeof (ds_svc_t *));
2327 
2328 	/* copy old table data to the new table */
2329 	(void) memcpy(newtbl, ds_svcs.tbl,
2330 	    ds_svcs.maxsvcs * sizeof (ds_svc_t *));
2331 
2332 	/* clean up the old table */
2333 	DS_FREE(ds_svcs.tbl, ds_svcs.maxsvcs * sizeof (ds_svc_t *));
2334 	ds_svcs.tbl = newtbl;
2335 	ds_svcs.maxsvcs = newmaxsvcs;
2336 
2337 	/* search for a free space again */
2338 	idx = ds_walk_svcs(ds_svc_isfree, NULL);
2339 
2340 	/* the table is locked so should find a free slot */
2341 	ASSERT(idx != ds_svcs.maxsvcs);
2342 
2343 found:
2344 	/* allocate a new svc structure if necessary */
2345 	if ((newsvc = ds_svcs.tbl[idx]) == NULL) {
2346 		/* allocate a new service */
2347 		newsvc = DS_MALLOC(sizeof (ds_svc_t));
2348 		ds_svcs.tbl[idx] = newsvc;
2349 	}
2350 
2351 	/* fill in the handle */
2352 	newsvc->hdl = DS_ALLOC_HDL(idx, DS_HDL2COUNT(newsvc->hdl));
2353 	newsvc->state = DS_SVC_FREE;	/* Mark as free temporarily */
2354 
2355 	return (newsvc);
2356 }
2357 
2358 static void
2359 ds_reset_svc(ds_svc_t *svc, ds_port_t *port)
2360 {
2361 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
2362 
2363 	if (svc->state != DS_SVC_UNREG_PENDING)
2364 		svc->state = DS_SVC_INACTIVE;
2365 	svc->ver_idx = 0;
2366 	svc->ver.major = 0;
2367 	svc->ver.minor = 0;
2368 	svc->port = NULL;
2369 	if (port) {
2370 		DS_PORTSET_DEL(svc->avail, port->id);
2371 	}
2372 }
2373 
2374 ds_svc_t *
2375 ds_get_svc(ds_svc_hdl_t hdl)
2376 {
2377 	int		idx;
2378 	ds_svc_t	*svc;
2379 
2380 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
2381 
2382 	if (hdl == DS_INVALID_HDL)
2383 		return (NULL);
2384 
2385 	idx = DS_HDL2IDX(hdl);
2386 
2387 	/* check if index is out of bounds */
2388 	if ((idx < 0) || (idx >= ds_svcs.maxsvcs))
2389 		return (NULL);
2390 
2391 	svc = ds_svcs.tbl[idx];
2392 
2393 	/* check for a valid service */
2394 	if (DS_SVC_ISFREE(svc))
2395 		return (NULL);
2396 
2397 	/* make sure the handle is an exact match */
2398 	if (svc->hdl != hdl)
2399 		return (NULL);
2400 
2401 	return (svc);
2402 }
2403 
2404 static void
2405 ds_port_reset(ds_port_t *port)
2406 {
2407 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
2408 	ASSERT(MUTEX_HELD(&port->lock));
2409 
2410 	/* connection went down, mark everything inactive */
2411 	(void) ds_walk_svcs(ds_svc_unregister, port);
2412 
2413 	port->ver_idx = 0;
2414 	port->ver.major = 0;
2415 	port->ver.minor = 0;
2416 	port->state = DS_PORT_LDC_INIT;
2417 }
2418 
2419 /*
2420  * Verify that a version array is sorted as expected for the
2421  * version negotiation to work correctly.
2422  */
2423 ds_vers_check_t
2424 ds_vers_isvalid(ds_ver_t *vers, int nvers)
2425 {
2426 	uint16_t	curr_major;
2427 	uint16_t	curr_minor;
2428 	int		idx;
2429 
2430 	curr_major = vers[0].major;
2431 	curr_minor = vers[0].minor;
2432 
2433 	/*
2434 	 * Walk the version array, verifying correct ordering.
2435 	 * The array must be sorted from highest supported
2436 	 * version to lowest supported version.
2437 	 */
2438 	for (idx = 0; idx < nvers; idx++) {
2439 		if (vers[idx].major > curr_major) {
2440 			DS_DBG(CE_NOTE, "ds_vers_isvalid: version array has "
2441 			    " increasing major versions" DS_EOL);
2442 			return (DS_VERS_INCREASING_MAJOR_ERR);
2443 		}
2444 
2445 		if (vers[idx].major < curr_major) {
2446 			curr_major = vers[idx].major;
2447 			curr_minor = vers[idx].minor;
2448 			continue;
2449 		}
2450 
2451 		if (vers[idx].minor > curr_minor) {
2452 			DS_DBG(CE_NOTE, "ds_vers_isvalid: version array has "
2453 			    " increasing minor versions" DS_EOL);
2454 			return (DS_VERS_INCREASING_MINOR_ERR);
2455 		}
2456 
2457 		curr_minor = vers[idx].minor;
2458 	}
2459 
2460 	return (DS_VERS_OK);
2461 }
2462 
2463 /*
2464  * Extended user capability init.
2465  */
2466 int
2467 ds_ucap_init(ds_capability_t *cap, ds_clnt_ops_t *ops, uint32_t flags,
2468     int instance, ds_svc_hdl_t *hdlp)
2469 {
2470 	ds_vers_check_t	status;
2471 	ds_svc_t	*svc;
2472 	int		rv = 0;
2473 	ds_svc_hdl_t 	lb_hdl, hdl;
2474 	int		is_loopback;
2475 	int		is_client;
2476 
2477 	/* sanity check the args */
2478 	if ((cap == NULL) || (ops == NULL)) {
2479 		cmn_err(CE_NOTE, "%s: invalid arguments" DS_EOL, __func__);
2480 		return (EINVAL);
2481 	}
2482 
2483 	/* sanity check the capability specifier */
2484 	if ((cap->svc_id == NULL) || (cap->vers == NULL) || (cap->nvers == 0)) {
2485 		cmn_err(CE_NOTE, "%s: invalid capability specifier" DS_EOL,
2486 		    __func__);
2487 		return (EINVAL);
2488 	}
2489 
2490 	/* sanity check the version array */
2491 	if ((status = ds_vers_isvalid(cap->vers, cap->nvers)) != DS_VERS_OK) {
2492 		cmn_err(CE_NOTE, "%s: invalid capability version array "
2493 		    "for %s service: %s" DS_EOL, __func__, cap->svc_id,
2494 		    (status == DS_VERS_INCREASING_MAJOR_ERR) ?
2495 		    "increasing major versions" :
2496 		    "increasing minor versions");
2497 		return (EINVAL);
2498 	}
2499 
2500 	/* data and register callbacks are required */
2501 	if ((ops->ds_data_cb == NULL) || (ops->ds_reg_cb == NULL)) {
2502 		cmn_err(CE_NOTE, "%s: invalid ops specifier for %s service"
2503 		    DS_EOL, __func__, cap->svc_id);
2504 		return (EINVAL);
2505 	}
2506 
2507 	flags &= DSSF_USERFLAGS;
2508 	is_client = flags & DSSF_ISCLIENT;
2509 
2510 	DS_DBG_USR(CE_NOTE, "%s: svc_id='%s', data_cb=0x%lx, cb_arg=0x%lx"
2511 	    DS_EOL, __func__, cap->svc_id, PTR_TO_LONG(ops->ds_data_cb),
2512 	    PTR_TO_LONG(ops->cb_arg));
2513 
2514 	mutex_enter(&ds_svcs.lock);
2515 
2516 	/* check if the service is already registered */
2517 	if (i_ds_hdl_lookup(cap->svc_id, is_client, NULL, 1) == 1) {
2518 		/* already registered */
2519 		cmn_err(CE_NOTE, "Service '%s'/%s already registered" DS_EOL,
2520 		    cap->svc_id,
2521 		    (flags & DSSF_ISCLIENT) ? "client" : "service");
2522 		mutex_exit(&ds_svcs.lock);
2523 		return (EALREADY);
2524 	}
2525 
2526 	svc = ds_alloc_svc();
2527 	if (is_client) {
2528 		DS_HDL_SET_ISCLIENT(svc->hdl);
2529 	}
2530 
2531 	svc->state = DS_SVC_FREE;
2532 	svc->svc_hdl = DS_BADHDL1;
2533 
2534 	svc->flags = flags;
2535 	svc->drvi = instance;
2536 	svc->drv_psp = NULL;
2537 
2538 	/*
2539 	 * Check for loopback.  "pri" is a legacy service that assumes it
2540 	 * will never use loopback mode.
2541 	 */
2542 	if (strcmp(cap->svc_id, "pri") == 0) {
2543 		is_loopback = 0;
2544 	} else if (i_ds_hdl_lookup(cap->svc_id, is_client == 0, &lb_hdl, 1)
2545 	    == 1) {
2546 		if ((rv = ds_loopback_set_svc(svc, cap, &lb_hdl)) != 0) {
2547 			DS_DBG_USR(CE_NOTE, "%s: ds_loopback_set_svc '%s' err "
2548 			    " (%d)" DS_EOL, __func__, cap->svc_id, rv);
2549 			mutex_exit(&ds_svcs.lock);
2550 			return (rv);
2551 		}
2552 		is_loopback = 1;
2553 	} else
2554 		is_loopback = 0;
2555 
2556 	/* copy over all the client information */
2557 	(void) memcpy(&svc->cap, cap, sizeof (ds_capability_t));
2558 
2559 	/* make a copy of the service name */
2560 	svc->cap.svc_id = ds_strdup(cap->svc_id);
2561 
2562 	/* make a copy of the version array */
2563 	svc->cap.vers = DS_MALLOC(cap->nvers * sizeof (ds_ver_t));
2564 	(void) memcpy(svc->cap.vers, cap->vers, cap->nvers * sizeof (ds_ver_t));
2565 
2566 	/* copy the client ops vector */
2567 	(void) memcpy(&svc->ops, ops, sizeof (ds_clnt_ops_t));
2568 
2569 	svc->state = DS_SVC_INACTIVE;
2570 	svc->ver_idx = 0;
2571 	DS_PORTSET_DUP(svc->avail, ds_allports);
2572 	DS_PORTSET_SETNULL(svc->tried);
2573 
2574 	ds_svcs.nsvcs++;
2575 
2576 	hdl = svc->hdl;
2577 
2578 	/*
2579 	 * kludge to allow user callback code to get handle and user args.
2580 	 * Make sure the callback arg points to the svc structure.
2581 	 */
2582 	if ((flags & DSSF_ISUSER) != 0) {
2583 		ds_cbarg_set_cookie(svc);
2584 	}
2585 
2586 	if (is_loopback) {
2587 		ds_loopback_register(hdl);
2588 		ds_loopback_register(lb_hdl);
2589 	}
2590 
2591 	/*
2592 	 * If this is a client or a non-loopback service provider, send
2593 	 * out register requests.
2594 	 */
2595 	if (!is_loopback || (flags & DSSF_ISCLIENT) != 0)
2596 		(void) ds_svc_register(svc, NULL);
2597 
2598 	if (hdlp) {
2599 		*hdlp = hdl;
2600 	}
2601 
2602 	mutex_exit(&ds_svcs.lock);
2603 
2604 	DS_DBG_USR(CE_NOTE, "%s: service '%s' assigned handle 0x%09lx" DS_EOL,
2605 	    __func__, svc->cap.svc_id, hdl);
2606 
2607 	return (0);
2608 }
2609 
2610 /*
2611  * ds_cap_init interface for previous revision.
2612  */
2613 int
2614 ds_cap_init(ds_capability_t *cap, ds_clnt_ops_t *ops)
2615 {
2616 	return (ds_ucap_init(cap, ops, 0, DS_INVALID_INSTANCE, NULL));
2617 }
2618 
2619 /*
2620  * Interface for ds_unreg_hdl in lds driver.
2621  */
2622 int
2623 ds_unreg_hdl(ds_svc_hdl_t hdl)
2624 {
2625 	ds_svc_t	*svc;
2626 	int		is_loopback;
2627 	ds_svc_hdl_t	lb_hdl;
2628 
2629 	DS_DBG_USR(CE_NOTE, "%s: hdl=0x%09lx" DS_EOL, __func__, hdl);
2630 
2631 	mutex_enter(&ds_svcs.lock);
2632 	if ((svc = ds_get_svc(hdl)) == NULL) {
2633 		mutex_exit(&ds_svcs.lock);
2634 		DS_DBG_USR(CE_NOTE, "%s: unknown hdl: 0x%llx" DS_EOL, __func__,
2635 		    (u_longlong_t)hdl);
2636 		return (ENXIO);
2637 	}
2638 
2639 	DS_DBG_USR(CE_NOTE, "%s: svcid='%s', hdl=0x%llx" DS_EOL, __func__,
2640 	    svc->cap.svc_id, (u_longlong_t)svc->hdl);
2641 
2642 	svc->state = DS_SVC_UNREG_PENDING;
2643 
2644 	is_loopback = ((svc->flags & DSSF_LOOPBACK) != 0);
2645 	lb_hdl = svc->svc_hdl;
2646 
2647 	if (svc->port) {
2648 		(void) ds_send_unreg_req(svc);
2649 	}
2650 
2651 	(void) ds_svc_unregister(svc, svc->port);
2652 
2653 	ds_delete_svc_entry(svc);
2654 
2655 	if (is_loopback) {
2656 		ds_loopback_unregister(lb_hdl);
2657 	}
2658 
2659 	mutex_exit(&ds_svcs.lock);
2660 
2661 	return (0);
2662 }
2663 
2664 int
2665 ds_cap_fini(ds_capability_t *cap)
2666 {
2667 	ds_svc_hdl_t	hdl;
2668 	int rv;
2669 	uint_t nhdls = 0;
2670 
2671 	DS_DBG(CE_NOTE, "%s: '%s'" DS_EOL, __func__, cap->svc_id);
2672 	if ((rv = ds_hdl_lookup(cap->svc_id, 0, &hdl, 1, &nhdls)) != 0) {
2673 		DS_DBG(CE_NOTE, "%s: ds_hdl_lookup '%s' err (%d)" DS_EOL,
2674 		    __func__, cap->svc_id, rv);
2675 		return (rv);
2676 	}
2677 
2678 	if (nhdls == 0) {
2679 		DS_DBG(CE_NOTE, "%s: no such service '%s'" DS_EOL,
2680 		    __func__, cap->svc_id);
2681 		return (ENXIO);
2682 	}
2683 
2684 	if ((rv = ds_is_my_hdl(hdl, DS_INVALID_INSTANCE)) != 0) {
2685 		DS_DBG(CE_NOTE, "%s: ds_is_my_handle err (%d)" DS_EOL, __func__,
2686 		    rv);
2687 		return (rv);
2688 	}
2689 
2690 	if ((rv = ds_unreg_hdl(hdl)) != 0) {
2691 		DS_DBG(CE_NOTE, "%s: ds_unreg_hdl err (%d)" DS_EOL, __func__,
2692 		    rv);
2693 		return (rv);
2694 	}
2695 
2696 	return (0);
2697 }
2698 
2699 int
2700 ds_cap_send(ds_svc_hdl_t hdl, void *buf, size_t len)
2701 {
2702 	int		rv;
2703 	ds_hdr_t	*hdr;
2704 	caddr_t		msg;
2705 	size_t		msglen;
2706 	size_t		hdrlen;
2707 	caddr_t		payload;
2708 	ds_svc_t	*svc;
2709 	ds_port_t	*port;
2710 	ds_data_handle_t *data;
2711 	ds_svc_hdl_t	svc_hdl;
2712 	int		is_client = 0;
2713 
2714 	DS_DBG(CE_NOTE, "%s: hdl: 0x%llx, buf: %lx, len: %ld" DS_EOL, __func__,
2715 	    (u_longlong_t)hdl, (ulong_t)buf, len);
2716 
2717 	mutex_enter(&ds_svcs.lock);
2718 
2719 	if ((svc = ds_get_svc(hdl)) == NULL) {
2720 		cmn_err(CE_WARN, "%s: invalid handle 0x%llx" DS_EOL, __func__,
2721 		    (u_longlong_t)hdl);
2722 		mutex_exit(&ds_svcs.lock);
2723 		return (ENXIO);
2724 	}
2725 
2726 	if (svc->state != DS_SVC_ACTIVE) {
2727 		/* channel is up, but svc is not registered */
2728 		DS_DBG(CE_NOTE, "%s: invalid service state 0x%x" DS_EOL,
2729 		    __func__, svc->state);
2730 		mutex_exit(&ds_svcs.lock);
2731 		return (ENOTCONN);
2732 	}
2733 
2734 	if (svc->flags & DSSF_LOOPBACK) {
2735 		hdl = svc->svc_hdl;
2736 		mutex_exit(&ds_svcs.lock);
2737 		ds_loopback_send(hdl, buf, len);
2738 		return (0);
2739 	}
2740 
2741 	if ((port = svc->port) == NULL) {
2742 		DS_DBG(CE_NOTE, "%s: service '%s' not associated with a port"
2743 		    DS_EOL, __func__, svc->cap.svc_id);
2744 		mutex_exit(&ds_svcs.lock);
2745 		return (ECONNRESET);
2746 	}
2747 
2748 	if (svc->flags & DSSF_ISCLIENT) {
2749 		is_client = 1;
2750 		svc_hdl = svc->svc_hdl;
2751 	}
2752 
2753 	mutex_exit(&ds_svcs.lock);
2754 
2755 	/* check that the LDC channel is ready */
2756 	if (port->ldc.state != LDC_UP) {
2757 		DS_DBG(CE_NOTE, "%s: LDC channel is not up" DS_EOL, __func__);
2758 		return (ECONNRESET);
2759 	}
2760 
2761 	hdrlen = DS_HDR_SZ + sizeof (ds_data_handle_t);
2762 
2763 	msg = DS_MALLOC(len + hdrlen);
2764 	hdr = (ds_hdr_t *)msg;
2765 	payload = msg + hdrlen;
2766 	msglen = len + hdrlen;
2767 
2768 	hdr->payload_len = len + sizeof (ds_data_handle_t);
2769 	hdr->msg_type = DS_DATA;
2770 
2771 	data = (ds_data_handle_t *)(msg + DS_HDR_SZ);
2772 	if (is_client) {
2773 		data->svc_handle = svc_hdl;
2774 	} else {
2775 		data->svc_handle = hdl;
2776 	}
2777 
2778 	if ((buf != NULL) && (len != 0)) {
2779 		(void) memcpy(payload, buf, len);
2780 	}
2781 
2782 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: data>: hdl=0x%llx, len=%ld, "
2783 	    " payload_len=%d" DS_EOL, PORTID(port), (u_longlong_t)svc->hdl,
2784 	    msglen, hdr->payload_len);
2785 	DS_DUMP_MSG(DS_DBG_FLAG_PRCL, msg, msglen);
2786 
2787 	if ((rv = ds_send_msg(port, msg, msglen)) != 0) {
2788 		rv = (rv == EIO) ? ECONNRESET : rv;
2789 	}
2790 	DS_FREE(msg, msglen);
2791 
2792 	return (rv);
2793 }
2794 
2795 void
2796 ds_port_common_init(ds_port_t *port)
2797 {
2798 	int rv;
2799 
2800 	if ((port->flags & DS_PORT_MUTEX_INITED) == 0) {
2801 		mutex_init(&port->lock, NULL, MUTEX_DRIVER, NULL);
2802 		mutex_init(&port->tx_lock, NULL, MUTEX_DRIVER, NULL);
2803 		mutex_init(&port->rcv_lock, NULL, MUTEX_DRIVER, NULL);
2804 		port->flags |= DS_PORT_MUTEX_INITED;
2805 	}
2806 
2807 	port->state = DS_PORT_INIT;
2808 	DS_PORTSET_ADD(ds_allports, port->id);
2809 
2810 	ds_sys_port_init(port);
2811 
2812 	mutex_enter(&port->lock);
2813 	rv = ds_ldc_init(port);
2814 	mutex_exit(&port->lock);
2815 
2816 	/*
2817 	 * If LDC successfully init'ed, try to kick off protocol for this port.
2818 	 */
2819 	if (rv == 0) {
2820 		ds_handle_up_event(port);
2821 	}
2822 }
2823 
2824 void
2825 ds_port_common_fini(ds_port_t *port)
2826 {
2827 	ASSERT(MUTEX_HELD(&port->lock));
2828 
2829 	port->state = DS_PORT_FREE;
2830 
2831 	DS_PORTSET_DEL(ds_allports, port->id);
2832 
2833 	ds_sys_port_fini(port);
2834 }
2835 
2836 /*
2837  * Initialize table of registered service classes
2838  */
2839 void
2840 ds_init_svcs_tbl(uint_t nentries)
2841 {
2842 	int	tblsz;
2843 
2844 	ds_svcs.maxsvcs = nentries;
2845 
2846 	tblsz = ds_svcs.maxsvcs * sizeof (ds_svc_t *);
2847 	ds_svcs.tbl = (ds_svc_t **)DS_MALLOC(tblsz);
2848 
2849 	ds_svcs.nsvcs = 0;
2850 }
2851 
2852 /*
2853  * Find the max and min version supported.
2854  * Hacked from zeus workspace, support.c
2855  */
2856 static void
2857 min_max_versions(int num_versions, ds_ver_t *sup_versionsp,
2858     uint16_t *min_major, uint16_t *max_major)
2859 {
2860 	int i;
2861 
2862 	*min_major = sup_versionsp[0].major;
2863 	*max_major = *min_major;
2864 
2865 	for (i = 1; i < num_versions; i++) {
2866 		if (sup_versionsp[i].major < *min_major)
2867 			*min_major = sup_versionsp[i].major;
2868 
2869 		if (sup_versionsp[i].major > *max_major)
2870 			*max_major = sup_versionsp[i].major;
2871 	}
2872 }
2873 
2874 /*
2875  * Check whether the major and minor numbers requested by the peer can be
2876  * satisfied. If the requested major is supported, true is returned, and the
2877  * agreed minor is returned in new_minor. If the requested major is not
2878  * supported, the routine returns false, and the closest major is returned in
2879  * *new_major, upon which the peer should re-negotiate. The closest major is
2880  * the just lower that the requested major number.
2881  *
2882  * Hacked from zeus workspace, support.c
2883  */
2884 boolean_t
2885 negotiate_version(int num_versions, ds_ver_t *sup_versionsp,
2886     uint16_t req_major, uint16_t *new_majorp, uint16_t *new_minorp)
2887 {
2888 	int i;
2889 	uint16_t major, lower_major;
2890 	uint16_t min_major = 0, max_major;
2891 	boolean_t found_match = B_FALSE;
2892 
2893 	min_max_versions(num_versions, sup_versionsp, &min_major, &max_major);
2894 
2895 	DS_DBG(CE_NOTE, "negotiate_version: req_major = %u, min = %u, max = %u"
2896 	    DS_EOL, req_major, min_major, max_major);
2897 
2898 	/*
2899 	 * If the minimum version supported is greater than
2900 	 * the version requested, return the lowest version
2901 	 * supported
2902 	 */
2903 	if (min_major > req_major) {
2904 		*new_majorp = min_major;
2905 		return (B_FALSE);
2906 	}
2907 
2908 	/*
2909 	 * If the largest version supported is lower than
2910 	 * the version requested, return the largest version
2911 	 * supported
2912 	 */
2913 	if (max_major < req_major) {
2914 		*new_majorp = max_major;
2915 		return (B_FALSE);
2916 	}
2917 
2918 	/*
2919 	 * Now we know that the requested version lies between the
2920 	 * min and max versions supported. Check if the requested
2921 	 * major can be found in supported versions.
2922 	 */
2923 	lower_major = min_major;
2924 	for (i = 0; i < num_versions; i++) {
2925 		major = sup_versionsp[i].major;
2926 		if (major == req_major) {
2927 			found_match = B_TRUE;
2928 			*new_majorp = req_major;
2929 			*new_minorp = sup_versionsp[i].minor;
2930 			break;
2931 		} else {
2932 			if ((major < req_major) && (major > lower_major))
2933 				lower_major = major;
2934 		}
2935 	}
2936 
2937 	/*
2938 	 * If no match is found, return the closest available number
2939 	 */
2940 	if (!found_match)
2941 		*new_majorp = lower_major;
2942 
2943 	return (found_match);
2944 }
2945 
2946 /*
2947  * Specific errno's that are used by ds.c and ldc.c
2948  */
2949 static struct {
2950 	int ds_errno;
2951 	char *estr;
2952 } ds_errno_to_str_tab[] = {
2953 	{ EIO,		"I/O error" },
2954 	{ ENXIO,	"No such device or address" },
2955 	{ EAGAIN,	"Resource temporarily unavailable" },
2956 	{ ENOMEM,	"Not enough space" },
2957 	{ EACCES,	"Permission denied" },
2958 	{ EFAULT,	"Bad address" },
2959 	{ EBUSY,	"Device busy" },
2960 	{ EINVAL,	"Invalid argument" },
2961 	{ ENOSPC,	"No space left on device" },
2962 	{ ENOMSG,	"No message of desired type" },
2963 #ifdef	ECHRNG
2964 	{ ECHRNG,	"Channel number out of range" },
2965 #endif
2966 	{ ENOTSUP,	"Operation not supported" },
2967 	{ EMSGSIZE,	"Message too long" },
2968 	{ EADDRINUSE,	"Address already in use" },
2969 	{ ECONNRESET,	"Connection reset by peer" },
2970 	{ ENOBUFS,	"No buffer space available" },
2971 	{ ENOTCONN,	"Socket is not connected" },
2972 	{ ECONNREFUSED,	"Connection refused" },
2973 	{ EALREADY,	"Operation already in progress" },
2974 	{ 0,		NULL },
2975 };
2976 
2977 char *
2978 ds_errno_to_str(int ds_errno, char *ebuf)
2979 {
2980 	int i, en;
2981 
2982 	for (i = 0; (en = ds_errno_to_str_tab[i].ds_errno) != 0; i++) {
2983 		if (en == ds_errno) {
2984 			(void) strcpy(ebuf, ds_errno_to_str_tab[i].estr);
2985 			return (ebuf);
2986 		}
2987 	}
2988 
2989 	(void) sprintf(ebuf, "ds_errno (%d)", ds_errno);
2990 	return (ebuf);
2991 }
2992 
2993 static void
2994 ds_loopback_register(ds_svc_hdl_t hdl)
2995 {
2996 	ds_ver_t ds_ver;
2997 	ds_svc_t *svc;
2998 
2999 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
3000 	DS_DBG_LOOP(CE_NOTE, "%s: entered hdl: 0x%llx" DS_EOL, __func__,
3001 	    (u_longlong_t)hdl);
3002 	if ((svc = ds_get_svc(hdl)) == NULL) {
3003 		DS_DBG_LOOP(CE_NOTE, "%s: invalid hdl: 0x%llx" DS_EOL, __func__,
3004 		    (u_longlong_t)hdl);
3005 		return;
3006 	}
3007 
3008 	svc->state = DS_SVC_ACTIVE;
3009 
3010 	if (svc->ops.ds_reg_cb) {
3011 		DS_DBG_LOOP(CE_NOTE, "%s: loopback regcb: hdl: 0x%llx" DS_EOL,
3012 		    __func__, (u_longlong_t)hdl);
3013 		ds_ver.major = svc->ver.major;
3014 		ds_ver.minor = svc->ver.minor;
3015 		(*svc->ops.ds_reg_cb)(svc->ops.cb_arg, &ds_ver, hdl);
3016 	}
3017 }
3018 
3019 static void
3020 ds_loopback_unregister(ds_svc_hdl_t hdl)
3021 {
3022 	ds_svc_t *svc;
3023 
3024 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
3025 	if ((svc = ds_get_svc(hdl)) == NULL) {
3026 		DS_DBG_LOOP(CE_NOTE, "%s: invalid hdl: 0x%llx" DS_EOL, __func__,
3027 		    (u_longlong_t)hdl);
3028 		return;
3029 	}
3030 
3031 	DS_DBG_LOOP(CE_NOTE, "%s: entered hdl: 0x%llx" DS_EOL, __func__,
3032 	    (u_longlong_t)hdl);
3033 
3034 	svc->flags &= ~DSSF_LOOPBACK;
3035 	svc->svc_hdl = DS_BADHDL2;
3036 	svc->state = DS_SVC_INACTIVE;
3037 
3038 	if (svc->ops.ds_unreg_cb) {
3039 		DS_DBG_LOOP(CE_NOTE, "%s: loopback unregcb: hdl: 0x%llx" DS_EOL,
3040 		    __func__, (u_longlong_t)hdl);
3041 		(*svc->ops.ds_unreg_cb)(svc->ops.cb_arg);
3042 	}
3043 }
3044 
3045 static void
3046 ds_loopback_send(ds_svc_hdl_t hdl, void *buf, size_t buflen)
3047 {
3048 	ds_svc_t *svc;
3049 
3050 	mutex_enter(&ds_svcs.lock);
3051 	if ((svc = ds_get_svc(hdl)) == NULL) {
3052 		mutex_exit(&ds_svcs.lock);
3053 		DS_DBG_LOOP(CE_NOTE, "%s: invalid hdl: 0x%llx" DS_EOL, __func__,
3054 		    (u_longlong_t)hdl);
3055 		return;
3056 	}
3057 	mutex_exit(&ds_svcs.lock);
3058 
3059 	DS_DBG_LOOP(CE_NOTE, "%s: entered hdl: 0x%llx" DS_EOL, __func__,
3060 	    (u_longlong_t)hdl);
3061 
3062 	if (svc->ops.ds_data_cb) {
3063 		DS_DBG_LOOP(CE_NOTE, "%s: loopback datacb hdl: 0x%llx" DS_EOL,
3064 		    __func__, (u_longlong_t)hdl);
3065 		(*svc->ops.ds_data_cb)(svc->ops.cb_arg, buf, buflen);
3066 	}
3067 }
3068 
3069 static int
3070 ds_loopback_set_svc(ds_svc_t *svc, ds_capability_t *cap, ds_svc_hdl_t *lb_hdlp)
3071 {
3072 	ds_svc_t *lb_svc;
3073 	ds_svc_hdl_t lb_hdl = *lb_hdlp;
3074 	int i;
3075 	int match = 0;
3076 	uint16_t new_major;
3077 	uint16_t new_minor;
3078 
3079 	if ((lb_svc = ds_get_svc(lb_hdl)) == NULL) {
3080 		DS_DBG_LOOP(CE_NOTE, "%s: loopback: hdl: 0x%llx invalid" DS_EOL,
3081 		    __func__, (u_longlong_t)lb_hdl);
3082 		return (ENXIO);
3083 	}
3084 
3085 	/* negotiate a version between loopback services, if possible */
3086 	for (i = 0; i < lb_svc->cap.nvers && match == 0; i++) {
3087 		match = negotiate_version(cap->nvers, cap->vers,
3088 		    lb_svc->cap.vers[i].major, &new_major, &new_minor);
3089 	}
3090 	if (!match) {
3091 		DS_DBG_LOOP(CE_NOTE, "%s: loopback version negotiate failed"
3092 		    DS_EOL, __func__);
3093 		return (ENOTSUP);
3094 	}
3095 	if (lb_svc->state != DS_SVC_INACTIVE) {
3096 		DS_DBG_LOOP(CE_NOTE, "%s: loopback active: hdl: 0x%llx"
3097 		    DS_EOL, __func__, (u_longlong_t)lb_hdl);
3098 		if ((lb_svc->flags & DSSF_ISCLIENT) == 0) {
3099 			DS_DBG_LOOP(CE_NOTE, "%s: loopback busy hdl: 0x%llx"
3100 			    DS_EOL, __func__, (u_longlong_t)lb_hdl);
3101 			return (EBUSY);
3102 		}
3103 		svc->state = DS_SVC_INACTIVE;	/* prevent alloc'ing svc */
3104 		lb_svc = ds_svc_clone(lb_svc);
3105 		DS_DBG_LOOP(CE_NOTE, "%s: loopback clone: ohdl: 0x%llx "
3106 		    "nhdl: 0x%llx" DS_EOL, __func__, (u_longlong_t)lb_hdl,
3107 		    (u_longlong_t)lb_svc->hdl);
3108 		*lb_hdlp = lb_svc->hdl;
3109 	}
3110 
3111 	svc->flags |= DSSF_LOOPBACK;
3112 	svc->svc_hdl = lb_svc->hdl;
3113 	svc->port = NULL;
3114 	svc->ver.major = new_major;
3115 	svc->ver.minor = new_minor;
3116 
3117 	lb_svc->flags |= DSSF_LOOPBACK;
3118 	lb_svc->svc_hdl = svc->hdl;
3119 	lb_svc->port = NULL;
3120 	lb_svc->ver.major = new_major;
3121 	lb_svc->ver.minor = new_minor;
3122 
3123 	DS_DBG_LOOP(CE_NOTE, "%s: setting loopback between: 0x%llx and 0x%llx"
3124 	    DS_EOL, __func__, (u_longlong_t)svc->hdl,
3125 	    (u_longlong_t)lb_svc->hdl);
3126 	return (0);
3127 }
3128 
3129 static ds_svc_t *
3130 ds_find_clnt_svc_by_hdl_port(ds_svc_hdl_t hdl, ds_port_t *port)
3131 {
3132 	int		idx;
3133 	ds_svc_t	*svc;
3134 
3135 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: %s looking up clnt hdl: 0x%llx" DS_EOL,
3136 	    PORTID(port), __func__, (u_longlong_t)hdl);
3137 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
3138 
3139 	/* walk every table entry */
3140 	for (idx = 0; idx < ds_svcs.maxsvcs; idx++) {
3141 		svc = ds_svcs.tbl[idx];
3142 		if (DS_SVC_ISFREE(svc))
3143 			continue;
3144 		if ((svc->flags & DSSF_ISCLIENT) != 0 &&
3145 		    svc->svc_hdl == hdl && svc->port == port) {
3146 			DS_DBG_PRCL(CE_NOTE, "ds@%lx: %s found clnt hdl "
3147 			    "0x%llx: svc%d" DS_EOL, PORTID(port), __func__,
3148 			    (u_longlong_t)hdl, (uint_t)DS_HDL2IDX(svc->hdl));
3149 			return (svc);
3150 		}
3151 	}
3152 	DS_DBG_PRCL(CE_NOTE, "ds@%lx: %s clnt hdl: 0x%llx not found" DS_EOL,
3153 	    PORTID(port), __func__, (u_longlong_t)hdl);
3154 
3155 	return (NULL);
3156 }
3157 
3158 static ds_svc_t *
3159 ds_svc_clone(ds_svc_t *svc)
3160 {
3161 	ds_svc_t *newsvc;
3162 	ds_svc_hdl_t hdl;
3163 
3164 	ASSERT(svc->flags & DSSF_ISCLIENT);
3165 
3166 	newsvc = ds_alloc_svc();
3167 
3168 	/* Can only clone clients for now */
3169 	hdl = newsvc->hdl | DS_HDL_ISCLIENT_BIT;
3170 	DS_DBG_USR(CE_NOTE, "%s: cloning client: old hdl: 0x%llx new hdl: "
3171 	    "0x%llx" DS_EOL, __func__, (u_longlong_t)svc->hdl,
3172 	    (u_longlong_t)hdl);
3173 	(void) memcpy(newsvc, svc, sizeof (ds_svc_t));
3174 	newsvc->hdl = hdl;
3175 	newsvc->flags &= ~DSSF_LOOPBACK;
3176 	newsvc->port = NULL;
3177 	newsvc->svc_hdl = DS_BADHDL2;
3178 	newsvc->cap.svc_id = ds_strdup(svc->cap.svc_id);
3179 	newsvc->cap.vers = DS_MALLOC(svc->cap.nvers * sizeof (ds_ver_t));
3180 	(void) memcpy(newsvc->cap.vers, svc->cap.vers,
3181 	    svc->cap.nvers * sizeof (ds_ver_t));
3182 
3183 	/*
3184 	 * Kludge to allow lds driver user callbacks to get access to current
3185 	 * svc structure.  Arg could be index to svc table or some other piece
3186 	 * of info to get to the svc table entry.
3187 	 */
3188 	if (newsvc->flags & DSSF_ISUSER) {
3189 		newsvc->ops.cb_arg = (ds_cb_arg_t)(newsvc);
3190 	}
3191 	return (newsvc);
3192 }
3193 
3194 /*
3195  * Internal handle lookup function.
3196  */
3197 static int
3198 i_ds_hdl_lookup(char *service, uint_t is_client, ds_svc_hdl_t *hdlp,
3199     uint_t maxhdls)
3200 {
3201 	int idx;
3202 	int nhdls = 0;
3203 	ds_svc_t *svc;
3204 	uint32_t client_flag = is_client ? DSSF_ISCLIENT : 0;
3205 
3206 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
3207 
3208 	for (idx = 0; idx < ds_svcs.maxsvcs && nhdls < maxhdls; idx++) {
3209 		svc = ds_svcs.tbl[idx];
3210 		if (DS_SVC_ISFREE(svc))
3211 			continue;
3212 		if (strcmp(svc->cap.svc_id, service) == 0 &&
3213 		    (svc->flags & DSSF_ISCLIENT) == client_flag) {
3214 			if (hdlp != NULL && nhdls < maxhdls) {
3215 				hdlp[nhdls] = svc->hdl;
3216 				nhdls++;
3217 			} else {
3218 				nhdls++;
3219 			}
3220 		}
3221 	}
3222 	return (nhdls);
3223 }
3224 
3225 /*
3226  * Interface for ds_hdl_lookup in lds driver.
3227  */
3228 int
3229 ds_hdl_lookup(char *service, uint_t is_client, ds_svc_hdl_t *hdlp,
3230     uint_t maxhdls, uint_t *nhdlsp)
3231 {
3232 	mutex_enter(&ds_svcs.lock);
3233 	*nhdlsp = i_ds_hdl_lookup(service, is_client, hdlp, maxhdls);
3234 	mutex_exit(&ds_svcs.lock);
3235 	return (0);
3236 }
3237 
3238 /*
3239  * After an UNREG REQ, check if this is a client service with multiple
3240  * handles.  If it is, then we can eliminate this entry.
3241  */
3242 static void
3243 ds_check_for_dup_services(ds_svc_t *svc)
3244 {
3245 	if ((svc->flags & DSSF_ISCLIENT) != 0 &&
3246 	    svc->state == DS_SVC_INACTIVE &&
3247 	    i_ds_hdl_lookup(svc->cap.svc_id, 1, NULL, 2) == 2) {
3248 		ds_delete_svc_entry(svc);
3249 	}
3250 }
3251 
3252 static void
3253 ds_delete_svc_entry(ds_svc_t *svc)
3254 {
3255 	ds_svc_hdl_t tmp_hdl;
3256 
3257 	ASSERT(MUTEX_HELD(&ds_svcs.lock));
3258 
3259 	/*
3260 	 * Clear out the structure, but do not deallocate the
3261 	 * memory. It can be reused for the next registration.
3262 	 */
3263 	DS_FREE(svc->cap.svc_id, strlen(svc->cap.svc_id) + 1);
3264 	DS_FREE(svc->cap.vers, svc->cap.nvers * sizeof (ds_ver_t));
3265 
3266 	/* save the handle to prevent reuse */
3267 	tmp_hdl = svc->hdl;
3268 	bzero((void *)svc, sizeof (ds_svc_t));
3269 
3270 	/* initialize for next use */
3271 	svc->hdl = tmp_hdl;
3272 	svc->state = DS_SVC_FREE;
3273 
3274 	ds_svcs.nsvcs--;
3275 }
3276