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 * Copyright 2009 Sun Microsystems, Inc. All rights reserved.
23 * Use is subject to license terms.
24 */
25
26 /* Network data replicator Client side */
27
28
29 #include <sys/types.h>
30 #include <sys/debug.h>
31 #include <sys/ksynch.h>
32 #include <sys/cmn_err.h>
33 #include <sys/kmem.h>
34 #include <sys/cred.h>
35 #include <sys/byteorder.h>
36 #include <sys/errno.h>
37
38 #ifdef _SunOS_2_6
39 /*
40 * on 2.6 both dki_lock.h and rpc/types.h define bool_t so we
41 * define enum_t here as it is all we need from rpc/types.h
42 * anyway and make it look like we included it. Yuck.
43 */
44 #define _RPC_TYPES_H
45 typedef int enum_t;
46 #else
47 #ifndef DS_DDICT
48 #include <rpc/types.h>
49 #endif
50 #endif /* _SunOS_2_6 */
51
52 #ifndef DS_DDICT
53 #include <rpc/auth.h>
54 #include <rpc/svc.h>
55 #include <rpc/xdr.h>
56 #endif
57 #include <sys/ddi.h>
58
59 #include <sys/nsc_thread.h>
60 #ifdef DS_DDICT
61 #include <sys/nsctl/contract.h>
62 #endif
63 #include <sys/nsctl/nsctl.h>
64
65 #include <sys/sdt.h> /* dtrace is S10 or later */
66
67 #include "rdc_io.h"
68 #include "rdc_clnt.h"
69 #include "rdc_bitmap.h"
70 #include "rdc_diskq.h"
71
72
73 kmutex_t rdc_clnt_lock;
74
75 #ifdef DEBUG
76 int noflush = 0;
77 #endif
78
79 int rdc_rpc_tmout = RDC_CLNT_TMOUT;
80 static void rdc_clnt_free(struct chtab *, CLIENT *);
81 static void _rdc_remote_flush(rdc_aio_t *);
82
83 void rdc_flush_memq(int index);
84 void rdc_flush_diskq(int index);
85 int rdc_drain_net_queue(int index);
86 void rdc_flusher_thread(int index);
87 int rdc_diskq_enqueue(rdc_k_info_t *krdc, rdc_aio_t *);
88 void rdc_init_diskq_header(rdc_group_t *grp, dqheader *hd);
89 void rdc_dump_iohdrs(disk_queue *dq);
90 rdc_aio_t *rdc_dequeue(rdc_k_info_t *krdc, int *rc);
91 void rdc_clr_iohdr(rdc_k_info_t *krdc, nsc_off_t qpos);
92 void rdc_close_diskq(rdc_group_t *krdc);
93
94 int rdc_writer(int index);
95
96 static struct chtab *rdc_chtable = NULL;
97 static int rdc_clnt_toomany;
98 #ifdef DEBUG
99 static int rdc_ooreply;
100 #endif
101
102 extern void rdc_fail_diskq(rdc_k_info_t *krdc, int wait, int flag);
103 extern int _rdc_rsrv_diskq(rdc_group_t *group);
104 extern void _rdc_rlse_diskq(rdc_group_t *group);
105
106 static enum clnt_stat
107 cl_call_sig(struct __client *rh, rpcproc_t proc,
108 xdrproc_t xargs, caddr_t argsp, xdrproc_t xres,
109 caddr_t resp, struct timeval secs)
110 {
111 enum clnt_stat stat;
112 k_sigset_t smask;
113 sigintr(&smask, 0);
114 rh->cl_nosignal = TRUE;
115 stat = ((*(rh)->cl_ops->cl_call)\
116 (rh, proc, xargs, argsp, xres, resp, secs));
117 rh->cl_nosignal = FALSE;
118 sigunintr(&smask);
119 return (stat);
120 }
121
122 int
123 rdc_net_getsize(int index, uint64_t *sizeptr)
124 {
125 struct timeval t;
126 int err, size;
127 rdc_k_info_t *krdc = &rdc_k_info[index];
128 int remote_index = krdc->remote_index;
129
130 *sizeptr = 0;
131 if (krdc->remote_index < 0)
132 return (EINVAL);
133
134 t.tv_sec = rdc_rpc_tmout;
135 t.tv_usec = 0;
136
137 #ifdef DEBUG
138 if (krdc->intf == NULL)
139 cmn_err(CE_WARN,
140 "!rdc_net_getsize: null intf for index %d", index);
141 #endif
142 if (krdc->rpc_version <= RDC_VERSION5) {
143 err = rdc_clnt_call(krdc->lsrv, RDCPROC_GETSIZE,
144 krdc->rpc_version, xdr_int, (char *)&remote_index,
145 xdr_int, (char *)&size, &t);
146 if (err == 0)
147 *sizeptr = size;
148 } else {
149 err = rdc_clnt_call(krdc->lsrv, RDCPROC_GETSIZE6,
150 krdc->rpc_version, xdr_int, (char *)&remote_index,
151 xdr_u_longlong_t, (char *)sizeptr, &t);
152 }
153 return (err);
154 }
155
156
157 int
158 rdc_net_state(int index, int options)
159 {
160 struct timeval t;
161 int err;
162 int remote_index = -1;
163 rdc_u_info_t *urdc = &rdc_u_info[index];
164 rdc_k_info_t *krdc = &rdc_k_info[index];
165 struct set_state s;
166 struct set_state4 s4;
167 char neta[32], rneta[32];
168 unsigned short *sp;
169
170 t.tv_sec = rdc_rpc_tmout;
171 t.tv_usec = 0;
172
173 if (krdc->rpc_version < RDC_VERSION7) {
174 s4.netaddrlen = urdc->primary.addr.len;
175 s4.rnetaddrlen = urdc->secondary.addr.len;
176 bcopy(urdc->primary.addr.buf, s4.netaddr, s4.netaddrlen);
177 bcopy(urdc->secondary.addr.buf, s4.rnetaddr, s4.rnetaddrlen);
178 (void) strncpy(s4.pfile, urdc->primary.file, RDC_MAXNAMLEN);
179 (void) strncpy(s4.sfile, urdc->secondary.file, RDC_MAXNAMLEN);
180 s4.flag = options;
181
182 err = rdc_clnt_call(krdc->lsrv, RDCPROC_STATE,
183 krdc->rpc_version, xdr_set_state4, (char *)&s4, xdr_int,
184 (char *)&remote_index, &t);
185 } else {
186 s.netaddrlen = urdc->primary.addr.len;
187 s.rnetaddrlen = urdc->secondary.addr.len;
188 s.netaddr.buf = neta;
189 s.rnetaddr.buf = rneta;
190 bcopy(urdc->primary.addr.buf, s.netaddr.buf, s.netaddrlen);
191 bcopy(urdc->secondary.addr.buf, s.rnetaddr.buf, s.rnetaddrlen);
192 s.netaddr.len = urdc->primary.addr.len;
193 s.rnetaddr.len = urdc->secondary.addr.len;
194 s.netaddr.maxlen = urdc->primary.addr.len;
195 s.rnetaddr.maxlen = urdc->secondary.addr.len;
196 sp = (unsigned short *)s.netaddr.buf;
197 *sp = htons(*sp);
198 sp = (unsigned short *)s.rnetaddr.buf;
199 *sp = htons(*sp);
200 s.pfile = urdc->primary.file;
201 s.sfile = urdc->secondary.file;
202 s.flag = options;
203
204 err = rdc_clnt_call(krdc->lsrv, RDCPROC_STATE,
205 krdc->rpc_version, xdr_set_state, (char *)&s, xdr_int,
206 (char *)&remote_index, &t);
207 }
208
209 if (err)
210 return (-1);
211 else
212 return (remote_index);
213 }
214
215
216 /*
217 * rdc_net_getbmap
218 * gets the bitmaps from remote side and or's them with remote bitmap
219 */
220 int
221 rdc_net_getbmap(int index, int size)
222 {
223 struct timeval t;
224 int err;
225 struct bmap b;
226 struct bmap6 b6;
227 rdc_k_info_t *krdc;
228
229 krdc = &rdc_k_info[index];
230
231 if (krdc->remote_index < 0)
232 return (EINVAL);
233
234 t.tv_sec = rdc_rpc_tmout;
235 t.tv_usec = 0;
236 #ifdef DEBUG
237 if (krdc->intf == NULL)
238 cmn_err(CE_WARN,
239 "!rdc_net_getbmap: null intf for index %d", index);
240 #endif
241
242 if (krdc->rpc_version <= RDC_VERSION5) {
243 b.cd = krdc->remote_index;
244 b.dual = index;
245 b.size = size;
246 err = rdc_clnt_call(krdc->lsrv, RDCPROC_BMAP,
247 krdc->rpc_version, xdr_bmap, (char *)&b, xdr_int,
248 (char *)&err, &t);
249
250 } else {
251 b6.cd = krdc->remote_index;
252 b6.dual = index;
253 b6.size = size;
254 err = rdc_clnt_call(krdc->lsrv, RDCPROC_BMAP6,
255 krdc->rpc_version, xdr_bmap6, (char *)&b6, xdr_int,
256 (char *)&err, &t);
257 }
258 return (err);
259 }
260
261 int sndr_proto = 0;
262
263 /*
264 * return state corresponding to rdc_host
265 */
266 int
267 rdc_net_getstate(rdc_k_info_t *krdc, int *serial_mode, int *use_mirror,
268 int *mirror_down, int network)
269 {
270 int err;
271 struct timeval t;
272 int state;
273 rdc_u_info_t *urdc = &rdc_u_info[krdc->index];
274 struct set_state s;
275 #ifdef sparc
276 struct set_state4 s4;
277 #endif
278 char neta[32];
279 char rneta[32];
280 unsigned short *sp;
281 char *setp = (char *)&s;
282 xdrproc_t xdr_proc = xdr_set_state;
283
284 if (krdc->lsrv && (krdc->intf == NULL || krdc->intf->if_down) &&
285 network) /* fail fast */
286 return (-1);
287
288 s.netaddrlen = urdc->primary.addr.len;
289 s.rnetaddrlen = urdc->secondary.addr.len;
290 s.pfile = urdc->primary.file;
291 s.sfile = urdc->secondary.file;
292 s.netaddr.buf = neta;
293 s.rnetaddr.buf = rneta;
294 bcopy(urdc->primary.addr.buf, s.netaddr.buf, s.netaddrlen);
295 bcopy(urdc->secondary.addr.buf, s.rnetaddr.buf, s.rnetaddrlen);
296 sp = (unsigned short *) s.netaddr.buf;
297 *sp = htons(*sp);
298 sp = (unsigned short *) s.rnetaddr.buf;
299 *sp = htons(*sp);
300 s.netaddr.len = urdc->primary.addr.len;
301 s.rnetaddr.len = urdc->secondary.addr.len;
302 s.netaddr.maxlen = urdc->primary.addr.maxlen;
303 s.rnetaddr.maxlen = urdc->secondary.addr.maxlen;
304 s.flag = 0;
305
306 t.tv_sec = rdc_rpc_tmout;
307 t.tv_usec = 0;
308
309 if (sndr_proto)
310 krdc->rpc_version = sndr_proto;
311 else
312 krdc->rpc_version = RDC_VERS_MAX;
313
314 again:
315 err = rdc_clnt_call(krdc->lsrv, RDCPROC_GETSTATE4, krdc->rpc_version,
316 xdr_proc, setp, xdr_int, (char *)&state, &t);
317
318 if (err == RPC_PROGVERSMISMATCH && (krdc->rpc_version !=
319 RDC_VERS_MIN)) {
320 if (krdc->rpc_version-- == RDC_VERSION7) {
321 /* set_state struct changed with v7 of protocol */
322 #ifdef sparc
323 s4.netaddrlen = urdc->primary.addr.len;
324 s4.rnetaddrlen = urdc->secondary.addr.len;
325 bcopy(urdc->primary.addr.buf, s4.netaddr,
326 s4.netaddrlen);
327 bcopy(urdc->secondary.addr.buf, s4.rnetaddr,
328 s4.rnetaddrlen);
329 (void) strncpy(s4.pfile, urdc->primary.file,
330 RDC_MAXNAMLEN);
331 (void) strncpy(s4.sfile, urdc->secondary.file,
332 RDC_MAXNAMLEN);
333 s4.flag = 0;
334 xdr_proc = xdr_set_state4;
335 setp = (char *)&s4;
336 #else
337 /* x64 can not use protocols < 7 */
338 return (-1);
339 #endif
340 }
341 goto again;
342 }
343 #ifdef DEBUG
344 cmn_err(CE_NOTE, "!sndr get_state: Protocol ver %d", krdc->rpc_version);
345 #endif
346
347 if (err) {
348 return (-1);
349 }
350
351 if (state == -1)
352 return (-1);
353
354 if (serial_mode)
355 *serial_mode = (state >> 2) & 1;
356 if (use_mirror)
357 *use_mirror = (state >> 1) & 1;
358 if (mirror_down)
359 *mirror_down = state & 1;
360
361 return (0);
362 }
363
364
365 static struct xdr_discrim rdres_discrim[2] = {
366 { (int)RDC_OK, xdr_readok },
367 { __dontcare__, NULL_xdrproc_t }
368 };
369
370
371 /*
372 * Reply from remote read (client side)
373 */
374 static bool_t
375 xdr_rdresult(XDR *xdrs, readres *rr)
376 {
377
378 return (xdr_union(xdrs, (enum_t *)&(rr->rr_status),
379 (caddr_t)&(rr->rr_ok), rdres_discrim, xdr_void));
380 }
381
382 static int
383 rdc_rrstatus_decode(int status)
384 {
385 int ret = 0;
386
387 if (status != RDC_OK) {
388 switch (status) {
389 case RDCERR_NOENT:
390 ret = ENOENT;
391 break;
392 case RDCERR_NOMEM:
393 ret = ENOMEM;
394 break;
395 default:
396 ret = EIO;
397 break;
398 }
399 }
400
401 return (ret);
402 }
403
404
405 int
406 rdc_net_read(int local_index, int remote_index, nsc_buf_t *handle,
407 nsc_off_t fba_pos, nsc_size_t fba_len)
408 {
409 struct rdcrdresult rr;
410 rdc_k_info_t *krdc;
411 rdc_u_info_t *urdc;
412 struct rread list;
413 struct rread6 list6;
414 struct timeval t;
415 uchar_t *sv_addr;
416 nsc_vec_t *vec;
417 int rpc_flag;
418 nsc_size_t sv_len;
419 int err;
420 int ret;
421 nsc_size_t len;
422 nsc_size_t maxfbas;
423 int transflag;
424
425 if (handle == NULL)
426 return (EINVAL);
427
428 if (!RDC_HANDLE_LIMITS(handle, fba_pos, fba_len)) {
429 #ifdef DEBUG
430 cmn_err(CE_NOTE, "!rdc_net_read: handle bounds");
431 #endif
432 return (EINVAL);
433 }
434
435 krdc = &rdc_k_info[local_index];
436 urdc = &rdc_u_info[local_index];
437
438 maxfbas = MAX_RDC_FBAS;
439
440 if (krdc->remote_fd && !(rdc_get_vflags(urdc) & RDC_FCAL_FAILED)) {
441 nsc_buf_t *remote_h = NULL;
442 int reserved = 0;
443
444 ret = nsc_reserve(krdc->remote_fd, NSC_MULTI);
445 if (RDC_SUCCESS(ret)) {
446 reserved = 1;
447 ret = nsc_alloc_buf(krdc->remote_fd, fba_pos, fba_len,
448 NSC_RDBUF, &remote_h);
449 }
450 if (RDC_SUCCESS(ret)) {
451 ret = nsc_copy(remote_h, handle, fba_pos, fba_pos,
452 fba_len);
453 if (RDC_SUCCESS(ret)) {
454 (void) nsc_free_buf(remote_h);
455 nsc_release(krdc->remote_fd);
456 return (0);
457 }
458 }
459 rdc_group_enter(krdc);
460 rdc_set_flags(urdc, RDC_FCAL_FAILED);
461 rdc_group_exit(krdc);
462 if (remote_h)
463 (void) nsc_free_buf(remote_h);
464 if (reserved)
465 nsc_release(krdc->remote_fd);
466 }
467
468 t.tv_sec = rdc_rpc_tmout;
469 t.tv_usec = 0;
470
471 if (rdc_get_vflags(urdc) & RDC_VOL_FAILED)
472 rpc_flag = RDC_RREAD_FAIL;
473 else
474 rpc_flag = 0;
475
476 #ifdef DEBUG
477 if (krdc->intf == NULL)
478 cmn_err(CE_WARN,
479 "!rdc_net_read: null intf for index %d", local_index);
480 #endif
481 /*
482 * switch on proto version.
483 */
484 len = fba_len; /* length (FBAs) still to xfer */
485 rr.rr_bufsize = 0; /* rpc data buffer length (bytes) */
486 rr.rr_data = NULL; /* rpc data buffer */
487 transflag = rpc_flag | RDC_RREAD_START; /* setup rpc */
488 if (krdc->rpc_version <= RDC_VERSION5) {
489 ASSERT(fba_pos <= INT32_MAX);
490 list.pos = (int)fba_pos; /* fba position of start of chunk */
491 list.cd = remote_index; /* remote end cd */
492 /* send setup rpc */
493 list.flag = transflag;
494 ASSERT(len <= INT32_MAX);
495 list.len = (int)len; /* total fba length */
496 err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ5,
497 krdc->rpc_version, xdr_rread, (char *)&list, xdr_int,
498 (char *)&ret, &t);
499
500 } else {
501 list6.pos = fba_pos; /* fba position of start of chunk */
502 list6.cd = remote_index; /* remote end cd */
503 /* send setup rpc */
504 list6.flag = transflag; /* setup rpc */
505 ASSERT(len <= INT32_MAX);
506 list6.len = (int)len; /* total fba length */
507 err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ6,
508 krdc->rpc_version, xdr_rread6, (char *)&list6, xdr_int,
509 (char *)&ret, &t);
510 }
511
512 if (err) {
513 #ifdef DEBUG
514 cmn_err(CE_NOTE, "!rdc_net_read: setup err %d", err);
515 #endif
516 if (err == RPC_INTR)
517 ret = EINTR;
518 else
519 ret = ENOLINK;
520
521 goto remote_rerror;
522 }
523
524 if (ret == 0) { /* No valid index from r_net_read */
525 #ifdef DEBUG
526 cmn_err(CE_NOTE,
527 "!rdc_net_read: no valid index from r_net_read");
528 #endif
529 return (ENOBUFS);
530 }
531 transflag = rpc_flag | RDC_RREAD_DATA;
532 if (krdc->rpc_version <= RDC_VERSION5) {
533 list.idx = ret; /* save idx to return to server */
534 list.flag = transflag;
535 /* move onto to data xfer rpcs */
536 } else {
537 list6.idx = ret; /* save idx to return to server */
538 list6.flag = transflag;
539 }
540
541 /* find starting position in handle */
542
543 vec = handle->sb_vec;
544
545 fba_pos -= handle->sb_pos;
546
547 for (; fba_pos >= FBA_NUM(vec->sv_len); vec++)
548 fba_pos -= FBA_NUM(vec->sv_len);
549
550 sv_addr = vec->sv_addr + FBA_SIZE(fba_pos); /* data in vector */
551 sv_len = vec->sv_len - FBA_SIZE(fba_pos); /* bytes in vector */
552
553 while (len) {
554 nsc_size_t translen;
555 if (len > maxfbas) {
556 translen = maxfbas;
557 } else {
558 translen = len;
559 }
560
561 if (FBA_SIZE(translen) > sv_len) {
562 translen = FBA_NUM(sv_len);
563 }
564
565 len -= translen;
566 if (len == 0) {
567 /* last data xfer rpc - tell server to cleanup */
568 transflag |= RDC_RREAD_END;
569 }
570
571 if (!rr.rr_data || (nsc_size_t)rr.rr_bufsize !=
572 FBA_SIZE(translen)) {
573 if (rr.rr_data)
574 kmem_free(rr.rr_data, rr.rr_bufsize);
575
576 ASSERT(FBA_SIZE(translen) <= INT32_MAX);
577 rr.rr_bufsize = FBA_SIZE(translen);
578 rr.rr_data = kmem_alloc(rr.rr_bufsize, KM_NOSLEEP);
579 }
580
581 if (!rr.rr_data) {
582 /* error */
583 #ifdef DEBUG
584 cmn_err(CE_NOTE, "!rdc_net_read: kmem_alloc failed");
585 #endif
586 return (ENOMEM);
587 }
588
589 /* get data from remote end */
590
591 #ifdef DEBUG
592 if (krdc->intf == NULL)
593 cmn_err(CE_WARN,
594 "!rdc_net_read: null intf for index %d",
595 local_index);
596 #endif
597 if (krdc->io_kstats) {
598 mutex_enter(krdc->io_kstats->ks_lock);
599 kstat_runq_enter(KSTAT_IO_PTR(krdc->io_kstats));
600 mutex_exit(krdc->io_kstats->ks_lock);
601 }
602 /*CONSTCOND*/
603 ASSERT(RDC_MAXDATA <= INT32_MAX);
604 ASSERT(translen <= RDC_MAXDATA);
605 if (krdc->rpc_version <= RDC_VERSION5) {
606 list.len = (int)translen;
607 list.flag = transflag;
608 err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ5,
609 krdc->rpc_version, xdr_rread, (char *)&list,
610 xdr_rdresult, (char *)&rr, &t);
611 } else {
612 list6.len = (int)translen;
613 list6.flag = transflag;
614 err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ6,
615 krdc->rpc_version, xdr_rread6, (char *)&list6,
616 xdr_rdresult, (char *)&rr, &t);
617 }
618
619 if (krdc->io_kstats) {
620 mutex_enter(krdc->io_kstats->ks_lock);
621 kstat_runq_exit(KSTAT_IO_PTR(krdc->io_kstats));
622 mutex_exit(krdc->io_kstats->ks_lock);
623 }
624
625 if (err) {
626 #ifdef DEBUG
627 cmn_err(CE_NOTE, "!rdc_net_read: rpc err %d", err);
628 #endif
629 if (err == RPC_INTR) {
630 ret = EINTR;
631 } else {
632 ret = ENOLINK;
633 }
634
635 goto remote_rerror;
636 }
637
638 if (rr.rr_status != RDC_OK) {
639 ret = rdc_rrstatus_decode(rr.rr_status);
640 if (!ret)
641 ret = EIO;
642
643 goto remote_rerror;
644 }
645
646 /* copy into handle */
647
648 bcopy(rr.rr_data, sv_addr, (size_t)rr.rr_bufsize);
649
650 /* update counters */
651
652 sv_addr += rr.rr_bufsize;
653 if (krdc->rpc_version <= RDC_VERSION5) {
654 list.pos += translen;
655 } else {
656 list6.pos += translen;
657 }
658 if (krdc->io_kstats) {
659 KSTAT_IO_PTR(krdc->io_kstats)->reads++;
660 KSTAT_IO_PTR(krdc->io_kstats)->nread += rr.rr_bufsize;
661 }
662 ASSERT(sv_len <= INT32_MAX);
663 ASSERT(sv_len >= (nsc_size_t)rr.rr_bufsize);
664 sv_len -= rr.rr_bufsize;
665
666 if (sv_len == 0) {
667 /* goto next vector */
668 vec++;
669 sv_addr = vec->sv_addr;
670 sv_len = vec->sv_len;
671 }
672 }
673
674 if (rr.rr_data)
675 kmem_free(rr.rr_data, rr.rr_bufsize);
676
677 return (0);
678
679 remote_rerror:
680 if (rr.rr_data)
681 kmem_free(rr.rr_data, rr.rr_bufsize);
682
683 return (ret ? ret : ENOLINK);
684 }
685
686 /*
687 * rdc_net_write
688 * Main remote write client side
689 * Handles protocol selection as well as requests for remote allocation
690 * and data transfer
691 * Does local IO for FCAL
692 * caller must clear bitmap on success
693 */
694
695 int
696 rdc_net_write(int local_index, int remote_index, nsc_buf_t *handle,
697 nsc_off_t fba_pos, nsc_size_t fba_len, uint_t aseq, int qpos,
698 netwriteres *netres)
699 {
700 rdc_k_info_t *krdc;
701 rdc_u_info_t *urdc;
702 struct timeval t;
703 nsc_vec_t *vec;
704 int sv_len;
705 nsc_off_t fpos;
706 int err;
707 struct netwriteres netret;
708 struct netwriteres *netresptr;
709 struct net_data5 dlist5;
710 struct net_data6 dlist6;
711 int ret;
712 nsc_size_t maxfbas;
713 int transflag;
714 int translen;
715 int transendoblk;
716 char *transptr;
717 int vflags;
718
719 if (handle == NULL)
720 return (EINVAL);
721
722 /* if not a diskq buffer */
723 if ((qpos == -1) && (!RDC_HANDLE_LIMITS(handle, fba_pos, fba_len))) {
724 #ifdef DEBUG
725 cmn_err(CE_NOTE, "!rdc_net_write: handle bounds");
726 #endif
727 return (EINVAL);
728 }
729
730
731 t.tv_sec = rdc_rpc_tmout;
732 t.tv_usec = 0;
733
734 krdc = &rdc_k_info[local_index];
735 urdc = &rdc_u_info[local_index];
736
737 maxfbas = MAX_RDC_FBAS;
738
739 /* FCAL IO */
740 if (krdc->remote_fd && !(rdc_get_vflags(urdc) & RDC_FCAL_FAILED)) {
741 nsc_buf_t *remote_h = NULL;
742 int reserved = 0;
743
744 ret = nsc_reserve(krdc->remote_fd, NSC_MULTI);
745 if (RDC_SUCCESS(ret)) {
746 reserved = 1;
747 ret = nsc_alloc_buf(krdc->remote_fd, fba_pos, fba_len,
748 NSC_WRBUF, &remote_h);
749 }
750 if (RDC_SUCCESS(ret)) {
751 ret = nsc_copy(handle, remote_h, fba_pos, fba_pos,
752 fba_len);
753 if (RDC_SUCCESS(ret))
754 ret = nsc_write(remote_h, fba_pos, fba_len, 0);
755 if (RDC_SUCCESS(ret)) {
756 (void) nsc_free_buf(remote_h);
757 nsc_release(krdc->remote_fd);
758 return (0);
759 }
760 }
761 rdc_group_enter(krdc);
762 rdc_set_flags(urdc, RDC_FCAL_FAILED);
763 rdc_group_exit(krdc);
764 if (remote_h)
765 (void) nsc_free_buf(remote_h);
766 if (reserved)
767 nsc_release(krdc->remote_fd);
768 }
769
770 /*
771 * At this point we must decide which protocol we are using and
772 * do the right thing
773 */
774 netret.vecdata.vecdata_val = NULL;
775 netret.vecdata.vecdata_len = 0;
776 if (netres) {
777 netresptr = netres;
778 } else {
779 netresptr = &netret;
780 }
781
782 vflags = rdc_get_vflags(urdc);
783
784 if (vflags & (RDC_VOL_FAILED|RDC_BMP_FAILED))
785 transflag = RDC_RWRITE_FAIL;
786 else
787 transflag = 0;
788
789
790 #ifdef DEBUG
791 if (krdc->intf == NULL)
792 cmn_err(CE_WARN, "!rdc_net_write: null intf for index %d",
793 local_index);
794 #endif
795
796 vec = handle->sb_vec;
797
798 /*
799 * find starting position in vector
800 */
801 if ((qpos == -1) || (handle->sb_user == RDC_NULLBUFREAD))
802 fpos = fba_pos - handle->sb_pos;
803 else
804 fpos = (qpos + 1) - handle->sb_pos;
805
806 for (; fpos >= FBA_NUM(vec->sv_len); vec++)
807 fpos -= FBA_NUM(vec->sv_len);
808 sv_len = vec->sv_len - FBA_SIZE(fpos); /* bytes in vector */
809 transptr = (char *)vec->sv_addr + FBA_SIZE(fpos);
810
811 if (krdc->rpc_version <= RDC_VERSION5) {
812 dlist5.local_cd = local_index;
813 dlist5.cd = remote_index;
814 ASSERT(fba_len <= INT32_MAX);
815 ASSERT(fba_pos <= INT32_MAX);
816 dlist5.len = (int)fba_len;
817 dlist5.pos = (int)fba_pos;
818 dlist5.idx = -1; /* Starting index */
819 dlist5.flag = transflag;
820 dlist5.seq = aseq; /* sequence number */
821 dlist5.sfba = (int)fba_pos; /* starting fba for this xfer */
822 } else {
823 dlist6.local_cd = local_index;
824 dlist6.cd = remote_index;
825 ASSERT(fba_len <= INT32_MAX);
826 dlist6.len = (int)fba_len;
827 dlist6.qpos = qpos;
828 dlist6.pos = fba_pos;
829 dlist6.idx = -1; /* Starting index */
830 dlist6.flag = transflag;
831 dlist6.seq = aseq; /* sequence number */
832 dlist6.sfba = fba_pos; /* starting fba for this xfer */
833 }
834
835 transendoblk = 0;
836 while (fba_len) {
837 if (!transptr) {
838 #ifdef DEBUG
839 cmn_err(CE_WARN,
840 "!rdc_net_write: walked off end of handle!");
841 #endif
842 ret = EINVAL;
843 goto remote_error;
844 }
845
846 if (fba_len > maxfbas) {
847 ASSERT(maxfbas <= INT32_MAX);
848 translen = (int)maxfbas;
849 } else {
850 ASSERT(fba_len <= INT32_MAX);
851 translen = (int)fba_len;
852 }
853
854 if (FBA_SIZE(translen) > sv_len) {
855 translen = FBA_NUM(sv_len);
856 }
857
858 fba_len -= translen;
859 if (fba_len == 0) {
860 /* last data xfer - tell server to commit */
861 transendoblk = 1;
862 }
863
864
865 #ifdef DEBUG
866 if (krdc->intf == NULL)
867 cmn_err(CE_WARN,
868 "!rdc_net_write: null intf for index %d",
869 local_index);
870 #endif
871 DTRACE_PROBE(rdc_netwrite_clntcall_start);
872
873 if (krdc->io_kstats) {
874 mutex_enter(krdc->io_kstats->ks_lock);
875 kstat_runq_enter(KSTAT_IO_PTR(krdc->io_kstats));
876 mutex_exit(krdc->io_kstats->ks_lock);
877 }
878 if (krdc->rpc_version <= RDC_VERSION5) {
879 ret = 0;
880 dlist5.nfba = translen;
881 dlist5.endoblk = transendoblk;
882 dlist5.data.data_len = FBA_SIZE(translen);
883 dlist5.data.data_val = transptr;
884 err = rdc_clnt_call(krdc->lsrv, RDCPROC_WRITE5,
885 krdc->rpc_version, xdr_net_data5,
886 (char *)&dlist5, xdr_int,
887 (char *)&ret, &t);
888 if (ret >= 0) {
889 netresptr->result = 0;
890 netresptr->index = ret;
891 } else {
892 netresptr->result = ret;
893 }
894 } else {
895 netresptr->result = 0;
896 dlist6.nfba = translen;
897 dlist6.endoblk = transendoblk;
898 dlist6.data.data_len = FBA_SIZE(translen);
899 dlist6.data.data_val = transptr;
900 err = rdc_clnt_call(krdc->lsrv, RDCPROC_WRITE6,
901 krdc->rpc_version, xdr_net_data6,
902 (char *)&dlist6, xdr_netwriteres,
903 (char *)netresptr, &t);
904 }
905
906 if (krdc->io_kstats) {
907 mutex_enter(krdc->io_kstats->ks_lock);
908 kstat_runq_exit(KSTAT_IO_PTR(krdc->io_kstats));
909 mutex_exit(krdc->io_kstats->ks_lock);
910 }
911
912 DTRACE_PROBE(rdc_netwrite_clntcall_end);
913 ret = netresptr->result;
914 if (err) {
915 if (err == RPC_INTR)
916 ret = EINTR;
917 else if (err && ret != EPROTO)
918 ret = ENOLINK;
919 #ifdef DEBUG
920 cmn_err(CE_NOTE,
921 "!rdc_net_write(5): cd %d err %d ret %d",
922 remote_index, err, ret);
923 #endif
924 goto remote_error;
925 }
926 /* Error from r_net_write5 */
927 if (netresptr->result < 0) {
928 #ifdef DEBUG
929 cmn_err(CE_NOTE,
930 "!rdc_net_write: r_net_write(5) "
931 "returned: %d",
932 -netresptr->result);
933 #endif
934 ret = -netresptr->result;
935 if (netret.vecdata.vecdata_val)
936 kmem_free(netret.vecdata.vecdata_val,
937 netret.vecdata.vecdata_len *
938 sizeof (net_pendvec_t));
939 goto remote_error;
940 } else if (netresptr->index == 0) {
941 #ifdef DEBUG
942 cmn_err(CE_NOTE,
943 "!rdc_net_write: no valid index from "
944 "r_net_write(5)");
945 #endif
946 ret = ENOBUFS;
947 if (netret.vecdata.vecdata_val)
948 kmem_free(netret.vecdata.vecdata_val,
949 netret.vecdata.vecdata_len *
950 sizeof (net_pendvec_t));
951 goto remote_error;
952 }
953 if (krdc->rpc_version <= RDC_VERSION5) {
954 dlist5.idx = netresptr->index;
955 dlist5.sfba += dlist5.nfba;
956 } else {
957 dlist6.idx = netresptr->index;
958 dlist6.sfba += dlist6.nfba;
959 }
960 /* update counters */
961 if (krdc->io_kstats) {
962 KSTAT_IO_PTR(krdc->io_kstats)->writes++;
963 KSTAT_IO_PTR(krdc->io_kstats)->nwritten +=
964 FBA_SIZE(translen);
965 }
966 transptr += FBA_SIZE(translen);
967 sv_len -= FBA_SIZE(translen);
968
969 if (sv_len <= 0) {
970 /* goto next vector */
971 vec++;
972 transptr = (char *)vec->sv_addr;
973 sv_len = vec->sv_len;
974 }
975 }
976 /*
977 * this can't happen.....
978 */
979 if (netret.vecdata.vecdata_val)
980 kmem_free(netret.vecdata.vecdata_val,
981 netret.vecdata.vecdata_len *
982 sizeof (net_pendvec_t));
983
984 return (0);
985
986 remote_error:
987 return (ret ? ret : ENOLINK);
988 }
989
990 void
991 rdc_fixlen(rdc_aio_t *aio)
992 {
993 nsc_vec_t *vecp = aio->qhandle->sb_vec;
994 nsc_size_t len = 0;
995
996 while (vecp->sv_addr) {
997 len += FBA_NUM(vecp->sv_len);
998 vecp++;
999 }
1000 aio->qhandle->sb_len = len;
1001 }
1002
1003 /*
1004 * rdc_dump_alloc_bufs_cd
1005 * Dump allocated buffers (rdc_net_hnd's) for the specified cd.
1006 * this could be the flusher failing, if so, don't do the delay forever
1007 * Returns: 0 (success), EAGAIN (caller needs to try again).
1008 */
1009 int
1010 rdc_dump_alloc_bufs_cd(int index)
1011 {
1012 rdc_k_info_t *krdc;
1013 rdc_aio_t *aio;
1014 net_queue *q;
1015 disk_queue *dq;
1016 kmutex_t *qlock;
1017
1018 krdc = &rdc_k_info[index];
1019
1020
1021 if (!krdc->c_fd) {
1022 /* cannot do anything! */
1023 #ifdef DEBUG
1024 cmn_err(CE_WARN, "!rdc_dump_alloc_bufs_cd(%d): c_fd NULL",
1025 index);
1026 #endif
1027 return (0);
1028 }
1029 rdc_dump_dsets(index);
1030
1031 dq = &krdc->group->diskq;
1032
1033 if (RDC_IS_DISKQ(krdc->group)) {
1034 qlock = QLOCK(dq);
1035 (void) _rdc_rsrv_diskq(krdc->group);
1036 } else {
1037 qlock = &krdc->group->ra_queue.net_qlock;
1038 }
1039
1040 /*
1041 * Now dump the async queue anonymous buffers
1042 * if we are a diskq, the we are using the diskq mutex.
1043 * However, we are flushing from diskq to memory queue
1044 * so we now need to grab the memory lock also
1045 */
1046
1047 q = &krdc->group->ra_queue;
1048
1049 if (RDC_IS_DISKQ(krdc->group)) {
1050 mutex_enter(&q->net_qlock);
1051 if (q->qfill_sleeping == RDC_QFILL_AWAKE) {
1052 int tries = 5;
1053 #ifdef DEBUG_DISKQ
1054 cmn_err(CE_NOTE,
1055 "!dumpalloccd sending diskq->memq flush to sleep");
1056 #endif
1057 q->qfflags |= RDC_QFILLSLEEP;
1058 mutex_exit(&q->net_qlock);
1059
1060 while (q->qfill_sleeping == RDC_QFILL_AWAKE && tries--)
1061 delay(5);
1062 mutex_enter(&q->net_qlock);
1063 }
1064 }
1065
1066 mutex_enter(qlock);
1067
1068 while ((q->net_qhead != NULL)) {
1069 rdc_k_info_t *tmpkrdc;
1070 aio = q->net_qhead;
1071 tmpkrdc = &rdc_k_info[aio->index];
1072
1073 if (RDC_IS_DISKQ(krdc->group)) {
1074 aio->qhandle->sb_user--;
1075 if (aio->qhandle->sb_user == 0) {
1076 rdc_fixlen(aio);
1077 (void) nsc_free_buf(aio->qhandle);
1078 aio->qhandle = NULL;
1079 aio->handle = NULL;
1080 }
1081 } else {
1082 if (aio->handle) {
1083 (void) nsc_free_buf(aio->handle);
1084 aio->handle = NULL;
1085 }
1086 }
1087
1088 if (tmpkrdc->io_kstats && !RDC_IS_DISKQ(krdc->group)) {
1089 mutex_enter(tmpkrdc->io_kstats->ks_lock);
1090 kstat_waitq_exit(KSTAT_IO_PTR(tmpkrdc->io_kstats));
1091 mutex_exit(tmpkrdc->io_kstats->ks_lock);
1092 }
1093 q->net_qhead = q->net_qhead->next;
1094 q->blocks -= aio->len;
1095 q->nitems--;
1096
1097 RDC_CHECK_BIT(tmpkrdc, aio->pos, aio->len);
1098
1099 kmem_free(aio, sizeof (*aio));
1100 }
1101 q->net_qtail = NULL;
1102
1103 if (krdc->group->asyncstall) {
1104 krdc->group->asyncdis = 1;
1105 cv_broadcast(&krdc->group->asyncqcv);
1106 }
1107 if (krdc->group->sleepq) {
1108 rdc_sleepqdiscard(krdc->group);
1109 }
1110
1111 krdc->group->seq = RDC_NEWSEQ;
1112 krdc->group->seqack = RDC_NEWSEQ;
1113 if (RDC_IS_DISKQ(krdc->group)) {
1114 rdc_dump_iohdrs(dq);
1115 SET_QNXTIO(dq, QHEAD(dq));
1116 SET_QCOALBOUNDS(dq, QHEAD(dq));
1117 }
1118 mutex_exit(qlock);
1119
1120 if (RDC_IS_DISKQ(krdc->group)) {
1121 mutex_exit(&q->net_qlock);
1122 _rdc_rlse_diskq(krdc->group);
1123 }
1124
1125 return (0);
1126 }
1127
1128
1129 /*
1130 * rdc_dump_alloc_bufs
1131 * We have an error on the link
1132 * Try to dump all of the allocated bufs so we can cleanly recover
1133 * and not hang
1134 */
1135 void
1136 rdc_dump_alloc_bufs(rdc_if_t *ip)
1137 {
1138 rdc_k_info_t *krdc;
1139 int repeat;
1140 int index;
1141
1142 for (index = 0; index < rdc_max_sets; index++) {
1143 do {
1144 krdc = &rdc_k_info[index];
1145 repeat = 0;
1146 if (krdc->intf == ip) {
1147 if (rdc_dump_alloc_bufs_cd(index) == EAGAIN) {
1148 repeat = 1;
1149 delay(2);
1150 }
1151 }
1152 } while (repeat);
1153 }
1154 }
1155
1156 /*
1157 * returns 1 if the the throttle should throttle, 0 if not.
1158 */
1159 int
1160 _rdc_diskq_isfull(disk_queue *q, long len)
1161 {
1162 /* ---T----H----N--- */
1163 mutex_enter(QLOCK(q));
1164
1165 if (FITSONQ(q, len + 1)) {
1166 mutex_exit(QLOCK(q));
1167 return (0);
1168 }
1169 mutex_exit(QLOCK(q));
1170 return (1);
1171 }
1172
1173 void
1174 _rdc_async_throttle(rdc_k_info_t *this, long len)
1175 {
1176 rdc_k_info_t *krdc;
1177 rdc_u_info_t *urdc;
1178 int print_msg = 1;
1179 int tries = RDC_FUTILE_ATTEMPTS;
1180
1181 /*
1182 * Throttle entries on queue
1183 */
1184
1185 /* Need to take the 1-many case into account, checking all sets */
1186
1187 /* ADD HANDY HUERISTIC HERE TO SLOW DOWN IO */
1188 for (krdc = this; /* CSTYLED */; krdc = krdc->many_next) {
1189 urdc = &rdc_u_info[krdc->index];
1190
1191 /*
1192 * this may be the last set standing in a one to many setup.
1193 * we may also be stuck in unintercept, after marking
1194 * the volume as not enabled, but have not removed it
1195 * from the many list resulting in an endless loop if
1196 * we just continue here. Lets jump over this stuff
1197 * and check to see if we are the only dude here.
1198 */
1199 if (!IS_ENABLED(urdc))
1200 goto thischeck;
1201
1202 if (IS_ASYNC(urdc) && RDC_IS_MEMQ(krdc->group)) {
1203 net_queue *q = &krdc->group->ra_queue;
1204 while ((q->blocks + q->inflbls) > urdc->maxqfbas ||
1205 (q->nitems + q->inflitems) > urdc->maxqitems) {
1206
1207 if (!IS_ENABLED(urdc)) /* disable race */
1208 goto thischeck;
1209
1210 if (!krdc->group->rdc_writer)
1211 (void) rdc_writer(krdc->index);
1212 delay(2);
1213 q->throttle_delay++;
1214 }
1215 }
1216
1217 /* do a much more aggressive delay, get disk flush going */
1218 if (IS_ASYNC(urdc) && RDC_IS_DISKQ(krdc->group)) {
1219 disk_queue *q = &krdc->group->diskq;
1220 while ((!IS_QSTATE(q, RDC_QNOBLOCK)) &&
1221 (_rdc_diskq_isfull(q, len)) &&
1222 (!IS_STATE(urdc, RDC_DISKQ_FAILED))) {
1223 if (print_msg) {
1224 cmn_err(CE_WARN, "!rdc async throttle:"
1225 " disk queue %s full",
1226 &urdc->disk_queue[0]);
1227
1228 print_msg = 0;
1229 }
1230 if (!IS_ENABLED(urdc)) /* disable race */
1231 goto thischeck;
1232
1233 if (!krdc->group->rdc_writer)
1234 (void) rdc_writer(krdc->index);
1235 delay(10);
1236 q->throttle_delay += 10;
1237
1238 if (!(tries--) && IS_STATE(urdc, RDC_QUEUING)) {
1239 cmn_err(CE_WARN, "!SNDR: disk queue "
1240 "%s full & not flushing. giving up",
1241 &urdc->disk_queue[0]);
1242 cmn_err(CE_WARN, "!SNDR: %s:%s entering"
1243 " logging mode",
1244 urdc->secondary.intf,
1245 urdc->secondary.file);
1246 rdc_fail_diskq(krdc, RDC_WAIT,
1247 RDC_DOLOG | RDC_NOFAIL);
1248 mutex_enter(QLOCK(q));
1249 cv_broadcast(&q->qfullcv);
1250 mutex_exit(QLOCK(q));
1251 }
1252
1253 }
1254 if ((IS_QSTATE(q, RDC_QNOBLOCK)) &&
1255 _rdc_diskq_isfull(q, len) &&
1256 !IS_STATE(urdc, RDC_DISKQ_FAILED)) {
1257 if (print_msg) {
1258 cmn_err(CE_WARN, "!disk queue %s full",
1259 &urdc->disk_queue[0]);
1260 print_msg = 0;
1261 }
1262 rdc_fail_diskq(krdc, RDC_WAIT,
1263 RDC_DOLOG | RDC_NOFAIL);
1264 mutex_enter(QLOCK(q));
1265 cv_broadcast(&q->qfullcv);
1266 mutex_exit(QLOCK(q));
1267 }
1268 }
1269
1270 thischeck:
1271 if (krdc->many_next == this)
1272 break;
1273 }
1274 }
1275
1276 int rdc_coalesce = 1;
1277 static int rdc_joins = 0;
1278
1279 int
1280 rdc_aio_coalesce(rdc_aio_t *queued, rdc_aio_t *new)
1281 {
1282 nsc_buf_t *h = NULL;
1283 int rc;
1284 rdc_k_info_t *krdc;
1285 uint_t bitmask;
1286
1287 if (rdc_coalesce == 0)
1288 return (0); /* don't even try */
1289
1290 if ((queued == NULL) ||
1291 (queued->handle == NULL) ||
1292 (new->handle == NULL)) {
1293 return (0); /* existing queue is empty */
1294 }
1295 if (queued->index != new->index || queued->len + new->len >
1296 MAX_RDC_FBAS) {
1297 return (0); /* I/O to big */
1298 }
1299 if ((queued->pos + queued->len == new->pos) ||
1300 (new->pos + new->len == queued->pos)) {
1301 rc = nsc_alloc_abuf(queued->pos, queued->len + new->len, 0,
1302 &h);
1303 if (!RDC_SUCCESS(rc)) {
1304 if (h != NULL)
1305 (void) nsc_free_buf(h);
1306 return (0); /* couldn't do coalesce */
1307 }
1308 rc = nsc_copy(queued->handle, h, queued->pos, queued->pos,
1309 queued->len);
1310 if (!RDC_SUCCESS(rc)) {
1311 (void) nsc_free_buf(h);
1312 return (0); /* couldn't do coalesce */
1313 }
1314 rc = nsc_copy(new->handle, h, new->pos, new->pos,
1315 new->len);
1316 if (!RDC_SUCCESS(rc)) {
1317 (void) nsc_free_buf(h);
1318 return (0); /* couldn't do coalesce */
1319 }
1320
1321 krdc = &rdc_k_info[queued->index];
1322
1323 RDC_SET_BITMASK(queued->pos, queued->len, &bitmask);
1324 RDC_CLR_BITMAP(krdc, queued->pos, queued->len, \
1325 bitmask, RDC_BIT_BUMP);
1326
1327 RDC_SET_BITMASK(new->pos, new->len, &bitmask);
1328 RDC_CLR_BITMAP(krdc, new->pos, new->len, \
1329 bitmask, RDC_BIT_BUMP);
1330
1331 (void) nsc_free_buf(queued->handle);
1332 (void) nsc_free_buf(new->handle);
1333 queued->handle = h;
1334 queued->len += new->len;
1335 bitmask = 0;
1336 /*
1337 * bump the ref count back up
1338 */
1339
1340 RDC_SET_BITMAP(krdc, queued->pos, queued->len, &bitmask);
1341 return (1); /* new I/O succeeds last I/O queued */
1342 }
1343 return (0);
1344 }
1345
1346 int
1347 rdc_memq_enqueue(rdc_k_info_t *krdc, rdc_aio_t *aio)
1348 {
1349 net_queue *q;
1350 rdc_group_t *group;
1351
1352 group = krdc->group;
1353 q = &group->ra_queue;
1354
1355 mutex_enter(&q->net_qlock);
1356
1357 if (rdc_aio_coalesce(q->net_qtail, aio)) {
1358 rdc_joins++;
1359 q->blocks += aio->len;
1360 kmem_free(aio, sizeof (*aio));
1361 goto out;
1362 }
1363 aio->seq = group->seq++;
1364 if (group->seq < aio->seq)
1365 group->seq = RDC_NEWSEQ + 1; /* skip magics */
1366
1367 if (q->net_qhead == NULL) {
1368 /* adding to empty q */
1369 q->net_qhead = q->net_qtail = aio;
1370
1371 #ifdef DEBUG
1372 if (q->blocks != 0 || q->nitems != 0) {
1373 cmn_err(CE_PANIC,
1374 "rdc enqueue: q %p, qhead 0, q blocks %" NSC_SZFMT
1375 ", nitems %" NSC_SZFMT,
1376 (void *) q, q->blocks, q->nitems);
1377 }
1378 #endif
1379
1380 } else {
1381 /* discontiguous, add aio to q tail */
1382 q->net_qtail->next = aio;
1383 q->net_qtail = aio;
1384 }
1385
1386 q->blocks += aio->len;
1387 q->nitems++;
1388
1389 if (krdc->io_kstats) {
1390 mutex_enter(krdc->io_kstats->ks_lock);
1391 kstat_waitq_enter(KSTAT_IO_PTR(krdc->io_kstats));
1392 mutex_exit(krdc->io_kstats->ks_lock);
1393 }
1394 out:
1395 #ifdef DEBUG
1396 /* sum the q and check for sanity */
1397 {
1398 nsc_size_t qblocks = 0;
1399 uint64_t nitems = 0;
1400 rdc_aio_t *a;
1401
1402 for (a = q->net_qhead; a != NULL; a = a->next) {
1403 qblocks += a->len;
1404 nitems++;
1405 }
1406
1407 if (qblocks != q->blocks || nitems != q->nitems) {
1408 cmn_err(CE_PANIC,
1409 "rdc enqueue: q %p, q blocks %" NSC_SZFMT " (%"
1410 NSC_SZFMT "), nitems %" NSC_SZFMT " (%"
1411 NSC_SZFMT ")", (void *) q, q->blocks, qblocks,
1412 q->nitems, nitems);
1413 }
1414 }
1415 #endif
1416
1417 mutex_exit(&q->net_qlock);
1418
1419 if (q->nitems > q->nitems_hwm) {
1420 q->nitems_hwm = q->nitems;
1421 }
1422
1423 if (q->blocks > q->blocks_hwm) {
1424 q->blocks_hwm = q->blocks;
1425 }
1426
1427 if (!krdc->group->rdc_writer)
1428 (void) rdc_writer(krdc->index);
1429
1430 return (0);
1431 }
1432
1433 int
1434 _rdc_enqueue_write(rdc_k_info_t *krdc, nsc_off_t pos, nsc_size_t len, int flag,
1435 nsc_buf_t *h)
1436 {
1437 rdc_aio_t *aio;
1438 rdc_group_t *group;
1439 rdc_u_info_t *urdc = &rdc_u_info[krdc->index];
1440 int rc;
1441
1442 aio = kmem_zalloc(sizeof (*aio), KM_NOSLEEP);
1443 if (!aio) {
1444 return (ENOMEM);
1445 }
1446
1447 group = krdc->group;
1448
1449 aio->pos = pos;
1450 aio->qpos = -1;
1451 aio->len = len;
1452 aio->flag = flag;
1453 aio->index = krdc->index;
1454 aio->handle = h;
1455
1456 if (group->flags & RDC_MEMQUE) {
1457 return (rdc_memq_enqueue(krdc, aio));
1458 } else if ((group->flags & RDC_DISKQUE) &&
1459 !IS_STATE(urdc, RDC_DISKQ_FAILED)) {
1460 rc = rdc_diskq_enqueue(krdc, aio);
1461 kmem_free(aio, sizeof (*aio));
1462 return (rc);
1463 }
1464 return (-1); /* keep lint quiet */
1465 }
1466
1467
1468
1469
1470 /*
1471 * Async Network RDC flusher
1472 */
1473
1474 /*
1475 * don't allow any new writer threads to start if a member of the set
1476 * is disable pending
1477 */
1478 int
1479 is_disable_pending(rdc_k_info_t *krdc)
1480 {
1481 rdc_k_info_t *this = krdc;
1482 int rc = 0;
1483
1484 do {
1485 if (krdc->type_flag & RDC_DISABLEPEND) {
1486 krdc = this;
1487 rc = 1;
1488 break;
1489 }
1490 krdc = krdc->group_next;
1491
1492 } while (krdc != this);
1493
1494 return (rc);
1495 }
1496
1497 /*
1498 * rdc_writer -- spawn new writer if not running already
1499 * called after enqueing the dirty blocks
1500 */
1501 int
1502 rdc_writer(int index)
1503 {
1504 rdc_k_info_t *krdc = &rdc_k_info[index];
1505 nsthread_t *t;
1506 rdc_group_t *group;
1507 kmutex_t *qlock;
1508 int tries;
1509 const int MAX_TRIES = 16;
1510
1511 group = krdc->group;
1512
1513 if (RDC_IS_DISKQ(group))
1514 qlock = &group->diskq.disk_qlock;
1515 else
1516 qlock = &group->ra_queue.net_qlock;
1517
1518 mutex_enter(qlock);
1519
1520 #ifdef DEBUG
1521 if (noflush) {
1522 mutex_exit(qlock);
1523 return (0);
1524 }
1525 #endif
1526
1527 if ((group->rdc_writer) || is_disable_pending(krdc)) {
1528 mutex_exit(qlock);
1529 return (0);
1530 }
1531
1532 if ((group->rdc_thrnum >= 1) && (group->seqack == RDC_NEWSEQ)) {
1533 /*
1534 * We also need to check if we are starting a new
1535 * sequence, and if so don't create a new thread,
1536 * as we must ensure that the start of new sequence
1537 * requests arrives first to re-init the server.
1538 */
1539 mutex_exit(qlock);
1540 return (0);
1541 }
1542 /*
1543 * For version 6,
1544 * see if we can fit in another thread.
1545 */
1546 group->rdc_thrnum++;
1547
1548 if (krdc->intf && (krdc->intf->rpc_version >= RDC_VERSION6)) {
1549 rdc_u_info_t *urdc = &rdc_u_info[index];
1550 if (group->rdc_thrnum >= urdc->asyncthr)
1551 group->rdc_writer = 1;
1552 } else {
1553 group->rdc_writer = 1;
1554 }
1555
1556 mutex_exit(qlock);
1557
1558
1559 /*
1560 * If we got here, we know that we have not exceeded the allowed
1561 * number of async threads for our group. If we run out of threads
1562 * in _rdc_flset, we add a new thread to the set.
1563 */
1564 tries = 0;
1565 do {
1566 /* first try to grab a thread from the free list */
1567 if (t = nst_create(_rdc_flset, rdc_flusher_thread,
1568 (blind_t)(unsigned long)index, 0)) {
1569 break;
1570 }
1571
1572 /* that failed; add a thread to the set and try again */
1573 if (nst_add_thread(_rdc_flset, 1) != 1) {
1574 cmn_err(CE_WARN, "!rdc_writer index %d nst_add_thread "
1575 "error, tries: %d", index, tries);
1576 break;
1577 }
1578 } while (++tries < MAX_TRIES);
1579
1580 if (tries) {
1581 mutex_enter(&group->addthrnumlk);
1582 group->rdc_addthrnum += tries;
1583 mutex_exit(&group->addthrnumlk);
1584 }
1585
1586 if (t) {
1587 return (1);
1588 }
1589
1590 cmn_err(CE_WARN, "!rdc_writer: index %d nst_create error", index);
1591 rdc_many_enter(krdc);
1592 mutex_enter(qlock);
1593 group->rdc_thrnum--;
1594 group->rdc_writer = 0;
1595 if ((group->count == 0) && (group->rdc_thrnum == 0)) {
1596 mutex_exit(qlock);
1597 /*
1598 * Race with remove_from_group while write thread was
1599 * failing to be created.
1600 */
1601 #ifdef DEBUG
1602 cmn_err(CE_WARN, "!rdc_writer: group being destroyed");
1603 #endif
1604 rdc_delgroup(group);
1605 krdc->group = NULL;
1606 rdc_many_exit(krdc);
1607 return (-1);
1608 }
1609 mutex_exit(qlock);
1610 rdc_many_exit(krdc);
1611 return (-1);
1612 }
1613
1614 /*
1615 * Either we need to flush the
1616 * kmem (net_queue) queue or the disk (disk_queue)
1617 * determine which, and do it.
1618 */
1619 void
1620 rdc_flusher_thread(int index)
1621 {
1622 rdc_k_info_t *krdc = &rdc_k_info[index];
1623
1624 if (krdc->group->flags & RDC_MEMQUE) {
1625 rdc_flush_memq(index);
1626 return;
1627 } else if (krdc->group->flags & RDC_DISKQUE) {
1628 rdc_flush_diskq(index);
1629 return;
1630 } else { /* uh-oh, big time */
1631 cmn_err(CE_PANIC, "flusher trying to flush unknown queue type");
1632 }
1633
1634 }
1635
1636 void
1637 rdc_flush_memq(int index)
1638 {
1639 rdc_k_info_t *krdc = &rdc_k_info[index];
1640 rdc_aio_t *aio;
1641 net_queue *q;
1642 int dowork;
1643 rdc_group_t *group = krdc->group;
1644 if (!group || group->count == 0) {
1645 #ifdef DEBUG
1646 cmn_err(CE_WARN, "!rdc_flush_memq: no group left!");
1647 #endif
1648 return;
1649 }
1650
1651 if (!krdc->c_fd) {
1652 #ifdef DEBUG
1653 cmn_err(CE_WARN, "!rdc_flush_memq: no c_fd!");
1654 #endif
1655 goto thread_death;
1656 }
1657
1658 #ifdef DEBUG_DISABLE
1659 if (index < 0 || (krdc->type_flag & RDC_DISABLEPEND)) {
1660 cmn_err(CE_WARN, "!rdc_flush_memq: DISABLE PENDING!");
1661 /*
1662 * Need to continue as we may be trying to flush IO
1663 * while trying to disable or suspend
1664 */
1665 }
1666 #endif
1667
1668 q = &group->ra_queue;
1669
1670 dowork = 1;
1671 /* CONSTCOND */
1672 while (dowork) {
1673 if (net_exit == ATM_EXIT)
1674 break;
1675
1676 group = krdc->group;
1677 if (!group || group->count == 0) {
1678 #ifdef DEBUG
1679 cmn_err(CE_WARN, "!rdc_flush_memq: no group left!");
1680 #endif
1681 break;
1682 }
1683
1684 mutex_enter(&q->net_qlock);
1685 aio = q->net_qhead;
1686
1687 if (aio == NULL) {
1688 #ifdef DEBUG
1689 if (q->nitems != 0 ||
1690 q->blocks != 0 ||
1691 q->net_qtail != 0) {
1692 cmn_err(CE_PANIC,
1693 "rdc_flush_memq(1): q %p, q blocks %"
1694 NSC_SZFMT ", nitems %" NSC_SZFMT
1695 ", qhead %p qtail %p",
1696 (void *) q, q->blocks, q->nitems,
1697 (void *) aio, (void *) q->net_qtail);
1698 }
1699 #endif
1700 mutex_exit(&q->net_qlock);
1701 break;
1702 }
1703
1704 /* aio remove from q */
1705
1706 q->net_qhead = aio->next;
1707 aio->next = NULL;
1708
1709 if (q->net_qtail == aio)
1710 q->net_qtail = q->net_qhead;
1711
1712 q->blocks -= aio->len;
1713 q->nitems--;
1714
1715 /*
1716 * in flight numbers.
1717 */
1718 q->inflbls += aio->len;
1719 q->inflitems++;
1720
1721 #ifdef DEBUG
1722 if (q->net_qhead == NULL) {
1723 if (q->nitems != 0 ||
1724 q->blocks != 0 ||
1725 q->net_qtail != 0) {
1726 cmn_err(CE_PANIC,
1727 "rdc_flush_memq(2): q %p, q blocks %"
1728 NSC_SZFMT ", nitems %" NSC_SZFMT
1729 ", qhead %p qtail %p",
1730 (void *) q, q->blocks, q->nitems,
1731 (void *) q->net_qhead,
1732 (void *) q->net_qtail);
1733 }
1734 }
1735
1736 #ifndef NSC_MULTI_TERABYTE
1737 if (q->blocks < 0) {
1738 cmn_err(CE_PANIC,
1739 "rdc_flush_memq(3): q %p, q blocks %" NSC_SZFMT
1740 ", nitems %d, qhead %p, qtail %p",
1741 (void *) q, q->blocks, q->nitems,
1742 (void *) q->net_qhead, (void *) q->net_qtail);
1743 }
1744 #else
1745 /* blocks and nitems are unsigned for NSC_MULTI_TERABYTE */
1746 #endif
1747 #endif
1748
1749 mutex_exit(&q->net_qlock);
1750
1751 aio->iostatus = RDC_IO_INIT;
1752
1753 _rdc_remote_flush(aio);
1754
1755 mutex_enter(&q->net_qlock);
1756 q->inflbls -= aio->len;
1757 q->inflitems--;
1758 if ((group->seqack == RDC_NEWSEQ) &&
1759 (group->seq != RDC_NEWSEQ + 1)) {
1760 if ((q->net_qhead == NULL) ||
1761 (q->net_qhead->seq != RDC_NEWSEQ + 1)) {
1762 /*
1763 * We are an old thread, and the
1764 * queue sequence has been reset
1765 * during the network write above.
1766 * As such we mustn't pull another
1767 * job from the queue until the
1768 * first sequence message has been ack'ed.
1769 * Just die instead. Unless this thread
1770 * is the first sequence that has just
1771 * been ack'ed
1772 */
1773 dowork = 0;
1774 }
1775 }
1776 mutex_exit(&q->net_qlock);
1777
1778 if ((aio->iostatus != RDC_IO_DONE) && (group->count)) {
1779 rdc_k_info_t *krdctmp = &rdc_k_info[aio->index];
1780 if (krdctmp->type_flag & RDC_DISABLEPEND) {
1781 kmem_free(aio, sizeof (*aio));
1782 goto thread_death;
1783 }
1784 rdc_group_enter(krdc);
1785 ASSERT(krdc->group);
1786 rdc_group_log(krdc, RDC_NOFLUSH | RDC_ALLREMOTE,
1787 "memq flush aio status not RDC_IO_DONE");
1788 rdc_group_exit(krdc);
1789 rdc_dump_queue(aio->index);
1790 }
1791 kmem_free(aio, sizeof (*aio));
1792
1793 if (krdc->remote_index < 0 || !krdc->lsrv || !krdc->intf)
1794 break;
1795 }
1796
1797 thread_death:
1798 rdc_many_enter(krdc);
1799 mutex_enter(&group->ra_queue.net_qlock);
1800 group->rdc_thrnum--;
1801 group->rdc_writer = 0;
1802 /*
1803 * all threads must be dead.
1804 */
1805 if ((group->count == 0) && (group->rdc_thrnum == 0)) {
1806 mutex_exit(&group->ra_queue.net_qlock);
1807 /*
1808 * Group now empty, so destroy
1809 * Race with remove_from_group while write thread was running
1810 */
1811 #ifdef DEBUG
1812 cmn_err(CE_WARN, "!rdc_flush_memq: group being destroyed");
1813 #endif
1814 rdc_delgroup(group);
1815 krdc->group = NULL;
1816 rdc_many_exit(krdc);
1817 return;
1818 }
1819 mutex_exit(&group->ra_queue.net_qlock);
1820 rdc_many_exit(krdc);
1821 }
1822
1823 /*
1824 * rdc_flush_diskq
1825 * disk queue flusher
1826 */
1827 void
1828 rdc_flush_diskq(int index)
1829 {
1830 rdc_k_info_t *krdc = &rdc_k_info[index];
1831 rdc_u_info_t *urdc = &rdc_u_info[index];
1832 rdc_aio_t *aio = NULL;
1833 disk_queue *q;
1834 net_queue *nq;
1835 int dowork;
1836 int rc;
1837 rdc_group_t *group = krdc->group;
1838
1839 if (!group || group->count == 0) {
1840 #ifdef DEBUG
1841 cmn_err(CE_WARN, "!rdc_flush_diskq: no group left!");
1842 #endif
1843 return;
1844 }
1845
1846 if (!krdc->c_fd) {
1847 #ifdef DEBUG
1848 cmn_err(CE_WARN, "!rdc_flush_diskq: no c_fd!");
1849 #endif
1850 return;
1851 }
1852
1853 #ifdef DEBUG_DISABLE
1854 if (index < 0 || (krdc->type_flag & RDC_DISABLEPEND)) {
1855 cmn_err(CE_WARN, "!rdc_flush_diskq: DISABLE PENDING!");
1856 /*
1857 * Need to continue as we may be trying to flush IO
1858 * while trying to disable or suspend
1859 */
1860 }
1861 #endif
1862 q = &group->diskq;
1863 nq = &group->ra_queue;
1864
1865 if (IS_QSTATE(q, RDC_QDISABLEPEND) || IS_STATE(urdc, RDC_LOGGING)) {
1866 #ifdef DEBUG
1867 cmn_err(CE_NOTE, "!flusher thread death 1 %x", QSTATE(q));
1868 #endif
1869 goto thread_death;
1870 }
1871
1872 dowork = 1;
1873 /* CONSTCOND */
1874 while (dowork) {
1875 if (net_exit == ATM_EXIT)
1876 break;
1877
1878 group = krdc->group;
1879 if (!group || group->count == 0) {
1880 #ifdef DEBUG
1881 cmn_err(CE_WARN, "!rdc_flush_diskq: no group left!");
1882 #endif
1883 break;
1884 }
1885
1886 do {
1887 rc = 0;
1888 if ((IS_STATE(urdc, RDC_LOGGING)) ||
1889 (IS_STATE(urdc, RDC_SYNCING)) ||
1890 (nq->qfflags & RDC_QFILLSLEEP))
1891 goto thread_death;
1892
1893 aio = rdc_dequeue(krdc, &rc);
1894
1895 if ((IS_STATE(urdc, RDC_LOGGING)) ||
1896 (IS_STATE(urdc, RDC_SYNCING)) ||
1897 (nq->qfflags & RDC_QFILLSLEEP)) {
1898 goto thread_death;
1899 }
1900 if (rc == EAGAIN) {
1901 delay(40);
1902 }
1903
1904 } while (rc == EAGAIN);
1905
1906 if (aio == NULL) {
1907 break;
1908 }
1909
1910 aio->iostatus = RDC_IO_INIT;
1911
1912 mutex_enter(QLOCK(q));
1913 q->inflbls += aio->len;
1914 q->inflitems++;
1915 mutex_exit(QLOCK(q));
1916
1917 _rdc_remote_flush(aio);
1918
1919 mutex_enter(QLOCK(q));
1920 q->inflbls -= aio->len;
1921 q->inflitems--;
1922
1923 if ((group->seqack == RDC_NEWSEQ) &&
1924 (group->seq != RDC_NEWSEQ + 1)) {
1925 if ((nq->net_qhead == NULL) ||
1926 (nq->net_qhead->seq != RDC_NEWSEQ + 1)) {
1927 /*
1928 * We are an old thread, and the
1929 * queue sequence has been reset
1930 * during the network write above.
1931 * As such we mustn't pull another
1932 * job from the queue until the
1933 * first sequence message has been ack'ed.
1934 * Just die instead. Unless of course,
1935 * this thread is the first sequence that
1936 * has just been ack'ed.
1937 */
1938 dowork = 0;
1939 }
1940 }
1941 mutex_exit(QLOCK(q));
1942
1943 if (aio->iostatus == RDC_IO_CANCELLED) {
1944 rdc_dump_queue(aio->index);
1945 kmem_free(aio, sizeof (*aio));
1946 aio = NULL;
1947 if (group) { /* seq gets bumped on dequeue */
1948 mutex_enter(QLOCK(q));
1949 rdc_dump_iohdrs(q);
1950 SET_QNXTIO(q, QHEAD(q));
1951 SET_QCOALBOUNDS(q, QHEAD(q));
1952 group->seq = RDC_NEWSEQ;
1953 group->seqack = RDC_NEWSEQ;
1954 mutex_exit(QLOCK(q));
1955 }
1956 break;
1957 }
1958
1959 if ((aio->iostatus != RDC_IO_DONE) && (group->count)) {
1960 rdc_k_info_t *krdctmp = &rdc_k_info[aio->index];
1961 if (krdctmp->type_flag & RDC_DISABLEPEND) {
1962 kmem_free(aio, sizeof (*aio));
1963 aio = NULL;
1964 goto thread_death;
1965 }
1966 rdc_group_enter(krdc);
1967 rdc_group_log(krdc,
1968 RDC_NOFLUSH | RDC_ALLREMOTE | RDC_QUEUING,
1969 "diskq flush aio status not RDC_IO_DONE");
1970 rdc_group_exit(krdc);
1971 rdc_dump_queue(aio->index);
1972 }
1973
1974 kmem_free(aio, sizeof (*aio));
1975 aio = NULL;
1976
1977 #ifdef DEBUG_DISABLE
1978 if (krdc->type_flag & RDC_DISABLEPEND) {
1979 cmn_err(CE_WARN,
1980 "!rdc_flush_diskq: DISABLE PENDING after IO!");
1981 }
1982 #endif
1983 if (krdc->remote_index < 0 || !krdc->lsrv || !krdc->intf)
1984 break;
1985
1986 if (IS_QSTATE(q, RDC_QDISABLEPEND)) {
1987 #ifdef DEBUG
1988 cmn_err(CE_NOTE, "!flusher thread death 2");
1989 #endif
1990 break;
1991 }
1992 }
1993 thread_death:
1994 rdc_many_enter(krdc);
1995 mutex_enter(QLOCK(q));
1996 group->rdc_thrnum--;
1997 group->rdc_writer = 0;
1998
1999 if (aio && aio->qhandle) {
2000 aio->qhandle->sb_user--;
2001 if (aio->qhandle->sb_user == 0) {
2002 (void) _rdc_rsrv_diskq(krdc->group);
2003 rdc_fixlen(aio);
2004 (void) nsc_free_buf(aio->qhandle);
2005 aio->qhandle = NULL;
2006 aio->handle = NULL;
2007 _rdc_rlse_diskq(krdc->group);
2008 }
2009 }
2010 if ((group->count == 0) && (group->rdc_thrnum == 0)) {
2011 mutex_exit(QLOCK(q));
2012 /*
2013 * Group now empty, so destroy
2014 * Race with remove_from_group while write thread was running
2015 */
2016 #ifdef DEBUG
2017 cmn_err(CE_WARN, "!rdc_flush_diskq: group being destroyed");
2018 #endif
2019 mutex_enter(&group->diskqmutex);
2020 rdc_close_diskq(group);
2021 mutex_exit(&group->diskqmutex);
2022 rdc_delgroup(group);
2023 krdc->group = NULL;
2024 rdc_many_exit(krdc);
2025 return;
2026 }
2027 mutex_exit(QLOCK(q));
2028 rdc_many_exit(krdc);
2029 }
2030
2031 /*
2032 * _rdc_remote_flush
2033 * Flush a single block ANON block
2034 * this function will flush from either the disk queue
2035 * or the memory queue. The appropriate locks must be
2036 * taken out etc, etc ...
2037 */
2038 static void
2039 _rdc_remote_flush(rdc_aio_t *aio)
2040 {
2041 rdc_k_info_t *krdc = &rdc_k_info[aio->index];
2042 rdc_u_info_t *urdc = &rdc_u_info[aio->index];
2043 disk_queue *q = &krdc->group->diskq;
2044 kmutex_t *qlock;
2045 rdc_group_t *group;
2046 nsc_buf_t *h = NULL;
2047 int reserved = 0;
2048 int rtype = RDC_RAW;
2049 int rc;
2050 uint_t maxseq;
2051 struct netwriteres netret;
2052 int waitq = 1;
2053 int vflags;
2054
2055 group = krdc->group;
2056 netret.vecdata.vecdata_val = NULL;
2057 netret.vecdata.vecdata_len = 0;
2058
2059 /* Where did we get this aio from anyway? */
2060 if (RDC_IS_DISKQ(group)) {
2061 qlock = &group->diskq.disk_qlock;
2062 } else {
2063 qlock = &group->ra_queue.net_qlock;
2064 }
2065
2066 /*
2067 * quench transmission if we are too far ahead of the
2068 * server Q, or it will overflow.
2069 * Must fail all requests while asyncdis is set.
2070 * It will be cleared when the last thread to be discarded
2071 * sets the asyncstall counter to zero.
2072 * Note the thread within rdc_net_write
2073 * also bumps the asyncstall counter.
2074 */
2075
2076 mutex_enter(qlock);
2077 if (group->asyncdis) {
2078 aio->iostatus = RDC_IO_CANCELLED;
2079 mutex_exit(qlock);
2080 goto failed;
2081 }
2082 /* don't go to sleep if we have gone logging! */
2083 vflags = rdc_get_vflags(urdc);
2084 if ((vflags & (RDC_BMP_FAILED|RDC_VOL_FAILED|RDC_LOGGING))) {
2085 if ((vflags & RDC_LOGGING) && RDC_IS_DISKQ(group))
2086 aio->iostatus = RDC_IO_CANCELLED;
2087
2088 mutex_exit(qlock);
2089 goto failed;
2090 }
2091
2092 while (maxseq = group->seqack + RDC_MAXPENDQ + 1,
2093 maxseq = (maxseq < group->seqack) ? maxseq + RDC_NEWSEQ + 1
2094 : maxseq, !RDC_INFRONT(aio->seq, maxseq)) {
2095 group->asyncstall++;
2096 ASSERT(!IS_STATE(urdc, RDC_LOGGING));
2097 cv_wait(&group->asyncqcv, qlock);
2098 group->asyncstall--;
2099 ASSERT(group->asyncstall >= 0);
2100 if (group->asyncdis) {
2101 if (group->asyncstall == 0) {
2102 group->asyncdis = 0;
2103 }
2104 aio->iostatus = RDC_IO_CANCELLED;
2105 mutex_exit(qlock);
2106 goto failed;
2107 }
2108 /*
2109 * See if we have gone into logging mode
2110 * since sleeping.
2111 */
2112 vflags = rdc_get_vflags(urdc);
2113 if (vflags & (RDC_BMP_FAILED|RDC_VOL_FAILED|RDC_LOGGING)) {
2114 if ((vflags & RDC_LOGGING) && RDC_IS_DISKQ(group))
2115 aio->iostatus = RDC_IO_CANCELLED;
2116
2117 mutex_exit(qlock);
2118 goto failed;
2119 }
2120 }
2121 mutex_exit(qlock);
2122
2123 if ((krdc->io_kstats) && (!RDC_IS_DISKQ(krdc->group))) {
2124 mutex_enter(krdc->io_kstats->ks_lock);
2125 kstat_waitq_exit(KSTAT_IO_PTR(krdc->io_kstats));
2126 mutex_exit(krdc->io_kstats->ks_lock);
2127 waitq = 0;
2128 }
2129
2130
2131 rc = _rdc_rsrv_devs(krdc, rtype, RDC_INTERNAL);
2132 if (rc != 0) {
2133 #ifdef DEBUG
2134 cmn_err(CE_WARN, "!_rdc_remote_flush: reserve, index %d, rc %d",
2135 aio->index, rc);
2136 #endif
2137 goto failed;
2138 }
2139
2140 reserved = 1;
2141 /*
2142 * Case where we are multihop and calling with no ANON bufs
2143 * Need to do the read to fill the buf.
2144 */
2145 if (!aio->handle) {
2146 rc = nsc_alloc_buf(RDC_U_FD(krdc), aio->pos, aio->len,
2147 (aio->flag & ~NSC_WRITE) | NSC_READ, &h);
2148 if (!RDC_SUCCESS(rc)) {
2149 #ifdef DEBUG
2150 cmn_err(CE_WARN,
2151 "!_rdc_remote_flush: alloc_buf, index %d, pos %"
2152 NSC_SZFMT ", len %" NSC_SZFMT ", rc %d",
2153 aio->index, aio->pos, aio->len, rc);
2154 #endif
2155
2156 goto failed;
2157 }
2158 aio->handle = h;
2159 aio->handle->sb_user = RDC_NULLBUFREAD;
2160 }
2161
2162 mutex_enter(qlock);
2163 if (group->asyncdis) {
2164 if (group->asyncstall == 0) {
2165 group->asyncdis = 0;
2166 }
2167 aio->iostatus = RDC_IO_CANCELLED;
2168 mutex_exit(qlock);
2169 goto failed;
2170 }
2171 group->asyncstall++;
2172 mutex_exit(qlock);
2173
2174
2175 if (krdc->remote_index < 0) {
2176 /*
2177 * this should be ok, we are flushing, not rev syncing.
2178 * remote_index could be -1 if we lost a race with
2179 * resume and the flusher trys to flush an io from
2180 * another set that has not resumed
2181 */
2182 krdc->remote_index = rdc_net_state(krdc->index, CCIO_SLAVE);
2183 DTRACE_PROBE1(remote_index_negative, int, krdc->remote_index);
2184
2185 }
2186
2187 /*
2188 * double check for logging, no check in net_write()
2189 * skip the write if you can, otherwise, if logging
2190 * avoid clearing the bit .. you don't know whose bit it may
2191 * also be.
2192 */
2193 if (IS_STATE(urdc, RDC_LOGGING) || IS_STATE(urdc, RDC_SYNCING)) {
2194 aio->iostatus = RDC_IO_CANCELLED;
2195 mutex_enter(qlock);
2196 group->asyncstall--;
2197 mutex_exit(qlock);
2198 goto failed;
2199 }
2200
2201 rc = rdc_net_write(krdc->index, krdc->remote_index,
2202 aio->handle, aio->pos, aio->len, aio->seq, aio->qpos, &netret);
2203
2204 mutex_enter(qlock);
2205 group->asyncstall--;
2206 if (group->asyncdis) {
2207 if (group->asyncstall == 0) {
2208 group->asyncdis = 0;
2209 }
2210 aio->iostatus = RDC_IO_CANCELLED;
2211 mutex_exit(qlock);
2212 goto failed;
2213 }
2214
2215 if (IS_STATE(urdc, RDC_LOGGING) || IS_STATE(urdc, RDC_SYNCING)) {
2216 mutex_exit(qlock);
2217 aio->iostatus = RDC_IO_CANCELLED;
2218 goto failed;
2219 }
2220
2221 ASSERT(aio->handle);
2222 if (rc != 0) {
2223 #ifdef DEBUG
2224 cmn_err(CE_WARN,
2225 "!_rdc_remote_flush: write, index %d, pos %" NSC_SZFMT
2226 ", len %" NSC_SZFMT ", "
2227 "rc %d seq %u group seq %u seqack %u qpos %" NSC_SZFMT,
2228 aio->index, aio->pos, aio->len, rc, aio->seq,
2229 group->seq, group->seqack, aio->qpos);
2230 #endif
2231 if (rc == ENOLINK) {
2232 cmn_err(CE_WARN,
2233 "!Hard timeout detected (%d sec) "
2234 "on SNDR set %s:%s",
2235 rdc_rpc_tmout, urdc->secondary.intf,
2236 urdc->secondary.file);
2237 }
2238 mutex_exit(qlock);
2239 goto failed;
2240 } else {
2241 aio->iostatus = RDC_IO_DONE;
2242 }
2243
2244 if (RDC_IS_DISKQ(group)) {
2245 /* free locally alloc'd handle */
2246 if (aio->handle->sb_user == RDC_NULLBUFREAD) {
2247 (void) nsc_free_buf(aio->handle);
2248 aio->handle = NULL;
2249 }
2250 aio->qhandle->sb_user--;
2251 if (aio->qhandle->sb_user == 0) {
2252 (void) _rdc_rsrv_diskq(group);
2253 rdc_fixlen(aio);
2254 (void) nsc_free_buf(aio->qhandle);
2255 aio->qhandle = NULL;
2256 aio->handle = NULL;
2257 _rdc_rlse_diskq(group);
2258 }
2259
2260 } else {
2261 (void) nsc_free_buf(aio->handle);
2262 aio->handle = NULL;
2263 }
2264
2265 mutex_exit(qlock);
2266
2267 _rdc_rlse_devs(krdc, rtype);
2268
2269 if (netret.result == 0) {
2270 vflags = rdc_get_vflags(urdc);
2271
2272 if (!(vflags & (RDC_BMP_FAILED|RDC_VOL_FAILED|RDC_LOGGING))) {
2273 RDC_CLR_BITMAP(krdc, aio->pos, aio->len, \
2274 0xffffffff, RDC_BIT_BUMP);
2275
2276 if (RDC_IS_DISKQ(krdc->group)) {
2277 if (!IS_STATE(urdc, RDC_LOGGING)) {
2278 /* tell queue data has been flushed */
2279 rdc_clr_iohdr(krdc, aio->qpos);
2280 } else { /* throw away queue, logging */
2281 mutex_enter(qlock);
2282 rdc_dump_iohdrs(q);
2283 SET_QNXTIO(q, QHEAD(q));
2284 SET_QCOALBOUNDS(q, QHEAD(q));
2285 mutex_exit(qlock);
2286 }
2287 }
2288 }
2289
2290 mutex_enter(qlock);
2291 /*
2292 * Check to see if the reply has arrived out of
2293 * order, if so don't update seqack.
2294 */
2295 if (!RDC_INFRONT(aio->seq, group->seqack)) {
2296 group->seqack = aio->seq;
2297 }
2298 #ifdef DEBUG
2299 else {
2300 rdc_ooreply++;
2301 }
2302 #endif
2303 if (group->asyncstall) {
2304 cv_broadcast(&group->asyncqcv);
2305 }
2306 mutex_exit(qlock);
2307 } else if (netret.result < 0) {
2308 aio->iostatus = RDC_IO_FAILED;
2309 }
2310
2311 /*
2312 * see if we have any pending async requests we can mark
2313 * as done.
2314 */
2315
2316 if (netret.vecdata.vecdata_len) {
2317 net_pendvec_t *vecp;
2318 net_pendvec_t *vecpe;
2319 vecp = netret.vecdata.vecdata_val;
2320 vecpe = netret.vecdata.vecdata_val + netret.vecdata.vecdata_len;
2321 while (vecp < vecpe) {
2322 rdc_k_info_t *krdcp = &rdc_k_info[vecp->pindex];
2323 rdc_u_info_t *urdcp = &rdc_u_info[vecp->pindex];
2324 /*
2325 * we must always still be in the same group.
2326 */
2327 ASSERT(krdcp->group == group);
2328 vflags = rdc_get_vflags(urdcp);
2329
2330 if (!(vflags &
2331 (RDC_BMP_FAILED|RDC_VOL_FAILED|RDC_LOGGING))) {
2332 RDC_CLR_BITMAP(krdcp, vecp->apos, vecp->alen, \
2333 0xffffffff, RDC_BIT_BUMP);
2334 if (RDC_IS_DISKQ(krdcp->group)) {
2335 if (!IS_STATE(urdc, RDC_LOGGING)) {
2336 /* update queue info */
2337 rdc_clr_iohdr(krdc, vecp->qpos);
2338 } else { /* we've gone logging */
2339 mutex_enter(qlock);
2340 rdc_dump_iohdrs(q);
2341 SET_QNXTIO(q, QHEAD(q));
2342 SET_QCOALBOUNDS(q, QHEAD(q));
2343 mutex_exit(qlock);
2344 }
2345 }
2346 }
2347
2348 /*
2349 * see if we can re-start transmission
2350 */
2351 mutex_enter(qlock);
2352 if (!RDC_INFRONT(vecp->seq, group->seqack)) {
2353 group->seqack = vecp->seq;
2354 }
2355 #ifdef DEBUG
2356 else {
2357 rdc_ooreply++;
2358 }
2359 #endif
2360 DTRACE_PROBE1(pendvec_return, int, vecp->seq);
2361
2362 if (group->asyncstall) {
2363 cv_broadcast(&group->asyncqcv);
2364 }
2365 mutex_exit(qlock);
2366 vecp++;
2367 }
2368 }
2369 if (netret.vecdata.vecdata_val)
2370 kmem_free(netret.vecdata.vecdata_val,
2371 netret.vecdata.vecdata_len * sizeof (net_pendvec_t));
2372 return;
2373 failed:
2374
2375 /* perhaps we have a few threads stuck .. */
2376 if (group->asyncstall) {
2377 group->asyncdis = 1;
2378 cv_broadcast(&group->asyncqcv);
2379 }
2380 if (netret.vecdata.vecdata_val)
2381 kmem_free(netret.vecdata.vecdata_val,
2382 netret.vecdata.vecdata_len * sizeof (net_pendvec_t));
2383
2384 mutex_enter(qlock);
2385 if (RDC_IS_DISKQ(group)) {
2386 /* free locally alloc'd hanlde */
2387 if ((aio->handle) &&
2388 (aio->handle->sb_user == RDC_NULLBUFREAD)) {
2389 (void) nsc_free_buf(aio->handle);
2390 aio->handle = NULL;
2391 }
2392 aio->qhandle->sb_user--;
2393 if (aio->qhandle->sb_user == 0) {
2394 (void) _rdc_rsrv_diskq(group);
2395 rdc_fixlen(aio);
2396 (void) nsc_free_buf(aio->qhandle);
2397 aio->qhandle = NULL;
2398 aio->handle = NULL;
2399 _rdc_rlse_diskq(group);
2400 }
2401 } else {
2402 if (aio->handle) {
2403 (void) nsc_free_buf(aio->handle);
2404 aio->handle = NULL;
2405 }
2406 }
2407 mutex_exit(qlock);
2408
2409 if (reserved) {
2410 _rdc_rlse_devs(krdc, rtype);
2411 }
2412
2413 if ((waitq && krdc->io_kstats) && (!RDC_IS_DISKQ(krdc->group))) {
2414 mutex_enter(krdc->io_kstats->ks_lock);
2415 kstat_waitq_exit(KSTAT_IO_PTR(krdc->io_kstats));
2416 mutex_exit(krdc->io_kstats->ks_lock);
2417 }
2418
2419 /* make sure that the bit is still set */
2420 RDC_CHECK_BIT(krdc, aio->pos, aio->len);
2421
2422 if (aio->iostatus != RDC_IO_CANCELLED)
2423 aio->iostatus = RDC_IO_FAILED;
2424 }
2425
2426
2427 /*
2428 * rdc_drain_disk_queue
2429 * drain the async network queue for the whole group. Bail out if nothing
2430 * happens in 20 sec
2431 * returns -1 if it bails before the queues are drained.
2432 */
2433 #define NUM_RETRIES 15 /* Number of retries to wait if no progress */
2434 int
2435 rdc_drain_disk_queue(int index)
2436 {
2437 rdc_k_info_t *krdc = &rdc_k_info[index];
2438 volatile rdc_group_t *group;
2439 volatile disk_queue *diskq;
2440 int threads, counter;
2441 long blocks;
2442
2443 /* Sanity checking */
2444 if (index > rdc_max_sets)
2445 return (0);
2446
2447 /*
2448 * If there is no group or diskq configured, we can leave now
2449 */
2450 if (!(group = krdc->group) || !(diskq = &group->diskq))
2451 return (0);
2452
2453 /*
2454 * No need to wait if EMPTY and threads are gone
2455 */
2456 counter = 0;
2457 while (!QEMPTY(diskq) || group->rdc_thrnum) {
2458
2459 /*
2460 * Capture counters to determine if progress is being made
2461 */
2462 blocks = QBLOCKS(diskq);
2463 threads = group->rdc_thrnum;
2464
2465 /*
2466 * Wait
2467 */
2468 delay(HZ);
2469
2470 /*
2471 * Has the group or disk queue gone away while delayed?
2472 */
2473 if (!(group = krdc->group) || !(diskq = &group->diskq))
2474 return (0);
2475
2476 /*
2477 * Are we still seeing progress?
2478 */
2479 if (blocks == QBLOCKS(diskq) && threads == group->rdc_thrnum) {
2480 /*
2481 * No progress seen, increment retry counter
2482 */
2483 if (counter++ > NUM_RETRIES) {
2484 return (-1);
2485 }
2486 } else {
2487 /*
2488 * Reset counter, as we've made progress
2489 */
2490 counter = 0;
2491 }
2492 }
2493
2494 return (0);
2495 }
2496
2497 /*
2498 * decide what needs to be drained, disk or core
2499 * and drain it
2500 */
2501 int
2502 rdc_drain_queue(int index)
2503 {
2504 rdc_k_info_t *krdc = &rdc_k_info[index];
2505 rdc_group_t *group = krdc->group;
2506
2507 if (!group)
2508 return (0);
2509
2510 if (RDC_IS_DISKQ(group))
2511 return (rdc_drain_disk_queue(index));
2512 if (RDC_IS_MEMQ(group))
2513 return (rdc_drain_net_queue(index));
2514 /* oops.. */
2515 #ifdef DEBUG
2516 cmn_err(CE_WARN, "!rdc_drain_queue: "
2517 "attempting drain of unknown Q type");
2518 #endif
2519 return (0);
2520 }
2521
2522 /*
2523 * rdc_drain_net_queue
2524 * drain the async network queue for the whole group. Bail out if nothing
2525 * happens in 20 sec
2526 * returns -1 if it bails before the queues are drained.
2527 */
2528 int
2529 rdc_drain_net_queue(int index)
2530 {
2531 rdc_k_info_t *krdc = &rdc_k_info[index];
2532 volatile net_queue *q;
2533 int bail = 20; /* bail out in about 20 secs */
2534 nsc_size_t blocks;
2535
2536 /* Sanity checking */
2537 if (index > rdc_max_sets)
2538 return (0);
2539 if (!krdc->group)
2540 return (0);
2541 /* LINTED */
2542 if (!(q = &krdc->group->ra_queue))
2543 return (0);
2544
2545 /* CONSTCOND */
2546 while (1) {
2547
2548 if (((volatile rdc_aio_t *)q->net_qhead == NULL) &&
2549 (krdc->group->rdc_thrnum == 0)) {
2550 break;
2551 }
2552
2553 blocks = q->blocks;
2554
2555 q = (volatile net_queue *)&krdc->group->ra_queue;
2556
2557 if ((blocks == q->blocks) &&
2558 (--bail <= 0)) {
2559 break;
2560 }
2561
2562 delay(HZ);
2563 }
2564
2565 if (bail <= 0)
2566 return (-1);
2567
2568 return (0);
2569 }
2570
2571 /*
2572 * rdc_dump_queue
2573 * We want to release all the blocks currently on the network flushing queue
2574 * We already have them logged in the bitmap.
2575 */
2576 void
2577 rdc_dump_queue(int index)
2578 {
2579 rdc_k_info_t *krdc = &rdc_k_info[index];
2580 rdc_aio_t *aio;
2581 net_queue *q;
2582 rdc_group_t *group;
2583 disk_queue *dq;
2584 kmutex_t *qlock;
2585
2586 group = krdc->group;
2587
2588 q = &group->ra_queue;
2589 dq = &group->diskq;
2590
2591 /*
2592 * gotta have both locks here for diskq
2593 */
2594
2595 if (RDC_IS_DISKQ(group)) {
2596 mutex_enter(&q->net_qlock);
2597 if (q->qfill_sleeping == RDC_QFILL_AWAKE) {
2598 int tries = 3;
2599 #ifdef DEBUG_DISKQ
2600 cmn_err(CE_NOTE,
2601 "!dumpq sending diskq->memq flusher to sleep");
2602 #endif
2603 q->qfflags |= RDC_QFILLSLEEP;
2604 mutex_exit(&q->net_qlock);
2605 while (q->qfill_sleeping == RDC_QFILL_AWAKE && tries--)
2606 delay(5);
2607 mutex_enter(&q->net_qlock);
2608 }
2609 }
2610
2611 if (RDC_IS_DISKQ(group)) {
2612 qlock = &dq->disk_qlock;
2613 (void) _rdc_rsrv_diskq(group);
2614 } else {
2615 qlock = &q->net_qlock;
2616 }
2617
2618 mutex_enter(qlock);
2619
2620 group->seq = RDC_NEWSEQ; /* reset the sequence number */
2621 group->seqack = RDC_NEWSEQ;
2622
2623 /* if the q is on disk, dump the q->iohdr chain */
2624 if (RDC_IS_DISKQ(group)) {
2625 rdc_dump_iohdrs(dq);
2626
2627 /* back up the nxtio pointer */
2628 SET_QNXTIO(dq, QHEAD(dq));
2629 SET_QCOALBOUNDS(dq, QHEAD(dq));
2630 }
2631
2632 while (q->net_qhead) {
2633 rdc_k_info_t *tmpkrdc;
2634 aio = q->net_qhead;
2635 tmpkrdc = &rdc_k_info[aio->index];
2636
2637 if (RDC_IS_DISKQ(group)) {
2638 aio->qhandle->sb_user--;
2639 if (aio->qhandle->sb_user == 0) {
2640 rdc_fixlen(aio);
2641 (void) nsc_free_buf(aio->qhandle);
2642 aio->qhandle = NULL;
2643 aio->handle = NULL;
2644 }
2645 } else {
2646 if (aio->handle) {
2647 (void) nsc_free_buf(aio->handle);
2648 aio->handle = NULL;
2649 }
2650 }
2651
2652 q->net_qhead = aio->next;
2653 RDC_CHECK_BIT(tmpkrdc, aio->pos, aio->len);
2654
2655 kmem_free(aio, sizeof (*aio));
2656 if (tmpkrdc->io_kstats && !RDC_IS_DISKQ(group)) {
2657 mutex_enter(tmpkrdc->io_kstats->ks_lock);
2658 kstat_waitq_exit(KSTAT_IO_PTR(tmpkrdc->io_kstats));
2659 mutex_exit(tmpkrdc->io_kstats->ks_lock);
2660 }
2661
2662 }
2663
2664 q->net_qtail = NULL;
2665 q->blocks = 0;
2666 q->nitems = 0;
2667
2668 /*
2669 * See if we have stalled threads.
2670 */
2671 done:
2672 if (group->asyncstall) {
2673 group->asyncdis = 1;
2674 cv_broadcast(&group->asyncqcv);
2675 }
2676 mutex_exit(qlock);
2677 if (RDC_IS_DISKQ(group)) {
2678 mutex_exit(&q->net_qlock);
2679 _rdc_rlse_diskq(group);
2680 }
2681
2682 }
2683
2684
2685 /*
2686 * rdc_clnt_get
2687 * Get a CLIENT handle and cache it
2688 */
2689
2690 static int
2691 rdc_clnt_get(rdc_srv_t *svp, rpcvers_t vers, struct chtab **rch, CLIENT **clp)
2692 {
2693 uint_t max_msgsize;
2694 int retries;
2695 int ret;
2696 struct cred *cred;
2697 int num_clnts = 0;
2698 register struct chtab *ch;
2699 struct chtab **plistp;
2700 CLIENT *client = 0;
2701
2702 if (rch) {
2703 *rch = 0;
2704 }
2705
2706 if (clp) {
2707 *clp = 0;
2708 }
2709
2710 retries = 6; /* Never used for COTS in Solaris */
2711 cred = ddi_get_cred();
2712 max_msgsize = RDC_RPC_MAX;
2713
2714 mutex_enter(&rdc_clnt_lock);
2715
2716 ch = rdc_chtable;
2717 plistp = &rdc_chtable;
2718
2719 /* find the right ch_list chain */
2720
2721 for (ch = rdc_chtable; ch != NULL; ch = ch->ch_next) {
2722 if (ch->ch_prog == RDC_PROGRAM &&
2723 ch->ch_vers == vers &&
2724 ch->ch_dev == svp->ri_knconf->knc_rdev &&
2725 ch->ch_protofmly != NULL &&
2726 strcmp(ch->ch_protofmly,
2727 svp->ri_knconf->knc_protofmly) == 0) {
2728 /* found the correct chain to walk */
2729 break;
2730 }
2731 plistp = &ch->ch_next;
2732 }
2733
2734 if (ch != NULL) {
2735 /* walk the ch_list and try and find a free client */
2736
2737 for (num_clnts = 0; ch != NULL; ch = ch->ch_list, num_clnts++) {
2738 if (ch->ch_inuse == FALSE) {
2739 /* suitable handle to reuse */
2740 break;
2741 }
2742 plistp = &ch->ch_list;
2743 }
2744 }
2745
2746 if (ch == NULL && num_clnts >= MAXCLIENTS) {
2747 /* alloc a temporary handle and return */
2748
2749 rdc_clnt_toomany++;
2750 mutex_exit(&rdc_clnt_lock);
2751
2752 ret = clnt_tli_kcreate(svp->ri_knconf, &(svp->ri_addr),
2753 RDC_PROGRAM, vers, max_msgsize, retries, cred, &client);
2754
2755 if (ret != 0) {
2756 cmn_err(CE_NOTE,
2757 "!rdc_call: tli_kcreate failed %d", ret);
2758 return (ret);
2759 }
2760
2761 *rch = 0;
2762 *clp = client;
2763 (void) CLNT_CONTROL(client, CLSET_PROGRESS, NULL);
2764 return (ret);
2765 }
2766
2767 if (ch != NULL) {
2768 /* reuse a cached handle */
2769
2770 ch->ch_inuse = TRUE;
2771 ch->ch_timesused++;
2772 mutex_exit(&rdc_clnt_lock);
2773
2774 *rch = ch;
2775
2776 if (ch->ch_client == NULL) {
2777 ret = clnt_tli_kcreate(svp->ri_knconf, &(svp->ri_addr),
2778 RDC_PROGRAM, vers, max_msgsize, retries,
2779 cred, &ch->ch_client);
2780 if (ret != 0) {
2781 ch->ch_inuse = FALSE;
2782 return (ret);
2783 }
2784
2785 (void) CLNT_CONTROL(ch->ch_client, CLSET_PROGRESS,
2786 NULL);
2787 *clp = ch->ch_client;
2788
2789 return (0);
2790 } else {
2791 /*
2792 * Consecutive calls to CLNT_CALL() on the same client handle
2793 * get the same transaction ID. We want a new xid per call,
2794 * so we first reinitialise the handle.
2795 */
2796 (void) clnt_tli_kinit(ch->ch_client, svp->ri_knconf,
2797 &(svp->ri_addr), max_msgsize, retries, cred);
2798
2799 *clp = ch->ch_client;
2800 return (0);
2801 }
2802 }
2803
2804 /* create new handle and cache it */
2805 ch = (struct chtab *)kmem_zalloc(sizeof (*ch), KM_SLEEP);
2806
2807 if (ch) {
2808 ch->ch_inuse = TRUE;
2809 ch->ch_prog = RDC_PROGRAM;
2810 ch->ch_vers = vers;
2811 ch->ch_dev = svp->ri_knconf->knc_rdev;
2812 ch->ch_protofmly = (char *)kmem_zalloc(
2813 strlen(svp->ri_knconf->knc_protofmly)+1, KM_SLEEP);
2814 if (ch->ch_protofmly)
2815 (void) strcpy(ch->ch_protofmly,
2816 svp->ri_knconf->knc_protofmly);
2817 *plistp = ch;
2818 }
2819
2820 mutex_exit(&rdc_clnt_lock);
2821
2822 ret = clnt_tli_kcreate(svp->ri_knconf, &(svp->ri_addr),
2823 RDC_PROGRAM, vers, max_msgsize, retries, cred, clp);
2824
2825 if (ret != 0) {
2826 if (ch)
2827 ch->ch_inuse = FALSE;
2828 cmn_err(CE_NOTE, "!rdc_call: tli_kcreate failed %d", ret);
2829 return (ret);
2830 }
2831
2832 *rch = ch;
2833 if (ch)
2834 ch->ch_client = *clp;
2835
2836 (void) CLNT_CONTROL(*clp, CLSET_PROGRESS, NULL);
2837
2838 return (ret);
2839 }
2840
2841
2842 long rdc_clnt_count = 0;
2843
2844 /*
2845 * rdc_clnt_call
2846 * Arguments:
2847 * rdc_srv_t *svp - rdc servinfo
2848 * rpcproc_t proc; - rpcid
2849 * rpcvers_t vers; - protocol version
2850 * xdrproc_t xargs;- xdr function
2851 * caddr_t argsp;- args to xdr function
2852 * xdrproc_t xres;- xdr function
2853 * caddr_t resp;- args to xdr function
2854 * struct timeval timeout;
2855 * Performs RPC client call using specific protocol and version
2856 */
2857
2858 int
2859 rdc_clnt_call(rdc_srv_t *svp, rpcproc_t proc, rpcvers_t vers,
2860 xdrproc_t xargs, caddr_t argsp,
2861 xdrproc_t xres, caddr_t resp, struct timeval *timeout)
2862 {
2863 CLIENT *rh = NULL;
2864 int err;
2865 int tries = 0;
2866 struct chtab *ch = NULL;
2867
2868 err = rdc_clnt_get(svp, vers, &ch, &rh);
2869 if (err || !rh)
2870 return (err);
2871
2872 do {
2873 DTRACE_PROBE3(rdc_clnt_call_1,
2874 CLIENT *, rh, rpcproc_t, proc, xdrproc_t, xargs);
2875
2876 err = cl_call_sig(rh, proc, xargs, argsp, xres, resp, *timeout);
2877
2878 DTRACE_PROBE1(rdc_clnt_call_end, int, err);
2879
2880 switch (err) {
2881 case RPC_SUCCESS: /* bail now */
2882 goto done;
2883 case RPC_INTR: /* No recovery from this */
2884 goto done;
2885 case RPC_PROGVERSMISMATCH:
2886 goto done;
2887 case RPC_TLIERROR:
2888 /* fall thru */
2889 case RPC_XPRTFAILED:
2890 /* Delay here to err on side of caution */
2891 /* fall thru */
2892 case RPC_VERSMISMATCH:
2893
2894 default:
2895 if (IS_UNRECOVERABLE_RPC(err)) {
2896 goto done;
2897 }
2898 tries++;
2899 /*
2900 * The call is in progress (over COTS)
2901 * Try the CLNT_CALL again, but don't
2902 * print a noisy error message
2903 */
2904 if (err == RPC_INPROGRESS)
2905 break;
2906 cmn_err(CE_NOTE, "!SNDR client: err %d %s",
2907 err, clnt_sperrno(err));
2908 }
2909 } while (tries && (tries < 2));
2910 done:
2911 ++rdc_clnt_count;
2912 rdc_clnt_free(ch, rh);
2913 return (err);
2914 }
2915
2916
2917 /*
2918 * Call an rpc from the client side, not caring which protocol is used.
2919 */
2920 int
2921 rdc_clnt_call_any(rdc_srv_t *svp, rdc_if_t *ip, rpcproc_t proc,
2922 xdrproc_t xargs, caddr_t argsp,
2923 xdrproc_t xres, caddr_t resp, struct timeval *timeout)
2924 {
2925 rpcvers_t vers;
2926 int rc;
2927
2928 if (ip != NULL) {
2929 vers = ip->rpc_version;
2930 } else {
2931 vers = RDC_VERS_MAX;
2932 }
2933
2934 do {
2935 rc = rdc_clnt_call(svp, proc, vers, xargs, argsp,
2936 xres, resp, timeout);
2937
2938 if (rc == RPC_PROGVERSMISMATCH) {
2939 /*
2940 * Downgrade and try again.
2941 */
2942 vers--;
2943 }
2944 } while ((vers >= RDC_VERS_MIN) && (rc == RPC_PROGVERSMISMATCH));
2945
2946 if ((rc == 0) && (ip != NULL) && (vers != ip->rpc_version)) {
2947 mutex_enter(&rdc_ping_lock);
2948 ip->rpc_version = vers;
2949 mutex_exit(&rdc_ping_lock);
2950 }
2951
2952 return (rc);
2953 }
2954
2955 /*
2956 * Call an rpc from the client side, starting with protocol specified
2957 */
2958 int
2959 rdc_clnt_call_walk(rdc_k_info_t *krdc, rpcproc_t proc, xdrproc_t xargs,
2960 caddr_t argsp, xdrproc_t xres, caddr_t resp,
2961 struct timeval *timeout)
2962 {
2963 int rc;
2964 rpcvers_t vers;
2965 rdc_srv_t *svp = krdc->lsrv;
2966 rdc_if_t *ip = krdc->intf;
2967 vers = krdc->rpc_version;
2968
2969 do {
2970 rc = rdc_clnt_call(svp, proc, vers, xargs, argsp,
2971 xres, resp, timeout);
2972
2973 if (rc == RPC_PROGVERSMISMATCH) {
2974 /*
2975 * Downgrade and try again.
2976 */
2977 vers--;
2978 }
2979 } while ((vers >= RDC_VERS_MIN) && (rc == RPC_PROGVERSMISMATCH));
2980
2981 if ((rc == 0) && (ip != NULL) && (vers != ip->rpc_version)) {
2982 mutex_enter(&rdc_ping_lock);
2983 ip->rpc_version = vers;
2984 mutex_exit(&rdc_ping_lock);
2985 }
2986
2987 return (rc);
2988 }
2989
2990 /*
2991 * rdc_clnt_free
2992 * Free a client structure into the cache, or if this was a temporary
2993 * handle allocated above MAXCLIENTS, destroy it.
2994 */
2995 static void
2996 rdc_clnt_free(struct chtab *ch, CLIENT *clp)
2997 {
2998 if (ch != NULL) {
2999 /* cached client, just clear inuse flag and return */
3000 ASSERT(ch->ch_client == clp);
3001 ch->ch_inuse = FALSE;
3002 return;
3003 }
3004
3005 /* temporary handle allocated above MAXCLIENTS, so destroy it */
3006
3007 if (clp->cl_auth) {
3008 AUTH_DESTROY(clp->cl_auth);
3009 clp->cl_auth = 0;
3010 }
3011
3012 CLNT_DESTROY(clp);
3013 }
3014
3015
3016 /*
3017 * _rdc_clnt_destroy
3018 * Free a chain (ch_list or ch_next) of cached clients
3019 */
3020 static int
3021 _rdc_clnt_destroy(struct chtab **p, const int list)
3022 {
3023 struct chtab *ch;
3024 int leak = 0;
3025
3026 if (!p)
3027 return (0);
3028
3029 while (*p != NULL) {
3030 ch = *p;
3031
3032 /*
3033 * unlink from the chain
3034 * - this leaks the client if it was inuse
3035 */
3036
3037 *p = list ? ch->ch_list : ch->ch_next;
3038
3039 if (!ch->ch_inuse) {
3040 /* unused client - destroy it */
3041
3042 if (ch->ch_client) {
3043 if (ch->ch_client->cl_auth) {
3044 AUTH_DESTROY(ch->ch_client->cl_auth);
3045 ch->ch_client->cl_auth = 0;
3046 }
3047
3048 CLNT_DESTROY(ch->ch_client);
3049 ch->ch_client = 0;
3050 }
3051
3052 if (ch->ch_protofmly)
3053 kmem_free(ch->ch_protofmly,
3054 strlen(ch->ch_protofmly)+1);
3055
3056 kmem_free(ch, sizeof (*ch));
3057 } else {
3058 /* remember client leak */
3059 leak++;
3060 }
3061 }
3062
3063 return (leak);
3064 }
3065
3066
3067 /*
3068 * rdc_clnt_destroy
3069 * Free client caching table on unconfigure
3070 */
3071 void
3072 rdc_clnt_destroy(void)
3073 {
3074 struct chtab *ch;
3075 int leak = 0;
3076
3077 mutex_enter(&rdc_clnt_lock);
3078
3079 /* destroy each ch_list chain */
3080
3081 for (ch = rdc_chtable; ch; ch = ch->ch_next) {
3082 leak += _rdc_clnt_destroy(&ch->ch_list, 1);
3083 }
3084
3085 /* destroy the main ch_next chain */
3086 leak += _rdc_clnt_destroy(&rdc_chtable, 0);
3087
3088 if (leak) {
3089 /* we are about to leak clients */
3090 cmn_err(CE_WARN,
3091 "!rdc_clnt_destroy: leaking %d inuse clients", leak);
3092 }
3093
3094 mutex_exit(&rdc_clnt_lock);
3095 }
3096
3097 #ifdef DEBUG
3098 /*
3099 * Function to send an asynchronous net_data6 request
3100 * direct to a server to allow the generation of
3101 * out of order requests for ZatoIchi tests.
3102 */
3103 int
3104 rdc_async6(void *arg, int mode, int *rvp)
3105 {
3106 int index;
3107 rdc_async6_t async6;
3108 struct net_data6 data6;
3109 rdc_k_info_t *krdc;
3110 rdc_u_info_t *urdc;
3111 char *data;
3112 int datasz;
3113 char *datap;
3114 int rc;
3115 struct timeval t;
3116 struct netwriteres netret;
3117 int i;
3118
3119 rc = 0;
3120 *rvp = 0;
3121 /*
3122 * copyin the user's arguments.
3123 */
3124 if (ddi_copyin(arg, &async6, sizeof (async6), mode) < 0) {
3125 return (EFAULT);
3126 }
3127
3128 /*
3129 * search by the secondary host and file.
3130 */
3131 mutex_enter(&rdc_conf_lock);
3132 for (index = 0; index < rdc_max_sets; index++) {
3133 urdc = &rdc_u_info[index];
3134 krdc = &rdc_k_info[index];
3135
3136 if (!IS_CONFIGURED(krdc))
3137 continue;
3138 if (!IS_ENABLED(urdc))
3139 continue;
3140 if (!IS_ASYNC(urdc))
3141 continue;
3142 if (krdc->rpc_version < RDC_VERSION6)
3143 continue;
3144
3145 if ((strncmp(urdc->secondary.intf, async6.sechost,
3146 MAX_RDC_HOST_SIZE) == 0) &&
3147 (strncmp(urdc->secondary.file, async6.secfile,
3148 NSC_MAXPATH) == 0)) {
3149 break;
3150 }
3151 }
3152 mutex_exit(&rdc_conf_lock);
3153 if (index >= rdc_max_sets) {
3154 return (ENOENT);
3155 }
3156
3157 if (async6.spos != -1) {
3158 if ((async6.spos < async6.pos) ||
3159 ((async6.spos + async6.slen) >
3160 (async6.pos + async6.len))) {
3161 cmn_err(CE_WARN, "!Sub task not within range "
3162 "start %d length %d sub start %d sub length %d",
3163 async6.pos, async6.len, async6.spos, async6.slen);
3164 return (EIO);
3165 }
3166 }
3167
3168 datasz = FBA_SIZE(1);
3169 data = kmem_alloc(datasz, KM_SLEEP);
3170 datap = data;
3171 while (datap < &data[datasz]) {
3172 /* LINTED */
3173 *datap++ = async6.pat;
3174 }
3175
3176 /*
3177 * Fill in the net databuffer prior to transmission.
3178 */
3179
3180 data6.local_cd = krdc->index;
3181 if (krdc->remote_index == -1) {
3182 cmn_err(CE_WARN, "!Remote index not known");
3183 kmem_free(data, datasz);
3184 return (EIO);
3185 } else {
3186 data6.cd = krdc->remote_index;
3187 }
3188 data6.pos = async6.pos;
3189 data6.len = async6.len;
3190 data6.flag = 0;
3191 data6.idx = async6.idx;
3192 data6.seq = async6.seq;
3193
3194 if (async6.spos == -1) {
3195 data6.sfba = async6.pos;
3196 data6.nfba = async6.len;
3197 data6.endoblk = 1;
3198
3199 } else {
3200 data6.sfba = async6.spos;
3201 data6.nfba = async6.slen;
3202 data6.endoblk = async6.endind;
3203 }
3204
3205 data6.data.data_len = datasz;
3206 data6.data.data_val = data;
3207
3208 t.tv_sec = rdc_rpc_tmout;
3209 t.tv_usec = 0;
3210
3211 netret.vecdata.vecdata_val = NULL;
3212 netret.vecdata.vecdata_len = 0;
3213
3214
3215 rc = rdc_clnt_call(krdc->lsrv, RDCPROC_WRITE6, krdc->rpc_version,
3216 xdr_net_data6, (char *)&data6, xdr_netwriteres, (char *)&netret,
3217 &t);
3218
3219 kmem_free(data, datasz);
3220 if (rc == 0) {
3221 if (netret.result < 0) {
3222 rc = -netret.result;
3223 }
3224 cmn_err(CE_NOTE, "!async6: seq %u result %d index %d "
3225 "pendcnt %d",
3226 netret.seq, netret.result, netret.index,
3227 netret.vecdata.vecdata_len);
3228 for (i = 0; i < netret.vecdata.vecdata_len; i++) {
3229 net_pendvec_t pvec;
3230 bcopy(netret.vecdata.vecdata_val + i, &pvec,
3231 sizeof (net_pendvec_t));
3232 cmn_err(CE_NOTE, "!Seq %u pos %llu len %llu",
3233 pvec.seq, (unsigned long long)pvec.apos,
3234 (unsigned long long)pvec.alen);
3235 }
3236 if (netret.vecdata.vecdata_val)
3237 kmem_free(netret.vecdata.vecdata_val,
3238 netret.vecdata.vecdata_len *
3239 sizeof (net_pendvec_t));
3240 } else {
3241 cmn_err(CE_NOTE, "!async6: rpc call failed %d", rc);
3242 }
3243 *rvp = netret.index;
3244 return (rc);
3245 }
3246
3247 /*
3248 * Function to send an net_read6 request
3249 * direct to a server to allow the generation of
3250 * read requests.
3251 */
3252 int
3253 rdc_readgen(void *arg, int mode, int *rvp)
3254 {
3255 int index;
3256 rdc_readgen_t readgen;
3257 rdc_readgen32_t readgen32;
3258 struct rread6 read6;
3259 struct rread read5;
3260 rdc_k_info_t *krdc;
3261 int ret;
3262 struct timeval t;
3263 struct rdcrdresult rr;
3264 int err;
3265
3266 *rvp = 0;
3267 rr.rr_bufsize = 0; /* rpc data buffer length (bytes) */
3268 rr.rr_data = NULL; /* rpc data buffer */
3269 if (ddi_model_convert_from(mode & FMODELS) == DDI_MODEL_ILP32) {
3270 if (ddi_copyin(arg, &readgen32, sizeof (readgen32), mode)) {
3271 return (EFAULT);
3272 }
3273 (void) strncpy(readgen.sechost, readgen32.sechost,
3274 MAX_RDC_HOST_SIZE);
3275 (void) strncpy(readgen.secfile, readgen32.secfile, NSC_MAXPATH);
3276 readgen.len = readgen32.len;
3277 readgen.pos = readgen32.pos;
3278 readgen.idx = readgen32.idx;
3279 readgen.flag = readgen32.flag;
3280 readgen.data = (void *)(unsigned long)readgen32.data;
3281 readgen.rpcversion = readgen32.rpcversion;
3282 } else {
3283 if (ddi_copyin(arg, &readgen, sizeof (readgen), mode)) {
3284 return (EFAULT);
3285 }
3286 }
3287 switch (readgen.rpcversion) {
3288 case 5:
3289 case 6:
3290 break;
3291 default:
3292 return (EINVAL);
3293 }
3294
3295 mutex_enter(&rdc_conf_lock);
3296 index = rdc_lookup_byhostdev(readgen.sechost, readgen.secfile);
3297 if (index >= 0) {
3298 krdc = &rdc_k_info[index];
3299 }
3300 if (index < 0 || (krdc->type_flag & RDC_DISABLEPEND)) {
3301 mutex_exit(&rdc_conf_lock);
3302 return (ENODEV);
3303 }
3304 /*
3305 * we should really call setbusy here.
3306 */
3307 mutex_exit(&rdc_conf_lock);
3308
3309 t.tv_sec = rdc_rpc_tmout;
3310 t.tv_usec = 0;
3311 if (krdc->remote_index == -1) {
3312 cmn_err(CE_WARN, "!Remote index not known");
3313 ret = EIO;
3314 goto out;
3315 }
3316 if (readgen.rpcversion == 6) {
3317 read6.cd = krdc->remote_index;
3318 read6.len = readgen.len;
3319 read6.pos = readgen.pos;
3320 read6.idx = readgen.idx;
3321 read6.flag = readgen.flag;
3322 } else {
3323 read5.cd = krdc->remote_index;
3324 read5.len = readgen.len;
3325 read5.pos = readgen.pos;
3326 read5.idx = readgen.idx;
3327 read5.flag = readgen.flag;
3328 }
3329
3330 if (readgen.flag & RDC_RREAD_START) {
3331 if (readgen.rpcversion == 6) {
3332 err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ6,
3333 RDC_VERSION6, xdr_rread6, (char *)&read6,
3334 xdr_int, (char *)&ret, &t);
3335 } else {
3336 err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ5,
3337 RDC_VERSION5, xdr_rread, (char *)&read5,
3338 xdr_int, (char *)&ret, &t);
3339 }
3340 if (err == 0) {
3341 *rvp = ret;
3342 ret = 0;
3343 } else {
3344 ret = EPROTO;
3345 }
3346 } else {
3347 if (readgen.rpcversion == 6) {
3348 err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ6,
3349 RDC_VERSION6, xdr_rread6, (char *)&read6,
3350 xdr_rdresult, (char *)&rr, &t);
3351 } else {
3352 err = rdc_clnt_call(krdc->lsrv, RDCPROC_READ5,
3353 RDC_VERSION5, xdr_rread, (char *)&read5,
3354 xdr_rdresult, (char *)&rr, &t);
3355 }
3356 if (err == 0) {
3357 if (rr.rr_status != RDC_OK) {
3358 ret = EIO;
3359 goto out;
3360 }
3361 *rvp = rr.rr_bufsize;
3362 if (ddi_copyout(rr.rr_data, readgen.data,
3363 rr.rr_bufsize, mode) != 0) {
3364 ret = EFAULT;
3365 goto out;
3366 }
3367 ret = 0;
3368 } else {
3369 ret = EPROTO;
3370 goto out;
3371 }
3372 }
3373 out:
3374 if (rr.rr_data) {
3375 kmem_free(rr.rr_data, rr.rr_bufsize);
3376 }
3377 return (ret);
3378 }
3379
3380
3381 #endif