#include <sys/param.h>
#include <sys/types.h>
#include <sys/user.h>
#include <sys/systm.h>
#include <sys/sysmacros.h>
#include <sys/proc.h>
#include <sys/socket.h>
#include <sys/file.h>
#include <sys/stream.h>
#include <sys/strsubr.h>
#include <sys/stropts.h>
#include <sys/strsun.h>
#include <sys/timod.h>
#include <sys/tiuser.h>
#include <sys/tihdr.h>
#include <sys/t_kuser.h>
#include <sys/fcntl.h>
#include <sys/errno.h>
#include <sys/kmem.h>
#include <sys/debug.h>
#include <sys/systm.h>
#include <sys/kstat.h>
#include <sys/t_lock.h>
#include <sys/ddi.h>
#include <sys/cmn_err.h>
#include <sys/time.h>
#include <sys/isa_defs.h>
#include <sys/callb.h>
#include <sys/sunddi.h>
#include <sys/atomic.h>
#include <sys/sdt.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <rpc/types.h>
#include <rpc/xdr.h>
#include <rpc/auth.h>
#include <rpc/clnt.h>
#include <rpc/rpc_msg.h>
#define COTS_DEFAULT_ALLOCSIZE 2048
#define WIRE_HDR_SIZE 20
#define MSG_OFFSET 128
const char *kinet_ntop6(uchar_t *, char *, size_t);
static int clnt_cots_ksettimers(CLIENT *, struct rpc_timers *,
struct rpc_timers *, int, void(*)(int, int, caddr_t), caddr_t, uint32_t);
static enum clnt_stat clnt_cots_kcallit(CLIENT *, rpcproc_t, xdrproc_t,
caddr_t, xdrproc_t, caddr_t, struct timeval);
static void clnt_cots_kabort(CLIENT *);
static void clnt_cots_kerror(CLIENT *, struct rpc_err *);
static bool_t clnt_cots_kfreeres(CLIENT *, xdrproc_t, caddr_t);
static void clnt_cots_kdestroy(CLIENT *);
static bool_t clnt_cots_kcontrol(CLIENT *, int, char *);
struct cm_xprt {
TIUSER *x_tiptr;
queue_t *x_wq;
clock_t x_time;
clock_t x_ctime;
int x_tidu_size;
union {
struct {
unsigned int
#ifdef _BIT_FIELDS_HTOL
b_closing: 1,
b_dead: 1,
b_doomed: 1,
b_connected: 1,
b_ordrel: 1,
b_thread: 1,
b_waitdis: 1,
b_needdis: 1,
b_needrel: 1,
b_early_disc: 1,
b_pad: 22;
#endif
#ifdef _BIT_FIELDS_LTOH
b_pad: 22,
b_early_disc: 1,
b_needrel: 1,
b_needdis: 1,
b_waitdis: 1,
b_thread: 1,
b_ordrel: 1,
b_connected: 1,
b_doomed: 1,
b_dead: 1,
b_closing: 1;
#endif
} bit; unsigned int word;
#define x_closing x_state.bit.b_closing
#define x_dead x_state.bit.b_dead
#define x_doomed x_state.bit.b_doomed
#define x_connected x_state.bit.b_connected
#define x_ordrel x_state.bit.b_ordrel
#define x_thread x_state.bit.b_thread
#define x_waitdis x_state.bit.b_waitdis
#define x_needdis x_state.bit.b_needdis
#define x_needrel x_state.bit.b_needrel
#define x_early_disc x_state.bit.b_early_disc
#define x_state_flags x_state.word
#define X_CLOSING 0x80000000
#define X_DEAD 0x40000000
#define X_DOOMED 0x20000000
#define X_CONNECTED 0x10000000
#define X_ORDREL 0x08000000
#define X_THREAD 0x04000000
#define X_WAITDIS 0x02000000
#define X_NEEDDIS 0x01000000
#define X_NEEDREL 0x00800000
#define X_EARLYDISC 0x00400000
#define X_BADSTATES (X_CLOSING | X_DEAD | X_DOOMED)
} x_state;
int x_ref;
int x_family;
dev_t x_rdev;
struct cm_xprt *x_next;
struct netbuf x_server;
struct netbuf x_src;
kmutex_t x_lock;
kcondvar_t x_cv;
kcondvar_t x_conn_cv;
kstat_t *x_ksp;
kcondvar_t x_dis_cv;
zoneid_t x_zoneid;
};
typedef struct cm_kstat_xprt {
kstat_named_t x_wq;
kstat_named_t x_server;
kstat_named_t x_family;
kstat_named_t x_rdev;
kstat_named_t x_time;
kstat_named_t x_state;
kstat_named_t x_ref;
kstat_named_t x_port;
} cm_kstat_xprt_t;
static cm_kstat_xprt_t cm_kstat_template = {
{ "write_queue", KSTAT_DATA_UINT32 },
{ "server", KSTAT_DATA_STRING },
{ "addr_family", KSTAT_DATA_UINT32 },
{ "device", KSTAT_DATA_UINT32 },
{ "time_stamp", KSTAT_DATA_UINT32 },
{ "status", KSTAT_DATA_UINT32 },
{ "ref_count", KSTAT_DATA_INT32 },
{ "port", KSTAT_DATA_UINT32 },
};
#define CONN_HOLD(Cm_entry) {\
mutex_enter(&(Cm_entry)->x_lock); \
(Cm_entry)->x_ref++; \
mutex_exit(&(Cm_entry)->x_lock); \
}
typedef struct cku_private_s {
CLIENT cku_client;
calllist_t cku_call;
struct rpc_err cku_err;
struct netbuf cku_srcaddr;
int cku_addrfmly;
struct netbuf cku_addr;
dev_t cku_device;
uint_t cku_flags;
#define CKU_ONQUEUE 0x1
#define CKU_SENT 0x2
bool_t cku_progress;
uint32_t cku_xid;
clock_t cku_ctime;
uint_t cku_recv_attempts;
XDR cku_outxdr;
XDR cku_inxdr;
char cku_rpchdr[WIRE_HDR_SIZE + 4];
uint_t cku_outbuflen;
struct cred *cku_cred;
bool_t cku_nodelayonerr;
int cku_useresvport;
int cku_bindsrc;
struct rpc_cots_client *cku_stats;
} cku_private_t;
static struct cm_xprt *connmgr_wrapconnect(struct cm_xprt *,
const struct timeval *, struct netbuf *, int, struct netbuf *,
struct rpc_err *, bool_t, bool_t, cred_t *);
static bool_t connmgr_connect(struct cm_xprt *, queue_t *, struct netbuf *,
int, calllist_t *, int *, bool_t reconnect,
const struct timeval *, bool_t, cred_t *);
static void *connmgr_opt_getoff(mblk_t *mp, t_uscalar_t offset,
t_uscalar_t length, uint_t align_size);
static bool_t connmgr_setbufsz(calllist_t *e, queue_t *wq, cred_t *cr);
static bool_t connmgr_getopt_int(queue_t *wq, int level, int name, int *val,
calllist_t *e, cred_t *cr);
static bool_t connmgr_setopt_int(queue_t *wq, int level, int name, int val,
calllist_t *e, cred_t *cr);
static bool_t connmgr_setopt(queue_t *, int, int, calllist_t *, cred_t *cr);
static void connmgr_sndrel(struct cm_xprt *);
static void connmgr_snddis(struct cm_xprt *);
static void connmgr_close(struct cm_xprt *);
static void connmgr_release(struct cm_xprt *);
static struct cm_xprt *connmgr_wrapget(struct netbuf *, const struct timeval *,
cku_private_t *, bool_t);
static struct cm_xprt *connmgr_get(struct netbuf *, const struct timeval *,
struct netbuf *, int, struct netbuf *, struct rpc_err *, dev_t,
bool_t, int, cred_t *, bool_t);
static void connmgr_cancelconn(struct cm_xprt *);
static enum clnt_stat connmgr_cwait(struct cm_xprt *, const struct timeval *,
bool_t);
static void connmgr_dis_and_wait(struct cm_xprt *);
static int clnt_dispatch_send(queue_t *, mblk_t *, calllist_t *, uint_t,
uint_t);
static int clnt_delay(clock_t, bool_t);
static int waitforack(calllist_t *, t_scalar_t, const struct timeval *, bool_t);
static struct clnt_ops tcp_ops = {
clnt_cots_kcallit,
clnt_cots_kabort,
clnt_cots_kerror,
clnt_cots_kfreeres,
clnt_cots_kdestroy,
clnt_cots_kcontrol,
clnt_cots_ksettimers,
};
static int rpc_kstat_instance = 0;
static struct cm_xprt *cm_hd = NULL;
static kmutex_t connmgr_lock;
extern kmutex_t clnt_max_msg_lock;
static calllist_t *clnt_pending = NULL;
extern kmutex_t clnt_pending_lock;
static int clnt_cots_hash_size = DEFAULT_HASH_SIZE;
static call_table_t *cots_call_ht;
static const struct rpc_cots_client {
kstat_named_t rccalls;
kstat_named_t rcbadcalls;
kstat_named_t rcbadxids;
kstat_named_t rctimeouts;
kstat_named_t rcnewcreds;
kstat_named_t rcbadverfs;
kstat_named_t rctimers;
kstat_named_t rccantconn;
kstat_named_t rcnomem;
kstat_named_t rcintrs;
} cots_rcstat_tmpl = {
{ "calls", KSTAT_DATA_UINT64 },
{ "badcalls", KSTAT_DATA_UINT64 },
{ "badxids", KSTAT_DATA_UINT64 },
{ "timeouts", KSTAT_DATA_UINT64 },
{ "newcreds", KSTAT_DATA_UINT64 },
{ "badverfs", KSTAT_DATA_UINT64 },
{ "timers", KSTAT_DATA_UINT64 },
{ "cantconn", KSTAT_DATA_UINT64 },
{ "nomem", KSTAT_DATA_UINT64 },
{ "interrupts", KSTAT_DATA_UINT64 }
};
#define COTSRCSTAT_INCR(p, x) \
atomic_inc_64(&(p)->x.value.ui64)
#define CLNT_MAX_CONNS 1
int clnt_max_conns = CLNT_MAX_CONNS;
#define CLNT_MIN_TIMEOUT 10
#define CLNT_MIN_CONNTIMEOUT 5
int clnt_cots_min_tout = CLNT_MIN_TIMEOUT;
int clnt_cots_min_conntout = CLNT_MIN_CONNTIMEOUT;
#define CLNT_MAXRECV_WITHOUT_RETRY 3
uint_t clnt_cots_maxrecv = CLNT_MAXRECV_WITHOUT_RETRY;
uint_t *clnt_max_msg_sizep;
void (*clnt_stop_idle)(queue_t *wq);
#define ptoh(p) (&((p)->cku_client))
#define htop(h) ((cku_private_t *)((h)->cl_private))
#define REFRESHES 2
int clnt_cots_do_bindresvport = 1;
static zone_key_t zone_cots_key;
int rpc_send_bufsz = 1024*1024;
int rpc_recv_bufsz = 1024*1024;
int rpc_default_tcp_bufsz = 0;
static void
clnt_zone_destroy(zoneid_t zoneid, void *unused)
{
struct cm_xprt **cmp;
struct cm_xprt *cm_entry;
struct cm_xprt *freelist = NULL;
mutex_enter(&connmgr_lock);
cmp = &cm_hd;
while ((cm_entry = *cmp) != NULL) {
if (cm_entry->x_zoneid == zoneid) {
*cmp = cm_entry->x_next;
cm_entry->x_next = freelist;
freelist = cm_entry;
} else {
cmp = &cm_entry->x_next;
}
}
mutex_exit(&connmgr_lock);
while ((cm_entry = freelist) != NULL) {
freelist = cm_entry->x_next;
connmgr_close(cm_entry);
}
}
int
clnt_cots_kcreate(dev_t dev, struct netbuf *addr, int family, rpcprog_t prog,
rpcvers_t vers, uint_t max_msgsize, cred_t *cred, CLIENT **ncl)
{
CLIENT *h;
cku_private_t *p;
struct rpc_msg call_msg;
struct rpcstat *rpcstat;
RPCLOG(8, "clnt_cots_kcreate: prog %u\n", prog);
rpcstat = zone_getspecific(rpcstat_zone_key, rpc_zone());
ASSERT(rpcstat != NULL);
p = kmem_zalloc(sizeof (*p), KM_SLEEP);
h = ptoh(p);
h->cl_private = (caddr_t)p;
h->cl_auth = authkern_create();
h->cl_ops = &tcp_ops;
cv_init(&p->cku_call.call_cv, NULL, CV_DEFAULT, NULL);
mutex_init(&p->cku_call.call_lock, NULL, MUTEX_DEFAULT, NULL);
if (max_msgsize != 0 && clnt_max_msg_sizep != NULL &&
max_msgsize > *clnt_max_msg_sizep) {
mutex_enter(&clnt_max_msg_lock);
if (max_msgsize > *clnt_max_msg_sizep)
*clnt_max_msg_sizep = max_msgsize;
mutex_exit(&clnt_max_msg_lock);
}
p->cku_outbuflen = COTS_DEFAULT_ALLOCSIZE;
call_msg.rm_xid = 0;
call_msg.rm_direction = CALL;
call_msg.rm_call.cb_rpcvers = RPC_MSG_VERSION;
call_msg.rm_call.cb_prog = prog;
call_msg.rm_call.cb_vers = vers;
xdrmem_create(&p->cku_outxdr, p->cku_rpchdr, WIRE_HDR_SIZE, XDR_ENCODE);
if (!xdr_callhdr(&p->cku_outxdr, &call_msg)) {
XDR_DESTROY(&p->cku_outxdr);
RPCLOG0(1, "clnt_cots_kcreate - Fatal header serialization "
"error\n");
auth_destroy(h->cl_auth);
kmem_free(p, sizeof (cku_private_t));
RPCLOG0(1, "clnt_cots_kcreate: create failed error EINVAL\n");
return (EINVAL);
}
XDR_DESTROY(&p->cku_outxdr);
p->cku_cred = cred;
p->cku_device = dev;
p->cku_addrfmly = family;
p->cku_addr.buf = kmem_zalloc(addr->maxlen, KM_SLEEP);
p->cku_addr.maxlen = addr->maxlen;
p->cku_addr.len = addr->len;
bcopy(addr->buf, p->cku_addr.buf, addr->len);
p->cku_stats = rpcstat->rpc_cots_client;
p->cku_useresvport = -1;
p->cku_bindsrc = 0;
*ncl = h;
return (0);
}
static void
clnt_cots_kabort(CLIENT *h)
{
}
static void
clnt_cots_kerror(CLIENT *h, struct rpc_err *err)
{
cku_private_t *p = htop(h);
*err = p->cku_err;
}
static bool_t
clnt_cots_kfreeres(CLIENT *h, xdrproc_t xdr_res, caddr_t res_ptr)
{
xdr_free(xdr_res, res_ptr);
return (TRUE);
}
static bool_t
clnt_cots_kcontrol(CLIENT *h, int cmd, char *arg)
{
cku_private_t *p = htop(h);
switch (cmd) {
case CLSET_PROGRESS:
p->cku_progress = TRUE;
return (TRUE);
case CLSET_XID:
if (arg == NULL)
return (FALSE);
p->cku_xid = *((uint32_t *)arg);
return (TRUE);
case CLGET_XID:
if (arg == NULL)
return (FALSE);
*((uint32_t *)arg) = p->cku_xid;
return (TRUE);
case CLSET_NODELAYONERR:
if (arg == NULL)
return (FALSE);
if (*((bool_t *)arg) == TRUE) {
p->cku_nodelayonerr = TRUE;
return (TRUE);
}
if (*((bool_t *)arg) == FALSE) {
p->cku_nodelayonerr = FALSE;
return (TRUE);
}
return (FALSE);
case CLGET_NODELAYONERR:
if (arg == NULL)
return (FALSE);
*((bool_t *)arg) = p->cku_nodelayonerr;
return (TRUE);
case CLSET_BINDRESVPORT:
if (arg == NULL)
return (FALSE);
if (*(int *)arg != 1 && *(int *)arg != 0)
return (FALSE);
p->cku_useresvport = *(int *)arg;
return (TRUE);
case CLGET_BINDRESVPORT:
if (arg == NULL)
return (FALSE);
*(int *)arg = p->cku_useresvport;
return (TRUE);
case CLSET_BINDSRCADDR:
if (arg == NULL)
return (FALSE);
struct netbuf *addr = (struct netbuf *)arg;
clnt_dup_netbuf(addr, &p->cku_srcaddr);
p->cku_bindsrc = 1;
return (TRUE);
default:
return (FALSE);
}
}
static void
clnt_cots_kdestroy(CLIENT *h)
{
cku_private_t *p = htop(h);
calllist_t *call = &p->cku_call;
RPCLOG(8, "clnt_cots_kdestroy h: %p\n", (void *)h);
RPCLOG(8, "clnt_cots_kdestroy h: xid=0x%x\n", p->cku_xid);
if (p->cku_flags & CKU_ONQUEUE) {
RPCLOG(64, "clnt_cots_kdestroy h: removing call for xid 0x%x "
"from dispatch list\n", p->cku_xid);
call_table_remove(call);
}
if (call->call_reply)
freemsg(call->call_reply);
cv_destroy(&call->call_cv);
mutex_destroy(&call->call_lock);
kmem_free(p->cku_srcaddr.buf, p->cku_srcaddr.maxlen);
kmem_free(p->cku_addr.buf, p->cku_addr.maxlen);
kmem_free(p, sizeof (*p));
}
static int clnt_cots_pulls;
#define RM_HDR_SIZE 4
static enum clnt_stat
clnt_cots_kcallit(CLIENT *h, rpcproc_t procnum, xdrproc_t xdr_args,
caddr_t argsp, xdrproc_t xdr_results, caddr_t resultsp, struct timeval wait)
{
cku_private_t *p = htop(h);
calllist_t *call = &p->cku_call;
XDR *xdrs;
struct rpc_msg reply_msg;
mblk_t *mp;
#ifdef RPCDEBUG
clock_t time_sent;
#endif
struct netbuf *retryaddr;
struct cm_xprt *cm_entry = NULL;
queue_t *wq;
int len, waitsecs, max_waitsecs;
int mpsize;
int refreshes = REFRESHES;
int interrupted;
int tidu_size;
enum clnt_stat status;
struct timeval cwait;
bool_t delay_first = FALSE;
bool_t srcbind_only = FALSE;
clock_t ticks, now;
RPCLOG(2, "clnt_cots_kcallit, procnum %u\n", procnum);
COTSRCSTAT_INCR(p->cku_stats, rccalls);
RPCLOG(2, "clnt_cots_kcallit: wait.tv_sec: %ld\n", wait.tv_sec);
RPCLOG(2, "clnt_cots_kcallit: wait.tv_usec: %ld\n", wait.tv_usec);
if (wait.tv_sec < clnt_cots_min_conntout) {
cwait.tv_sec = clnt_cots_min_conntout;
cwait.tv_usec = 0;
RPCLOG(8, "clnt_cots_kcallit: wait.tv_sec (%ld) too low,",
wait.tv_sec);
RPCLOG(8, " setting to: %d\n", clnt_cots_min_conntout);
} else {
cwait = wait;
}
call_again:
if (cm_entry) {
connmgr_release(cm_entry);
cm_entry = NULL;
}
mp = NULL;
if (p->cku_xid == 0) {
p->cku_xid = alloc_xid();
call->call_zoneid = rpc_zoneid();
ASSERT(p->cku_xid != 0);
if (p->cku_bindsrc) {
retryaddr = &p->cku_srcaddr;
srcbind_only = TRUE;
} else {
retryaddr = NULL;
}
p->cku_flags &= ~CKU_SENT;
if (p->cku_flags & CKU_ONQUEUE) {
RPCLOG(8, "clnt_cots_kcallit: new call, dequeuing old"
" one (%p)\n", (void *)call);
call_table_remove(call);
p->cku_flags &= ~CKU_ONQUEUE;
RPCLOG(64, "clnt_cots_kcallit: removing call from "
"dispatch list because xid was zero (now 0x%x)\n",
p->cku_xid);
}
if (call->call_reply != NULL) {
freemsg(call->call_reply);
call->call_reply = NULL;
}
} else if (p->cku_srcaddr.buf == NULL || p->cku_srcaddr.len == 0) {
retryaddr = NULL;
} else if (p->cku_flags & CKU_SENT) {
retryaddr = &p->cku_srcaddr;
srcbind_only = FALSE;
} else {
if (p->cku_bindsrc) {
retryaddr = &p->cku_srcaddr;
srcbind_only = TRUE;
} else {
retryaddr = NULL;
}
}
RPCLOG(64, "clnt_cots_kcallit: xid = 0x%x", p->cku_xid);
RPCLOG(64, " flags = 0x%x\n", p->cku_flags);
p->cku_err.re_status = RPC_TIMEDOUT;
p->cku_err.re_errno = p->cku_err.re_terrno = 0;
cm_entry = connmgr_wrapget(retryaddr, &cwait, p, srcbind_only);
if (cm_entry == NULL) {
RPCLOG(1, "clnt_cots_kcallit: can't connect status %s\n",
clnt_sperrno(p->cku_err.re_status));
delay_first = TRUE;
COTSRCSTAT_INCR(p->cku_stats, rccantconn);
switch (p->cku_err.re_status) {
case RPC_INTR:
p->cku_err.re_errno = EINTR;
delay_first = FALSE;
break;
case RPC_TIMEDOUT:
p->cku_err.re_errno = ETIMEDOUT;
delay_first = FALSE;
break;
case RPC_SYSTEMERROR:
case RPC_TLIERROR:
ASSERT(p->cku_err.re_errno != 0);
break;
case RPC_CANTCONNECT:
RPCLOG0(8, "clnt_cots_kcallit: connection failed,");
RPCLOG0(8, " re_status=RPC_CANTCONNECT,");
RPCLOG(8, " re_errno=%d,", p->cku_err.re_errno);
RPCLOG(8, " cku_nodelayonerr=%d", p->cku_nodelayonerr);
if (p->cku_nodelayonerr == TRUE)
delay_first = FALSE;
p->cku_err.re_errno = EIO;
break;
case RPC_XPRTFAILED:
if (p->cku_err.re_errno == 0)
p->cku_err.re_errno = EIO;
RPCLOG(1, "clnt_cots_kcallit: transport failed: %d\n",
p->cku_err.re_errno);
break;
default:
p->cku_err.re_status = RPC_CANTCONNECT;
p->cku_err.re_errno = EIO;
break;
}
if (delay_first == TRUE)
ticks = clnt_cots_min_tout * drv_usectohz(1000000);
goto cots_done;
}
if ((p->cku_flags & CKU_SENT) == 0 ||
p->cku_ctime != cm_entry->x_ctime) {
p->cku_ctime = cm_entry->x_ctime;
} else if ((p->cku_flags & CKU_SENT) && (p->cku_flags & CKU_ONQUEUE) &&
(call->call_reply != NULL ||
p->cku_recv_attempts < clnt_cots_maxrecv)) {
p->cku_recv_attempts++;
goto read_again;
}
tidu_size = cm_entry->x_tidu_size;
len = MSG_OFFSET + MAX(tidu_size, RM_HDR_SIZE + WIRE_HDR_SIZE);
while ((mp = allocb(len, BPRI_MED)) == NULL) {
if (strwaitbuf(len, BPRI_MED)) {
p->cku_err.re_status = RPC_SYSTEMERROR;
p->cku_err.re_errno = ENOSR;
COTSRCSTAT_INCR(p->cku_stats, rcnomem);
goto cots_done;
}
}
xdrs = &p->cku_outxdr;
xdrmblk_init(xdrs, mp, XDR_ENCODE, tidu_size);
mpsize = MBLKSIZE(mp);
ASSERT(mpsize >= len);
ASSERT(mp->b_rptr == mp->b_datap->db_base);
if ((mpsize >> 1) <= len)
mp->b_rptr += (mpsize - len);
mp->b_rptr += (MSG_OFFSET + RM_HDR_SIZE);
if (h->cl_auth->ah_cred.oa_flavor != RPCSEC_GSS) {
bcopy(p->cku_rpchdr, mp->b_rptr, WIRE_HDR_SIZE);
XDR_SETPOS(xdrs, (uint_t)(mp->b_rptr - mp->b_datap->db_base +
WIRE_HDR_SIZE));
ASSERT((mp->b_wptr - mp->b_rptr) == WIRE_HDR_SIZE);
if ((!XDR_PUTINT32(xdrs, (int32_t *)&procnum)) ||
(!AUTH_MARSHALL(h->cl_auth, xdrs, p->cku_cred)) ||
(!(*xdr_args)(xdrs, argsp))) {
XDR_DESTROY(xdrs);
p->cku_err.re_status = RPC_CANTENCODEARGS;
p->cku_err.re_errno = EIO;
goto cots_done;
}
(*(uint32_t *)(mp->b_rptr)) = p->cku_xid;
} else {
uint32_t *uproc = (uint32_t *)&p->cku_rpchdr[WIRE_HDR_SIZE];
IXDR_PUT_U_INT32(uproc, procnum);
(*(uint32_t *)(&p->cku_rpchdr[0])) = p->cku_xid;
XDR_SETPOS(xdrs, (uint_t)(mp->b_rptr - mp->b_datap->db_base));
if (!AUTH_WRAP(h->cl_auth, p->cku_rpchdr, WIRE_HDR_SIZE+4,
xdrs, xdr_args, argsp)) {
XDR_DESTROY(xdrs);
p->cku_err.re_status = RPC_CANTENCODEARGS;
p->cku_err.re_errno = EIO;
goto cots_done;
}
}
XDR_DESTROY(xdrs);
RPCLOG(2, "clnt_cots_kcallit: connected, sending call, tidu_size %d\n",
tidu_size);
wq = cm_entry->x_wq;
waitsecs = 0;
dispatch_again:
status = clnt_dispatch_send(wq, mp, call, p->cku_xid,
(p->cku_flags & CKU_ONQUEUE));
if ((status == RPC_CANTSEND) && (call->call_reason == ENOBUFS)) {
max_waitsecs = wait.tv_sec ? wait.tv_sec : clnt_cots_min_tout;
if ((waitsecs++ < max_waitsecs) &&
!(zone_status_get(curproc->p_zone) >=
ZONE_IS_SHUTTING_DOWN)) {
if (clnt_delay(drv_usectohz(1000000),
h->cl_nosignal) == EINTR) {
p->cku_err.re_errno = EINTR;
p->cku_err.re_status = RPC_INTR;
goto cots_done;
}
goto dispatch_again;
}
p->cku_err.re_status = status;
p->cku_err.re_errno = call->call_reason;
DTRACE_PROBE(krpc__e__clntcots__kcallit__cantsend);
goto cots_done;
}
if (waitsecs) {
wait.tv_sec -= waitsecs;
if (wait.tv_sec < 0) {
wait.tv_sec = 0;
}
DTRACE_PROBE2(clnt_cots__sendwait, CLIENT *, h,
int, waitsecs);
}
RPCLOG(64, "clnt_cots_kcallit: sent call for xid 0x%x\n",
(uint_t)p->cku_xid);
p->cku_flags = (CKU_ONQUEUE|CKU_SENT);
p->cku_recv_attempts = 1;
#ifdef RPCDEBUG
time_sent = ddi_get_lbolt();
#endif
read_again:
mutex_enter(&call->call_lock);
interrupted = 0;
if (call->call_status == RPC_TIMEDOUT) {
clock_t cv_wait_ret;
clock_t timout;
clock_t oldlbolt;
klwp_t *lwp = ttolwp(curthread);
if (lwp != NULL)
lwp->lwp_nostop++;
oldlbolt = ddi_get_lbolt();
timout = wait.tv_sec * drv_usectohz(1000000) +
drv_usectohz(wait.tv_usec) + oldlbolt;
if (h->cl_nosignal)
while ((cv_wait_ret = cv_timedwait(&call->call_cv,
&call->call_lock, timout)) > 0 &&
call->call_status == RPC_TIMEDOUT)
;
else
while ((cv_wait_ret = cv_timedwait_sig(
&call->call_cv,
&call->call_lock, timout)) > 0 &&
call->call_status == RPC_TIMEDOUT)
;
switch (cv_wait_ret) {
case 0:
interrupted = 1;
break;
case -1:
break;
default:
if (call->call_status != RPC_SUCCESS) {
clock_t curlbolt;
clock_t diff;
curlbolt = ddi_get_lbolt();
ticks = clnt_cots_min_tout *
drv_usectohz(1000000);
diff = curlbolt - oldlbolt;
if (diff < ticks) {
delay_first = TRUE;
if (diff > 0)
ticks -= diff;
}
}
break;
}
if (lwp != NULL)
lwp->lwp_nostop--;
}
mp = call->call_reply;
call->call_reply = NULL;
status = call->call_status;
p->cku_err = call->call_err;
mutex_exit(&call->call_lock);
if (status != RPC_SUCCESS) {
switch (status) {
case RPC_TIMEDOUT:
now = ddi_get_lbolt();
if (interrupted) {
COTSRCSTAT_INCR(p->cku_stats, rcintrs);
p->cku_err.re_status = RPC_INTR;
p->cku_err.re_errno = EINTR;
RPCLOG(1, "clnt_cots_kcallit: xid 0x%x",
p->cku_xid);
RPCLOG(1, "signal interrupted at %ld", now);
RPCLOG(1, ", was sent at %ld\n", time_sent);
} else {
COTSRCSTAT_INCR(p->cku_stats, rctimeouts);
p->cku_err.re_errno = ETIMEDOUT;
RPCLOG(1, "clnt_cots_kcallit: timed out at %ld",
now);
RPCLOG(1, ", was sent at %ld\n", time_sent);
}
break;
case RPC_XPRTFAILED:
if (p->cku_err.re_errno == 0)
p->cku_err.re_errno = EIO;
RPCLOG(1, "clnt_cots_kcallit: transport failed: %d\n",
p->cku_err.re_errno);
break;
case RPC_SYSTEMERROR:
ASSERT(p->cku_err.re_errno);
RPCLOG(1, "clnt_cots_kcallit: system error: %d\n",
p->cku_err.re_errno);
break;
default:
p->cku_err.re_status = RPC_SYSTEMERROR;
p->cku_err.re_errno = EIO;
RPCLOG(1, "clnt_cots_kcallit: error: %s\n",
clnt_sperrno(status));
break;
}
if (p->cku_err.re_status != RPC_TIMEDOUT) {
if (p->cku_flags & CKU_ONQUEUE) {
call_table_remove(call);
p->cku_flags &= ~CKU_ONQUEUE;
}
RPCLOG(64, "clnt_cots_kcallit: non TIMEOUT so xid 0x%x "
"taken off dispatch list\n", p->cku_xid);
if (call->call_reply) {
freemsg(call->call_reply);
call->call_reply = NULL;
}
} else if (wait.tv_sec != 0) {
if (p->cku_progress == TRUE &&
p->cku_recv_attempts < clnt_cots_maxrecv) {
p->cku_err.re_status = RPC_INPROGRESS;
}
}
goto cots_done;
}
xdrs = &p->cku_inxdr;
xdrmblk_init(xdrs, mp, XDR_DECODE, 0);
reply_msg.rm_direction = REPLY;
reply_msg.rm_reply.rp_stat = MSG_ACCEPTED;
reply_msg.acpted_rply.ar_stat = SUCCESS;
reply_msg.acpted_rply.ar_verf = _null_auth;
reply_msg.acpted_rply.ar_results.where = NULL;
reply_msg.acpted_rply.ar_results.proc = xdr_void;
if (xdr_replymsg(xdrs, &reply_msg)) {
enum clnt_stat re_status;
_seterr_reply(&reply_msg, &p->cku_err);
re_status = p->cku_err.re_status;
if (re_status == RPC_SUCCESS) {
if (!AUTH_VALIDATE(h->cl_auth,
&reply_msg.acpted_rply.ar_verf)) {
COTSRCSTAT_INCR(p->cku_stats, rcbadverfs);
RPCLOG0(1, "clnt_cots_kcallit: validation "
"failure\n");
freemsg(mp);
(void) xdr_rpc_free_verifier(xdrs, &reply_msg);
XDR_DESTROY(xdrs);
mutex_enter(&call->call_lock);
if (call->call_reply == NULL)
call->call_status = RPC_TIMEDOUT;
mutex_exit(&call->call_lock);
goto read_again;
} else if (!AUTH_UNWRAP(h->cl_auth, xdrs,
xdr_results, resultsp)) {
RPCLOG0(1, "clnt_cots_kcallit: validation "
"failure (unwrap)\n");
p->cku_err.re_status = RPC_CANTDECODERES;
p->cku_err.re_errno = EIO;
}
} else {
if (re_status != RPC_VERSMISMATCH &&
re_status != RPC_AUTHERROR &&
re_status != RPC_PROGVERSMISMATCH)
p->cku_err.re_errno = EIO;
if (re_status == RPC_AUTHERROR) {
if (cm_entry) {
connmgr_cancelconn(cm_entry);
cm_entry = NULL;
}
(void) xdr_rpc_free_verifier(xdrs,
&reply_msg);
XDR_DESTROY(xdrs);
if (p->cku_flags & CKU_ONQUEUE) {
call_table_remove(call);
p->cku_flags &= ~CKU_ONQUEUE;
}
RPCLOG(64,
"clnt_cots_kcallit: AUTH_ERROR, xid"
" 0x%x removed off dispatch list\n",
p->cku_xid);
if (call->call_reply) {
freemsg(call->call_reply);
call->call_reply = NULL;
}
if ((refreshes > 0) &&
AUTH_REFRESH(h->cl_auth, &reply_msg,
p->cku_cred)) {
refreshes--;
freemsg(mp);
mp = NULL;
COTSRCSTAT_INCR(p->cku_stats,
rcbadcalls);
COTSRCSTAT_INCR(p->cku_stats,
rcnewcreds);
goto call_again;
}
p->cku_err.re_status = RPC_AUTHERROR;
switch (p->cku_err.re_why) {
case AUTH_TOOWEAK:
if (p->cku_useresvport != 1) {
p->cku_useresvport = 1;
p->cku_xid = 0;
freemsg(mp);
mp = NULL;
goto call_again;
}
case AUTH_BADCRED:
case AUTH_BADVERF:
case AUTH_INVALIDRESP:
case AUTH_FAILED:
case RPCSEC_GSS_NOCRED:
case RPCSEC_GSS_FAILED:
p->cku_err.re_errno = EACCES;
break;
case AUTH_REJECTEDCRED:
case AUTH_REJECTEDVERF:
default: p->cku_err.re_errno = EIO;
break;
}
RPCLOG(1, "clnt_cots_kcallit : authentication"
" failed with RPC_AUTHERROR of type %d\n",
(int)p->cku_err.re_why);
goto cots_done;
}
}
} else {
p->cku_err.re_status = RPC_CANTDECODERES;
p->cku_err.re_errno = EIO;
RPCLOG0(1, "clnt_cots_kcallit: decode failure\n");
}
(void) xdr_rpc_free_verifier(xdrs, &reply_msg);
XDR_DESTROY(xdrs);
if (p->cku_flags & CKU_ONQUEUE) {
call_table_remove(call);
p->cku_flags &= ~CKU_ONQUEUE;
}
RPCLOG(64, "clnt_cots_kcallit: xid 0x%x taken off dispatch list",
p->cku_xid);
RPCLOG(64, " status is %s\n", clnt_sperrno(p->cku_err.re_status));
cots_done:
if (cm_entry)
connmgr_release(cm_entry);
if (mp != NULL)
freemsg(mp);
if ((p->cku_flags & CKU_ONQUEUE) == 0 && call->call_reply) {
freemsg(call->call_reply);
call->call_reply = NULL;
}
if (p->cku_err.re_status != RPC_SUCCESS) {
RPCLOG0(1, "clnt_cots_kcallit: tail-end failure\n");
COTSRCSTAT_INCR(p->cku_stats, rcbadcalls);
}
if (delay_first == TRUE &&
!(zone_status_get(curproc->p_zone) >= ZONE_IS_SHUTTING_DOWN)) {
if (clnt_delay(ticks, h->cl_nosignal) == EINTR) {
p->cku_err.re_errno = EINTR;
p->cku_err.re_status = RPC_INTR;
}
}
return (p->cku_err.re_status);
}
void
clnt_cots_kinit(CLIENT *h, dev_t dev, int family, struct netbuf *addr,
int max_msgsize, cred_t *cred)
{
cku_private_t *p = htop(h);
calllist_t *call = &p->cku_call;
h->cl_ops = &tcp_ops;
if (p->cku_flags & CKU_ONQUEUE) {
call_table_remove(call);
p->cku_flags &= ~CKU_ONQUEUE;
RPCLOG(64, "clnt_cots_kinit: removing call for xid 0x%x from"
" dispatch list\n", p->cku_xid);
}
if (call->call_reply != NULL) {
freemsg(call->call_reply);
call->call_reply = NULL;
}
call->call_bucket = NULL;
call->call_hash = 0;
p->cku_xid = 0;
p->cku_device = dev;
p->cku_addrfmly = family;
p->cku_cred = cred;
if (p->cku_addr.maxlen < addr->maxlen) {
if (p->cku_addr.maxlen != 0 && p->cku_addr.buf != NULL)
kmem_free(p->cku_addr.buf, p->cku_addr.maxlen);
p->cku_addr.buf = kmem_zalloc(addr->maxlen, KM_SLEEP);
p->cku_addr.maxlen = addr->maxlen;
}
p->cku_addr.len = addr->len;
bcopy(addr->buf, p->cku_addr.buf, addr->len);
if (max_msgsize != 0 && clnt_max_msg_sizep != NULL &&
max_msgsize > *clnt_max_msg_sizep) {
mutex_enter(&clnt_max_msg_lock);
if (max_msgsize > *clnt_max_msg_sizep)
*clnt_max_msg_sizep = max_msgsize;
mutex_exit(&clnt_max_msg_lock);
}
}
static int
clnt_cots_ksettimers(CLIENT *h, struct rpc_timers *t, struct rpc_timers *all,
int minimum, void (*feedback)(int, int, caddr_t), caddr_t arg, uint32_t xid)
{
cku_private_t *p = htop(h);
if (xid)
p->cku_xid = xid;
COTSRCSTAT_INCR(p->cku_stats, rctimers);
return (0);
}
extern void rpc_poptimod(struct vnode *);
extern int kstr_push(struct vnode *, char *);
int
conn_kstat_update(kstat_t *ksp, int rw)
{
struct cm_xprt *cm_entry;
struct cm_kstat_xprt *cm_ksp_data;
uchar_t *b;
char *fbuf;
if (rw == KSTAT_WRITE)
return (EACCES);
if (ksp == NULL || ksp->ks_private == NULL)
return (EIO);
cm_entry = (struct cm_xprt *)ksp->ks_private;
cm_ksp_data = (struct cm_kstat_xprt *)ksp->ks_data;
cm_ksp_data->x_wq.value.ui32 = (uint32_t)(uintptr_t)cm_entry->x_wq;
cm_ksp_data->x_family.value.ui32 = cm_entry->x_family;
cm_ksp_data->x_rdev.value.ui32 = (uint32_t)cm_entry->x_rdev;
cm_ksp_data->x_time.value.ui32 = cm_entry->x_time;
cm_ksp_data->x_ref.value.ui32 = cm_entry->x_ref;
cm_ksp_data->x_state.value.ui32 = cm_entry->x_state_flags;
if (cm_entry->x_server.buf) {
fbuf = cm_ksp_data->x_server.value.str.addr.ptr;
if (cm_entry->x_family == AF_INET &&
cm_entry->x_server.len ==
sizeof (struct sockaddr_in)) {
struct sockaddr_in *sa;
sa = (struct sockaddr_in *)
cm_entry->x_server.buf;
b = (uchar_t *)&sa->sin_addr;
(void) sprintf(fbuf,
"%d.%d.%d.%d", b[0] & 0xFF, b[1] & 0xFF,
b[2] & 0xFF, b[3] & 0xFF);
cm_ksp_data->x_port.value.ui32 = ntohs(sa->sin_port);
} else if (cm_entry->x_family == AF_INET6 &&
cm_entry->x_server.len >=
sizeof (struct sockaddr_in6)) {
struct sockaddr_in6 *sin6;
sin6 = (struct sockaddr_in6 *)cm_entry->x_server.buf;
(void) kinet_ntop6((uchar_t *)&sin6->sin6_addr, fbuf,
INET6_ADDRSTRLEN);
cm_ksp_data->x_port.value.ui32 = ntohs(sin6->sin6_port);
} else {
struct sockaddr_in *sa;
sa = (struct sockaddr_in *)cm_entry->x_server.buf;
b = (uchar_t *)&sa->sin_addr;
(void) sprintf(fbuf,
"%d.%d.%d.%d", b[0] & 0xFF, b[1] & 0xFF,
b[2] & 0xFF, b[3] & 0xFF);
}
KSTAT_NAMED_STR_BUFLEN(&cm_ksp_data->x_server) =
strlen(fbuf) + 1;
}
return (0);
}
static int
clnt_delay(clock_t ticks, bool_t nosignal)
{
if (nosignal == TRUE) {
delay(ticks);
return (0);
}
return (delay_sig(ticks));
}
static enum clnt_stat
connmgr_cwait(struct cm_xprt *cm_entry, const struct timeval *waitp,
bool_t nosignal)
{
bool_t interrupted;
clock_t timout, cv_stat;
enum clnt_stat clstat;
unsigned int old_state;
ASSERT(MUTEX_HELD(&connmgr_lock));
clstat = RPC_TIMEDOUT;
interrupted = FALSE;
old_state = cm_entry->x_state_flags;
timout = waitp->tv_sec * drv_usectohz(1000000) +
drv_usectohz(waitp->tv_usec) + ddi_get_lbolt();
if (nosignal) {
while ((cv_stat = cv_timedwait(&cm_entry->x_conn_cv,
&connmgr_lock, timout)) > 0 &&
cm_entry->x_state_flags == old_state)
;
} else {
while ((cv_stat = cv_timedwait_sig(&cm_entry->x_conn_cv,
&connmgr_lock, timout)) > 0 &&
cm_entry->x_state_flags == old_state)
;
if (cv_stat == 0)
interrupted = TRUE;
}
if ((cm_entry->x_state_flags & (X_BADSTATES|X_CONNECTED)) ==
X_CONNECTED) {
clstat = RPC_SUCCESS;
} else {
if (interrupted == TRUE)
clstat = RPC_INTR;
RPCLOG(1, "connmgr_cwait: can't connect, error: %s\n",
clnt_sperrno(clstat));
}
return (clstat);
}
static struct cm_xprt *
connmgr_wrapget(
struct netbuf *retryaddr,
const struct timeval *waitp,
cku_private_t *p,
bool_t srcbind_only)
{
struct cm_xprt *cm_entry;
cm_entry = connmgr_get(retryaddr, waitp, &p->cku_addr, p->cku_addrfmly,
&p->cku_srcaddr, &p->cku_err, p->cku_device,
p->cku_client.cl_nosignal, p->cku_useresvport, p->cku_cred,
srcbind_only);
if (cm_entry == NULL) {
if (p->cku_err.re_errno == EINTR) {
if (p->cku_client.cl_nosignal == TRUE)
p->cku_err.re_errno = EIO;
else
p->cku_err.re_status = RPC_INTR;
}
}
return (cm_entry);
}
static struct cm_xprt *
connmgr_get(
struct netbuf *retryaddr,
const struct timeval *waitp,
struct netbuf *destaddr,
int addrfmly,
struct netbuf *srcaddr,
struct rpc_err *rpcerr,
dev_t device,
bool_t nosignal,
int useresvport,
cred_t *cr,
bool_t do_srcbind)
{
struct cm_xprt *cm_entry;
struct cm_xprt *lru_entry;
struct cm_xprt **cmp, **prev;
queue_t *wq;
TIUSER *tiptr;
int i;
int retval;
int tidu_size;
bool_t connected;
zoneid_t zoneid = rpc_zoneid();
mutex_enter(&connmgr_lock);
if (retryaddr == NULL || (do_srcbind && retryaddr != NULL)) {
use_new_conn:
i = 0;
cm_entry = lru_entry = NULL;
prev = cmp = &cm_hd;
while ((cm_entry = *cmp) != NULL) {
ASSERT(cm_entry != cm_entry->x_next);
if (cm_entry->x_needdis) {
CONN_HOLD(cm_entry);
connmgr_dis_and_wait(cm_entry);
connmgr_release(cm_entry);
goto use_new_conn;
}
if ((cm_entry->x_state_flags & (X_DEAD|X_THREAD)) ==
X_DEAD) {
mutex_enter(&cm_entry->x_lock);
if (cm_entry->x_ref != 0) {
cmp = &cm_entry->x_next;
mutex_exit(&cm_entry->x_lock);
continue;
}
mutex_exit(&cm_entry->x_lock);
*cmp = cm_entry->x_next;
mutex_exit(&connmgr_lock);
connmgr_close(cm_entry);
mutex_enter(&connmgr_lock);
goto use_new_conn;
}
if (retryaddr == NULL) {
do_srcbind = FALSE;
}
if ((cm_entry->x_state_flags & X_BADSTATES) == 0 &&
cm_entry->x_zoneid == zoneid &&
cm_entry->x_rdev == device &&
destaddr->len == cm_entry->x_server.len &&
bcmp(destaddr->buf, cm_entry->x_server.buf,
destaddr->len) == 0) {
if (do_srcbind) {
if (cm_entry->x_src.buf == NULL ||
clnt_cmp_netaddr(retryaddr,
&cm_entry->x_src) != 0) {
cmp = &cm_entry->x_next;
continue;
}
}
if (cm_entry->x_connected == FALSE) {
return (connmgr_wrapconnect(cm_entry,
waitp, destaddr, addrfmly, srcaddr,
rpcerr, TRUE, nosignal, cr));
}
i++;
lru_entry = cm_entry;
prev = cmp;
}
cmp = &cm_entry->x_next;
}
if (i > clnt_max_conns) {
RPCLOG(8, "connmgr_get: too many conns, dooming entry"
" %p\n", (void *)lru_entry->x_tiptr);
lru_entry->x_doomed = TRUE;
goto use_new_conn;
}
if (i == clnt_max_conns) {
if (srcaddr->len != lru_entry->x_src.len) {
kmem_free(srcaddr->buf, srcaddr->maxlen);
ASSERT(lru_entry->x_src.len != 0);
srcaddr->buf = kmem_alloc(
lru_entry->x_src.len, KM_SLEEP);
srcaddr->maxlen = srcaddr->len =
lru_entry->x_src.len;
}
bcopy(lru_entry->x_src.buf, srcaddr->buf, srcaddr->len);
RPCLOG(2, "connmgr_get: call going out on %p\n",
(void *)lru_entry);
lru_entry->x_time = ddi_get_lbolt();
CONN_HOLD(lru_entry);
if ((i > 1) && (prev != &cm_hd)) {
*prev = lru_entry->x_next;
lru_entry->x_next = cm_hd;
cm_hd = lru_entry;
}
mutex_exit(&connmgr_lock);
return (lru_entry);
}
} else {
start_retry_loop:
cmp = &cm_hd;
while ((cm_entry = *cmp) != NULL) {
ASSERT(cm_entry != cm_entry->x_next);
if (zoneid != cm_entry->x_zoneid ||
device != cm_entry->x_rdev ||
retryaddr->len != cm_entry->x_src.len ||
bcmp(retryaddr->buf, cm_entry->x_src.buf,
retryaddr->len) != 0) {
cmp = &cm_entry->x_next;
continue;
}
if (cm_entry->x_needdis) {
CONN_HOLD(cm_entry);
connmgr_dis_and_wait(cm_entry);
connmgr_release(cm_entry);
goto start_retry_loop;
}
if ((cm_entry->x_state_flags & (X_DEAD|X_THREAD)) ==
X_DEAD) {
mutex_enter(&cm_entry->x_lock);
if (cm_entry->x_ref != 0) {
cmp = &cm_entry->x_next;
mutex_exit(&cm_entry->x_lock);
continue;
}
mutex_exit(&cm_entry->x_lock);
*cmp = cm_entry->x_next;
mutex_exit(&connmgr_lock);
connmgr_close(cm_entry);
mutex_enter(&connmgr_lock);
goto start_retry_loop;
}
if (destaddr->len != cm_entry->x_server.len ||
bcmp(destaddr->buf, cm_entry->x_server.buf,
destaddr->len) != 0) {
RPCLOG(1, "connmgr_get: tiptr %p"
" is going to a different server"
" with the port that belongs"
" to us!\n", (void *)cm_entry->x_tiptr);
retryaddr = NULL;
goto use_new_conn;
}
if (cm_entry->x_connected == FALSE) {
return (connmgr_wrapconnect(cm_entry,
waitp, destaddr, addrfmly, NULL,
rpcerr, TRUE, nosignal, cr));
} else {
CONN_HOLD(cm_entry);
cm_entry->x_time = ddi_get_lbolt();
mutex_exit(&connmgr_lock);
RPCLOG(2, "connmgr_get: found old "
"transport %p for retry\n",
(void *)cm_entry);
return (cm_entry);
}
}
RPCLOG0(8, "connmgr_get: no entry in list for retry\n");
}
cm_entry = (struct cm_xprt *)
kmem_zalloc(sizeof (struct cm_xprt), KM_SLEEP);
cm_entry->x_server.buf = kmem_alloc(destaddr->len, KM_SLEEP);
bcopy(destaddr->buf, cm_entry->x_server.buf, destaddr->len);
cm_entry->x_server.len = cm_entry->x_server.maxlen = destaddr->len;
cm_entry->x_state_flags = X_THREAD;
cm_entry->x_ref = 1;
cm_entry->x_family = addrfmly;
cm_entry->x_rdev = device;
cm_entry->x_zoneid = zoneid;
mutex_init(&cm_entry->x_lock, NULL, MUTEX_DEFAULT, NULL);
cv_init(&cm_entry->x_cv, NULL, CV_DEFAULT, NULL);
cv_init(&cm_entry->x_conn_cv, NULL, CV_DEFAULT, NULL);
cv_init(&cm_entry->x_dis_cv, NULL, CV_DEFAULT, NULL);
cm_entry->x_next = cm_hd;
cm_hd = cm_entry;
mutex_exit(&connmgr_lock);
RPCLOG0(8, "connmgr_get: creating new connection\n");
rpcerr->re_status = RPC_TLIERROR;
i = t_kopen(NULL, device, FREAD|FWRITE|FNDELAY, &tiptr, zone_kcred());
if (i) {
RPCLOG(1, "connmgr_get: can't open cots device, error %d\n", i);
rpcerr->re_errno = i;
connmgr_cancelconn(cm_entry);
return (NULL);
}
rpc_poptimod(tiptr->fp->f_vnode);
if (i = strioctl(tiptr->fp->f_vnode, I_PUSH, (intptr_t)"rpcmod", 0,
K_TO_K, kcred, &retval)) {
RPCLOG(1, "connmgr_get: can't push cots module, %d\n", i);
(void) t_kclose(tiptr, 1);
rpcerr->re_errno = i;
connmgr_cancelconn(cm_entry);
return (NULL);
}
if (i = strioctl(tiptr->fp->f_vnode, RPC_CLIENT, 0, 0, K_TO_K,
kcred, &retval)) {
RPCLOG(1, "connmgr_get: can't set client status with cots "
"module, %d\n", i);
(void) t_kclose(tiptr, 1);
rpcerr->re_errno = i;
connmgr_cancelconn(cm_entry);
return (NULL);
}
mutex_enter(&connmgr_lock);
wq = tiptr->fp->f_vnode->v_stream->sd_wrq->q_next;
cm_entry->x_wq = wq;
mutex_exit(&connmgr_lock);
if (i = strioctl(tiptr->fp->f_vnode, I_PUSH, (intptr_t)"timod", 0,
K_TO_K, kcred, &retval)) {
RPCLOG(1, "connmgr_get: can't push timod, %d\n", i);
(void) t_kclose(tiptr, 1);
rpcerr->re_errno = i;
connmgr_cancelconn(cm_entry);
return (NULL);
}
if (useresvport == -1)
useresvport = clnt_cots_do_bindresvport;
if ((useresvport || retryaddr != NULL) &&
(addrfmly == AF_INET || addrfmly == AF_INET6)) {
bool_t alloc_src = FALSE;
if (srcaddr->len != destaddr->len) {
kmem_free(srcaddr->buf, srcaddr->maxlen);
srcaddr->buf = kmem_zalloc(destaddr->len, KM_SLEEP);
srcaddr->maxlen = destaddr->len;
srcaddr->len = destaddr->len;
alloc_src = TRUE;
}
if ((i = bindresvport(tiptr, retryaddr, srcaddr, TRUE)) != 0) {
(void) t_kclose(tiptr, 1);
RPCLOG(1, "connmgr_get: couldn't bind, retryaddr: "
"%p\n", (void *)retryaddr);
if (alloc_src == TRUE) {
kmem_free(srcaddr->buf, srcaddr->maxlen);
srcaddr->maxlen = srcaddr->len = 0;
srcaddr->buf = NULL;
}
rpcerr->re_errno = i;
connmgr_cancelconn(cm_entry);
return (NULL);
}
} else {
if ((i = t_kbind(tiptr, NULL, NULL)) != 0) {
RPCLOG(1, "clnt_cots_kcreate: t_kbind: %d\n", i);
(void) t_kclose(tiptr, 1);
rpcerr->re_errno = i;
connmgr_cancelconn(cm_entry);
return (NULL);
}
}
{
calllist_t call;
bzero(&call, sizeof (call));
cv_init(&call.call_cv, NULL, CV_DEFAULT, NULL);
connected = connmgr_connect(cm_entry, wq, destaddr, addrfmly,
&call, &tidu_size, FALSE, waitp, nosignal, cr);
*rpcerr = call.call_err;
cv_destroy(&call.call_cv);
}
mutex_enter(&connmgr_lock);
if (srcaddr->len > 0) {
cm_entry->x_src.buf = kmem_alloc(srcaddr->len, KM_SLEEP);
bcopy(srcaddr->buf, cm_entry->x_src.buf, srcaddr->len);
cm_entry->x_src.len = cm_entry->x_src.maxlen = srcaddr->len;
}
cm_entry->x_tiptr = tiptr;
cm_entry->x_time = ddi_get_lbolt();
if (tiptr->tp_info.servtype == T_COTS_ORD)
cm_entry->x_ordrel = TRUE;
else
cm_entry->x_ordrel = FALSE;
cm_entry->x_tidu_size = tidu_size;
if (cm_entry->x_early_disc) {
if (rpcerr->re_status == RPC_SUCCESS)
rpcerr->re_status = RPC_XPRTFAILED;
cm_entry->x_connected = FALSE;
} else
cm_entry->x_connected = connected;
cm_entry->x_early_disc = FALSE;
cm_entry->x_needdis = (cm_entry->x_connected == FALSE);
cm_entry->x_ctime = ddi_get_lbolt();
cm_entry->x_thread = FALSE;
cv_broadcast(&cm_entry->x_conn_cv);
if (cm_entry->x_connected == FALSE) {
mutex_exit(&connmgr_lock);
connmgr_release(cm_entry);
return (NULL);
}
mutex_exit(&connmgr_lock);
return (cm_entry);
}
static struct cm_xprt *
connmgr_wrapconnect(
struct cm_xprt *cm_entry,
const struct timeval *waitp,
struct netbuf *destaddr,
int addrfmly,
struct netbuf *srcaddr,
struct rpc_err *rpcerr,
bool_t reconnect,
bool_t nosignal,
cred_t *cr)
{
ASSERT(MUTEX_HELD(&connmgr_lock));
CONN_HOLD(cm_entry);
if (cm_entry->x_thread == TRUE) {
rpcerr->re_status = connmgr_cwait(cm_entry, waitp, nosignal);
if (rpcerr->re_status != RPC_SUCCESS) {
mutex_exit(&connmgr_lock);
connmgr_release(cm_entry);
return (NULL);
}
} else {
bool_t connected;
calllist_t call;
cm_entry->x_thread = TRUE;
while (cm_entry->x_needrel == TRUE) {
cm_entry->x_needrel = FALSE;
connmgr_sndrel(cm_entry);
delay(drv_usectohz(1000000));
mutex_enter(&connmgr_lock);
}
connmgr_dis_and_wait(cm_entry);
cm_entry->x_dead = FALSE;
mutex_exit(&connmgr_lock);
bzero(&call, sizeof (call));
cv_init(&call.call_cv, NULL, CV_DEFAULT, NULL);
connected = connmgr_connect(cm_entry, cm_entry->x_wq,
destaddr, addrfmly, &call, &cm_entry->x_tidu_size,
reconnect, waitp, nosignal, cr);
*rpcerr = call.call_err;
cv_destroy(&call.call_cv);
mutex_enter(&connmgr_lock);
if (cm_entry->x_early_disc) {
if (rpcerr->re_status == RPC_SUCCESS)
rpcerr->re_status = RPC_XPRTFAILED;
cm_entry->x_connected = FALSE;
} else
cm_entry->x_connected = connected;
cm_entry->x_early_disc = FALSE;
cm_entry->x_needdis = (cm_entry->x_connected == FALSE);
cm_entry->x_ctime = ddi_get_lbolt();
cm_entry->x_thread = FALSE;
cv_broadcast(&cm_entry->x_conn_cv);
if (cm_entry->x_connected == FALSE) {
mutex_exit(&connmgr_lock);
connmgr_release(cm_entry);
return (NULL);
}
}
if (srcaddr != NULL) {
if (srcaddr->len != cm_entry->x_src.len) {
kmem_free(srcaddr->buf, srcaddr->maxlen);
ASSERT(cm_entry->x_src.len != 0);
srcaddr->buf = kmem_alloc(cm_entry->x_src.len,
KM_SLEEP);
srcaddr->maxlen = srcaddr->len = cm_entry->x_src.len;
}
bcopy(cm_entry->x_src.buf, srcaddr->buf, srcaddr->len);
}
cm_entry->x_time = ddi_get_lbolt();
mutex_exit(&connmgr_lock);
return (cm_entry);
}
static void
connmgr_dis_and_wait(struct cm_xprt *cm_entry)
{
ASSERT(MUTEX_HELD(&connmgr_lock));
for (;;) {
while (cm_entry->x_needdis == TRUE) {
RPCLOG(8, "connmgr_dis_and_wait: need "
"T_DISCON_REQ for connection 0x%p\n",
(void *)cm_entry);
cm_entry->x_needdis = FALSE;
cm_entry->x_waitdis = TRUE;
connmgr_snddis(cm_entry);
mutex_enter(&connmgr_lock);
}
if (cm_entry->x_waitdis == TRUE) {
clock_t timout;
RPCLOG(8, "connmgr_dis_and_wait waiting for "
"T_DISCON_REQ's ACK for connection %p\n",
(void *)cm_entry);
timout = clnt_cots_min_conntout * drv_usectohz(1000000);
(void) cv_reltimedwait(&cm_entry->x_dis_cv,
&connmgr_lock, timout, TR_CLOCK_TICK);
}
if (cm_entry->x_waitdis == FALSE) {
cm_entry->x_dead = TRUE;
break;
} else {
RPCLOG(8, "connmgr_dis_and_wait: did"
"not get T_DISCON_REQ's ACK for "
"connection %p\n", (void *)cm_entry);
cm_entry->x_needdis = TRUE;
}
}
}
static void
connmgr_cancelconn(struct cm_xprt *cm_entry)
{
mutex_enter(&connmgr_lock);
cm_entry->x_dead = TRUE;
cm_entry->x_thread = FALSE;
cv_broadcast(&cm_entry->x_conn_cv);
mutex_exit(&connmgr_lock);
connmgr_release(cm_entry);
}
static void
connmgr_close(struct cm_xprt *cm_entry)
{
mutex_enter(&cm_entry->x_lock);
while (cm_entry->x_ref != 0) {
cv_wait(&cm_entry->x_cv, &cm_entry->x_lock);
}
if (cm_entry->x_tiptr != NULL)
(void) t_kclose(cm_entry->x_tiptr, 1);
mutex_exit(&cm_entry->x_lock);
if (cm_entry->x_ksp != NULL) {
mutex_enter(cm_entry->x_ksp->ks_lock);
cm_entry->x_ksp->ks_private = NULL;
mutex_exit(cm_entry->x_ksp->ks_lock);
if (((struct cm_kstat_xprt *)(cm_entry->x_ksp->ks_data))->
x_server.value.str.addr.ptr != NULL)
kmem_free(((struct cm_kstat_xprt *)(cm_entry->x_ksp->
ks_data))->x_server.value.str.addr.ptr,
INET6_ADDRSTRLEN);
kmem_free(cm_entry->x_ksp->ks_data,
cm_entry->x_ksp->ks_data_size);
kstat_delete(cm_entry->x_ksp);
}
mutex_destroy(&cm_entry->x_lock);
cv_destroy(&cm_entry->x_cv);
cv_destroy(&cm_entry->x_conn_cv);
cv_destroy(&cm_entry->x_dis_cv);
kmem_free(cm_entry->x_server.buf, cm_entry->x_server.maxlen);
kmem_free(cm_entry->x_src.buf, cm_entry->x_src.maxlen);
kmem_free(cm_entry, sizeof (struct cm_xprt));
}
static void
connmgr_release(struct cm_xprt *cm_entry)
{
mutex_enter(&cm_entry->x_lock);
cm_entry->x_ref--;
if (cm_entry->x_ref == 0)
cv_signal(&cm_entry->x_cv);
mutex_exit(&cm_entry->x_lock);
}
static bool_t
connmgr_setbufsz(calllist_t *e, queue_t *wq, cred_t *cr)
{
int ok = FALSE;
int val;
if (rpc_default_tcp_bufsz)
return (FALSE);
ok = connmgr_getopt_int(wq, SOL_SOCKET, SO_RCVBUF, &val, e, cr);
if ((ok == TRUE) && (val < rpc_send_bufsz)) {
ok = connmgr_setopt_int(wq, SOL_SOCKET, SO_RCVBUF,
rpc_send_bufsz, e, cr);
DTRACE_PROBE2(krpc__i__connmgr_rcvbufsz,
int, ok, calllist_t *, e);
}
ok = connmgr_getopt_int(wq, SOL_SOCKET, SO_SNDBUF, &val, e, cr);
if ((ok == TRUE) && (val < rpc_recv_bufsz)) {
ok = connmgr_setopt_int(wq, SOL_SOCKET, SO_SNDBUF,
rpc_recv_bufsz, e, cr);
DTRACE_PROBE2(krpc__i__connmgr_sndbufsz,
int, ok, calllist_t *, e);
}
return (TRUE);
}
static bool_t
connmgr_connect(
struct cm_xprt *cm_entry,
queue_t *wq,
struct netbuf *addr,
int addrfmly,
calllist_t *e,
int *tidu_ptr,
bool_t reconnect,
const struct timeval *waitp,
bool_t nosignal,
cred_t *cr)
{
mblk_t *mp;
struct T_conn_req *tcr;
struct T_info_ack *tinfo;
int interrupted, error;
int tidu_size, kstat_instance;
if (reconnect)
(void) putctl1(wq, M_FLUSH, FLUSHRW);
mp = allocb_cred(sizeof (*tcr) + addr->len, cr, NOPID);
if (mp == NULL) {
struct rpcstat *rpcstat;
rpcstat = zone_getspecific(rpcstat_zone_key, rpc_zone());
ASSERT(rpcstat != NULL);
RPCLOG0(1, "connmgr_connect: cannot alloc mp for "
"sending conn request\n");
COTSRCSTAT_INCR(rpcstat->rpc_cots_client, rcnomem);
e->call_status = RPC_SYSTEMERROR;
e->call_reason = ENOSR;
return (FALSE);
}
if (addrfmly == AF_INET || addrfmly == AF_INET6)
(void) connmgr_setbufsz(e, wq, cr);
mp->b_datap->db_type = M_PROTO;
tcr = (struct T_conn_req *)mp->b_rptr;
bzero(tcr, sizeof (*tcr));
tcr->PRIM_type = T_CONN_REQ;
tcr->DEST_length = addr->len;
tcr->DEST_offset = sizeof (struct T_conn_req);
mp->b_wptr = mp->b_rptr + sizeof (*tcr);
bcopy(addr->buf, mp->b_wptr, tcr->DEST_length);
mp->b_wptr += tcr->DEST_length;
RPCLOG(8, "connmgr_connect: sending conn request on queue "
"%p", (void *)wq);
RPCLOG(8, " call %p\n", (void *)wq);
if (clnt_dispatch_send(wq, mp, e, 0, 0) != RPC_SUCCESS) {
DTRACE_PROBE(krpc__e__connmgr__connect__cantsend);
freemsg(mp);
return (FALSE);
}
mutex_enter(&clnt_pending_lock);
interrupted = 0;
error = waitforack(e, T_INFO_ACK, waitp, nosignal);
if (error == EINTR)
interrupted = 1;
if (zone_status_get(curproc->p_zone) >= ZONE_IS_EMPTY) {
interrupted = 1;
}
#ifdef RPCDEBUG
if (error == ETIME)
RPCLOG0(8, "connmgr_connect: giving up "
"on connection attempt; "
"clnt_dispatch notifyconn "
"diagnostic 'no one waiting for "
"connection' should not be "
"unexpected\n");
#endif
if (e->call_prev)
e->call_prev->call_next = e->call_next;
else
clnt_pending = e->call_next;
if (e->call_next)
e->call_next->call_prev = e->call_prev;
mutex_exit(&clnt_pending_lock);
if (e->call_status != RPC_SUCCESS || error != 0) {
if (interrupted)
e->call_status = RPC_INTR;
else if (error == ETIME)
e->call_status = RPC_TIMEDOUT;
else if (error == EPROTO) {
e->call_status = RPC_SYSTEMERROR;
e->call_reason = EPROTO;
}
RPCLOG(8, "connmgr_connect: can't connect, status: "
"%s\n", clnt_sperrno(e->call_status));
if (e->call_reply) {
freemsg(e->call_reply);
e->call_reply = NULL;
}
return (FALSE);
}
ASSERT(e->call_reply != NULL);
mp = e->call_reply;
e->call_reply = NULL;
tinfo = (struct T_info_ack *)mp->b_rptr;
tidu_size = tinfo->TIDU_size;
tidu_size -= (tidu_size % BYTES_PER_XDR_UNIT);
if (tidu_size > COTS_DEFAULT_ALLOCSIZE || (tidu_size <= 0))
tidu_size = COTS_DEFAULT_ALLOCSIZE;
*tidu_ptr = tidu_size;
freemsg(mp);
if (addrfmly == AF_INET || addrfmly == AF_INET6) {
(void) connmgr_setopt(wq, IPPROTO_TCP, TCP_NODELAY, e, cr);
if (e->call_status == RPC_XPRTFAILED)
return (FALSE);
}
if (cm_entry->x_ksp != NULL)
return (TRUE);
kstat_instance = atomic_inc_32_nv((uint32_t *)&rpc_kstat_instance);
if ((cm_entry->x_ksp = kstat_create_zone("unix", kstat_instance,
"rpc_cots_connections", "rpc", KSTAT_TYPE_NAMED,
(uint_t)(sizeof (cm_kstat_xprt_t) / sizeof (kstat_named_t)),
KSTAT_FLAG_VIRTUAL, cm_entry->x_zoneid)) == NULL) {
return (TRUE);
}
cm_entry->x_ksp->ks_lock = &connmgr_lock;
cm_entry->x_ksp->ks_private = cm_entry;
cm_entry->x_ksp->ks_data_size = ((INET6_ADDRSTRLEN * sizeof (char))
+ sizeof (cm_kstat_template));
cm_entry->x_ksp->ks_data = kmem_alloc(cm_entry->x_ksp->ks_data_size,
KM_SLEEP);
bcopy(&cm_kstat_template, cm_entry->x_ksp->ks_data,
cm_entry->x_ksp->ks_data_size);
((struct cm_kstat_xprt *)(cm_entry->x_ksp->ks_data))->
x_server.value.str.addr.ptr =
kmem_alloc(INET6_ADDRSTRLEN, KM_SLEEP);
cm_entry->x_ksp->ks_update = conn_kstat_update;
kstat_install(cm_entry->x_ksp);
return (TRUE);
}
static void *
connmgr_opt_getoff(mblk_t *mp, t_uscalar_t offset,
t_uscalar_t length, uint_t align_size)
{
uintptr_t ptr1, ptr2;
ASSERT(mp && mp->b_wptr >= mp->b_rptr);
ptr1 = (uintptr_t)mp->b_rptr + offset;
ptr2 = (uintptr_t)ptr1 + length;
if (ptr1 < (uintptr_t)mp->b_rptr || ptr2 > (uintptr_t)mp->b_wptr) {
return (NULL);
}
if ((ptr1 & (align_size - 1)) != 0) {
return (NULL);
}
return ((void *)ptr1);
}
static bool_t
connmgr_getopt_int(queue_t *wq, int level, int name, int *val,
calllist_t *e, cred_t *cr)
{
mblk_t *mp;
struct opthdr *opt, *opt_res;
struct T_optmgmt_req *tor;
struct T_optmgmt_ack *opt_ack;
struct timeval waitp;
int error;
mp = allocb_cred(sizeof (struct T_optmgmt_req) +
sizeof (struct opthdr) + sizeof (int), cr, NOPID);
if (mp == NULL)
return (FALSE);
mp->b_datap->db_type = M_PROTO;
tor = (struct T_optmgmt_req *)(mp->b_rptr);
tor->PRIM_type = T_SVR4_OPTMGMT_REQ;
tor->MGMT_flags = T_CURRENT;
tor->OPT_length = sizeof (struct opthdr) + sizeof (int);
tor->OPT_offset = sizeof (struct T_optmgmt_req);
opt = (struct opthdr *)(mp->b_rptr + sizeof (struct T_optmgmt_req));
opt->level = level;
opt->name = name;
opt->len = sizeof (int);
mp->b_wptr += sizeof (struct T_optmgmt_req) + sizeof (struct opthdr) +
sizeof (int);
if (clnt_dispatch_send(wq, mp, e, 0, 0) != RPC_SUCCESS) {
DTRACE_PROBE(krpc__e__connmgr__getopt__cantsend);
freemsg(mp);
return (FALSE);
}
mutex_enter(&clnt_pending_lock);
waitp.tv_sec = clnt_cots_min_conntout;
waitp.tv_usec = 0;
error = waitforack(e, T_OPTMGMT_ACK, &waitp, 1);
if (e->call_prev)
e->call_prev->call_next = e->call_next;
else
clnt_pending = e->call_next;
if (e->call_next)
e->call_next->call_prev = e->call_prev;
mutex_exit(&clnt_pending_lock);
mp = e->call_reply;
e->call_reply = NULL;
if ((!mp) || (e->call_status != RPC_SUCCESS) || (error != 0)) {
DTRACE_PROBE4(krpc__e__connmgr_getopt, int, name,
int, e->call_status, int, error, mblk_t *, mp);
if (mp)
freemsg(mp);
return (FALSE);
}
opt_ack = (struct T_optmgmt_ack *)mp->b_rptr;
opt_res = (struct opthdr *)connmgr_opt_getoff(mp, opt_ack->OPT_offset,
opt_ack->OPT_length, __TPI_ALIGN_SIZE);
if (!opt_res) {
DTRACE_PROBE4(krpc__e__connmgr_optres, mblk_t *, mp, int, name,
int, opt_ack->OPT_offset, int, opt_ack->OPT_length);
freemsg(mp);
return (FALSE);
}
*val = *(int *)&opt_res[1];
DTRACE_PROBE2(connmgr_getopt__ok, int, name, int, *val);
freemsg(mp);
return (TRUE);
}
static bool_t
connmgr_setopt_int(queue_t *wq, int level, int name, int val,
calllist_t *e, cred_t *cr)
{
mblk_t *mp;
struct opthdr *opt;
struct T_optmgmt_req *tor;
struct timeval waitp;
int error;
mp = allocb_cred(sizeof (struct T_optmgmt_req) +
sizeof (struct opthdr) + sizeof (int), cr, NOPID);
if (mp == NULL) {
RPCLOG0(1, "connmgr_setopt: cannot alloc mp for option "
"request\n");
return (FALSE);
}
mp->b_datap->db_type = M_PROTO;
tor = (struct T_optmgmt_req *)(mp->b_rptr);
tor->PRIM_type = T_SVR4_OPTMGMT_REQ;
tor->MGMT_flags = T_NEGOTIATE;
tor->OPT_length = sizeof (struct opthdr) + sizeof (int);
tor->OPT_offset = sizeof (struct T_optmgmt_req);
opt = (struct opthdr *)(mp->b_rptr + sizeof (struct T_optmgmt_req));
opt->level = level;
opt->name = name;
opt->len = sizeof (int);
*(int *)((char *)opt + sizeof (*opt)) = val;
mp->b_wptr += sizeof (struct T_optmgmt_req) + sizeof (struct opthdr) +
sizeof (int);
if (clnt_dispatch_send(wq, mp, e, 0, 0) != RPC_SUCCESS) {
DTRACE_PROBE(krpc__e__connmgr__setopt__cantsend);
freemsg(mp);
return (FALSE);
}
mutex_enter(&clnt_pending_lock);
waitp.tv_sec = clnt_cots_min_conntout;
waitp.tv_usec = 0;
error = waitforack(e, T_OPTMGMT_ACK, &waitp, 1);
if (e->call_prev)
e->call_prev->call_next = e->call_next;
else
clnt_pending = e->call_next;
if (e->call_next)
e->call_next->call_prev = e->call_prev;
mutex_exit(&clnt_pending_lock);
if (e->call_reply != NULL) {
freemsg(e->call_reply);
e->call_reply = NULL;
}
if (e->call_status != RPC_SUCCESS || error != 0) {
RPCLOG(1, "connmgr_setopt: can't set option: %d\n", name);
return (FALSE);
}
RPCLOG(8, "connmgr_setopt: successfully set option: %d\n", name);
return (TRUE);
}
static bool_t
connmgr_setopt(queue_t *wq, int level, int name, calllist_t *e, cred_t *cr)
{
return (connmgr_setopt_int(wq, level, name, 1, e, cr));
}
#ifdef DEBUG
static int connmgr_failsnd;
#define CONN_SND_ALLOC(Size, Pri) \
((connmgr_failsnd-- > 0) ? NULL : allocb(Size, Pri))
#else
#define CONN_SND_ALLOC(Size, Pri) allocb(Size, Pri)
#endif
static void
connmgr_sndrel(struct cm_xprt *cm_entry)
{
struct T_ordrel_req *torr;
mblk_t *mp;
queue_t *q = cm_entry->x_wq;
ASSERT(MUTEX_HELD(&connmgr_lock));
mp = CONN_SND_ALLOC(sizeof (struct T_ordrel_req), BPRI_LO);
if (mp == NULL) {
cm_entry->x_needrel = TRUE;
mutex_exit(&connmgr_lock);
RPCLOG(1, "connmgr_sndrel: cannot alloc mp for sending ordrel "
"to queue %p\n", (void *)q);
return;
}
mutex_exit(&connmgr_lock);
mp->b_datap->db_type = M_PROTO;
torr = (struct T_ordrel_req *)(mp->b_rptr);
torr->PRIM_type = T_ORDREL_REQ;
mp->b_wptr = mp->b_rptr + sizeof (struct T_ordrel_req);
RPCLOG(8, "connmgr_sndrel: sending ordrel to queue %p\n", (void *)q);
put(q, mp);
}
static void
connmgr_snddis(struct cm_xprt *cm_entry)
{
struct T_discon_req *tdis;
mblk_t *mp;
queue_t *q = cm_entry->x_wq;
ASSERT(MUTEX_HELD(&connmgr_lock));
mp = CONN_SND_ALLOC(sizeof (*tdis), BPRI_LO);
if (mp == NULL) {
cm_entry->x_needdis = TRUE;
mutex_exit(&connmgr_lock);
RPCLOG(1, "connmgr_snddis: cannot alloc mp for sending discon "
"to queue %p\n", (void *)q);
return;
}
mutex_exit(&connmgr_lock);
mp->b_datap->db_type = M_PROTO;
tdis = (struct T_discon_req *)mp->b_rptr;
tdis->PRIM_type = T_DISCON_REQ;
mp->b_wptr = mp->b_rptr + sizeof (*tdis);
RPCLOG(8, "connmgr_snddis: sending discon to queue %p\n", (void *)q);
put(q, mp);
}
void
connmgr_destroy(queue_t *wq)
{
struct cm_xprt *cm_entry, *cm_prev;
mutex_enter(&connmgr_lock);
cm_entry = cm_hd;
cm_prev = NULL;
while (cm_entry) {
if (cm_entry->x_wq == wq) {
if (cm_prev)
cm_prev->x_next = cm_entry->x_next;
else
cm_hd = cm_entry->x_next;
cm_entry->x_next = NULL;
break;
}
cm_prev = cm_entry;
cm_entry = cm_entry->x_next;
}
mutex_exit(&connmgr_lock);
if (cm_entry)
connmgr_close(cm_entry);
}
static int
clnt_dispatch_send(queue_t *q, mblk_t *mp, calllist_t *e, uint_t xid,
uint_t queue_flag)
{
ASSERT(e != NULL);
e->call_status = RPC_TIMEDOUT;
e->call_reason = 0;
e->call_wq = q;
e->call_xid = xid;
e->call_notified = FALSE;
if (!canput(q)) {
e->call_status = RPC_CANTSEND;
e->call_reason = ENOBUFS;
return (RPC_CANTSEND);
}
if (queue_flag) {
put(q, mp);
return (RPC_SUCCESS);
}
if (xid != 0) {
RPCLOG(64, "clnt_dispatch_send: putting xid 0x%x on "
"dispatch list\n", xid);
e->call_hash = call_hash(xid, clnt_cots_hash_size);
e->call_bucket = &cots_call_ht[e->call_hash];
call_table_enter(e);
} else {
mutex_enter(&clnt_pending_lock);
if (clnt_pending)
clnt_pending->call_prev = e;
e->call_next = clnt_pending;
e->call_prev = NULL;
clnt_pending = e;
mutex_exit(&clnt_pending_lock);
}
put(q, mp);
return (RPC_SUCCESS);
}
bool_t
clnt_dispatch_notify(mblk_t *mp, zoneid_t zoneid)
{
calllist_t *e = NULL;
call_table_t *chtp;
uint32_t xid;
uint_t hash;
if ((IS_P2ALIGNED(mp->b_rptr, sizeof (uint32_t))) &&
(mp->b_wptr - mp->b_rptr) >= sizeof (xid))
xid = *((uint32_t *)mp->b_rptr);
else {
int i = 0;
unsigned char *p = (unsigned char *)&xid;
unsigned char *rptr;
mblk_t *tmp = mp;
while (tmp) {
rptr = tmp->b_rptr;
while (rptr < tmp->b_wptr) {
*p++ = *rptr++;
if (++i >= sizeof (xid))
goto done_xid_copy;
}
tmp = tmp->b_cont;
}
ASSERT(tmp == NULL && i < sizeof (xid));
RPCLOG0(1,
"clnt_dispatch_notify: message less than size of xid\n");
return (FALSE);
}
done_xid_copy:
hash = call_hash(xid, clnt_cots_hash_size);
chtp = &cots_call_ht[hash];
call_table_find(chtp, xid, e);
if (e != NULL) {
mutex_enter(&e->call_lock);
if (e->call_zoneid != zoneid) {
mutex_exit(&e->call_lock);
mutex_exit(&chtp->ct_lock);
RPCLOG0(1, "clnt_dispatch_notify: incorrect zoneid\n");
return (FALSE);
}
if (e->call_reply)
freemsg(e->call_reply);
e->call_reply = mp;
e->call_status = RPC_SUCCESS;
e->call_notified = TRUE;
cv_signal(&e->call_cv);
mutex_exit(&e->call_lock);
mutex_exit(&chtp->ct_lock);
return (TRUE);
} else {
zone_t *zone;
struct rpcstat *rpcstat;
mutex_exit(&chtp->ct_lock);
RPCLOG(65, "clnt_dispatch_notify: no caller for reply 0x%x\n",
xid);
zone = zone_find_by_id(zoneid);
if (zone == NULL) {
return (FALSE);
}
rpcstat = zone_getspecific(rpcstat_zone_key, zone);
if (zone_status_get(zone) >= ZONE_IS_SHUTTING_DOWN) {
zone_rele(zone);
return (FALSE);
}
COTSRCSTAT_INCR(rpcstat->rpc_cots_client, rcbadxids);
zone_rele(zone);
}
return (FALSE);
}
bool_t
clnt_dispatch_notifyconn(queue_t *q, mblk_t *mp)
{
calllist_t *e;
int type;
ASSERT((q->q_flag & QREADR) == 0);
type = ((union T_primitives *)mp->b_rptr)->type;
RPCLOG(8, "clnt_dispatch_notifyconn: prim type: [%s]\n",
rpc_tpiprim2name(type));
mutex_enter(&clnt_pending_lock);
for (e = clnt_pending; ; e = e->call_next) {
if (e == NULL) {
mutex_exit(&clnt_pending_lock);
RPCLOG(1, "clnt_dispatch_notifyconn: no one waiting "
"for connection on queue 0x%p\n", (void *)q);
return (FALSE);
}
if (e->call_wq == q)
break;
}
switch (type) {
case T_CONN_CON:
mutex_exit(&clnt_pending_lock);
ASSERT(mp->b_datap->db_lim - mp->b_datap->db_base >=
sizeof (struct T_info_req));
mp->b_rptr = mp->b_datap->db_base;
((union T_primitives *)mp->b_rptr)->type = T_INFO_REQ;
mp->b_wptr = mp->b_rptr + sizeof (struct T_info_req);
mp->b_datap->db_type = M_PCPROTO;
put(q, mp);
return (TRUE);
case T_INFO_ACK:
case T_OPTMGMT_ACK:
e->call_status = RPC_SUCCESS;
e->call_reply = mp;
e->call_notified = TRUE;
cv_signal(&e->call_cv);
break;
case T_ERROR_ACK:
e->call_status = RPC_CANTCONNECT;
e->call_reply = mp;
e->call_notified = TRUE;
cv_signal(&e->call_cv);
break;
case T_OK_ACK:
freemsg(mp);
break;
default:
mutex_exit(&clnt_pending_lock);
RPCLOG(1, "clnt_dispatch_notifyconn: bad type %d\n", type);
return (FALSE);
}
mutex_exit(&clnt_pending_lock);
return (TRUE);
}
void
clnt_dispatch_notifyall(queue_t *q, int32_t msg_type, int32_t reason)
{
calllist_t *e;
call_table_t *ctp;
struct cm_xprt *cm_entry;
int have_connmgr_lock;
int i;
ASSERT((q->q_flag & QREADR) == 0);
RPCLOG(1, "clnt_dispatch_notifyall on queue %p", (void *)q);
RPCLOG(1, " received a notifcation prim type [%s]",
rpc_tpiprim2name(msg_type));
RPCLOG(1, " and reason %d\n", reason);
have_connmgr_lock = 1;
mutex_enter(&connmgr_lock);
for (cm_entry = cm_hd; cm_entry; cm_entry = cm_entry->x_next) {
ASSERT(cm_entry != cm_entry->x_next);
if (cm_entry->x_wq == q) {
ASSERT(MUTEX_HELD(&connmgr_lock));
ASSERT(have_connmgr_lock == 1);
switch (msg_type) {
case T_ORDREL_REQ:
if (cm_entry->x_dead) {
RPCLOG(1, "idle timeout on dead "
"connection: %p\n",
(void *)cm_entry);
if (clnt_stop_idle != NULL)
(*clnt_stop_idle)(q);
break;
}
if (cm_entry->x_connected ||
cm_entry->x_doomed) {
if (cm_entry->x_ordrel) {
if (cm_entry->x_closing ==
TRUE) {
cm_entry->x_dead = TRUE;
mutex_exit(
&connmgr_lock);
have_connmgr_lock = 0;
if (clnt_stop_idle !=
NULL)
(*clnt_stop_idle)(q);
break;
}
cm_entry->x_closing = TRUE;
connmgr_sndrel(cm_entry);
have_connmgr_lock = 0;
} else {
cm_entry->x_dead = TRUE;
mutex_exit(&connmgr_lock);
have_connmgr_lock = 0;
if (clnt_stop_idle != NULL)
(*clnt_stop_idle)(q);
}
} else {
mutex_exit(&connmgr_lock);
have_connmgr_lock = 0;
if (clnt_stop_idle != NULL)
(*clnt_stop_idle)(q);
RPCLOG(1, "clnt_dispatch_notifyall:"
" ignoring timeout from rpcmod"
" (q %p) because we are not "
" connected\n", (void *)q);
}
break;
case T_ORDREL_IND:
cm_entry->x_dead = TRUE;
if (cm_entry->x_closing) {
mutex_exit(&connmgr_lock);
have_connmgr_lock = 0;
if (clnt_stop_idle != NULL)
(*clnt_stop_idle)(q);
} else {
if (cm_entry->x_thread)
cm_entry->x_early_disc = TRUE;
else
cm_entry->x_connected = FALSE;
cm_entry->x_waitdis = TRUE;
connmgr_snddis(cm_entry);
have_connmgr_lock = 0;
}
break;
case T_ERROR_ACK:
case T_OK_ACK:
cm_entry->x_waitdis = FALSE;
cv_signal(&cm_entry->x_dis_cv);
mutex_exit(&connmgr_lock);
return;
case T_DISCON_REQ:
if (cm_entry->x_thread)
cm_entry->x_early_disc = TRUE;
else
cm_entry->x_connected = FALSE;
cm_entry->x_waitdis = TRUE;
connmgr_snddis(cm_entry);
have_connmgr_lock = 0;
break;
case T_DISCON_IND:
default:
if (cm_entry->x_closing) {
cm_entry->x_dead = TRUE;
mutex_exit(&connmgr_lock);
have_connmgr_lock = 0;
if (clnt_stop_idle != NULL)
(*clnt_stop_idle)(q);
} else {
if (cm_entry->x_thread) {
cm_entry->x_early_disc = TRUE;
} else {
cm_entry->x_dead = TRUE;
cm_entry->x_connected = FALSE;
}
}
break;
}
break;
}
}
if (have_connmgr_lock)
mutex_exit(&connmgr_lock);
if (msg_type == T_ERROR_ACK || msg_type == T_OK_ACK) {
RPCLOG(1, "clnt_dispatch_notifyall: (wq %p) could not find "
"connmgr entry for discon ack\n", (void *)q);
return;
}
for (i = 0; i < clnt_cots_hash_size; i++) {
ctp = &cots_call_ht[i];
mutex_enter(&ctp->ct_lock);
for (e = ctp->ct_call_next;
e != (calllist_t *)ctp;
e = e->call_next) {
if (e->call_wq == q && e->call_notified == FALSE) {
RPCLOG(1,
"clnt_dispatch_notifyall for queue %p ",
(void *)q);
RPCLOG(1, "aborting clnt_pending call %p\n",
(void *)e);
if (msg_type == T_DISCON_IND)
e->call_reason = reason;
e->call_notified = TRUE;
e->call_status = RPC_XPRTFAILED;
cv_signal(&e->call_cv);
}
}
mutex_exit(&ctp->ct_lock);
}
mutex_enter(&clnt_pending_lock);
for (e = clnt_pending; e; e = e->call_next) {
if (e->call_wq == q && e->call_notified == FALSE) {
RPCLOG(1, "clnt_dispatch_notifyall for queue %p ",
(void *)q);
RPCLOG(1, " aborting clnt_pending call %p\n",
(void *)e);
if (msg_type == T_DISCON_IND)
e->call_reason = reason;
e->call_notified = TRUE;
e->call_status = RPC_XPRTFAILED;
cv_signal(&e->call_cv);
}
}
mutex_exit(&clnt_pending_lock);
}
boolean_t
connmgr_cpr_reset(void *arg, int code)
{
struct cm_xprt *cxp;
if (code == CB_CODE_CPR_CHKPT)
return (B_TRUE);
if (mutex_tryenter(&connmgr_lock) == 0)
return (B_FALSE);
for (cxp = cm_hd; cxp; cxp = cxp->x_next) {
if ((cxp->x_family == AF_INET || cxp->x_family == AF_INET6) &&
cxp->x_connected == TRUE) {
if (cxp->x_thread)
cxp->x_early_disc = TRUE;
else
cxp->x_connected = FALSE;
cxp->x_needdis = TRUE;
}
}
mutex_exit(&connmgr_lock);
return (B_TRUE);
}
void
clnt_cots_stats_init(zoneid_t zoneid, struct rpc_cots_client **statsp)
{
*statsp = (struct rpc_cots_client *)rpcstat_zone_init_common(zoneid,
"unix", "rpc_cots_client", (const kstat_named_t *)&cots_rcstat_tmpl,
sizeof (cots_rcstat_tmpl));
}
void
clnt_cots_stats_fini(zoneid_t zoneid, struct rpc_cots_client **statsp)
{
rpcstat_zone_fini_common(zoneid, "unix", "rpc_cots_client");
kmem_free(*statsp, sizeof (cots_rcstat_tmpl));
}
void
clnt_cots_init(void)
{
mutex_init(&connmgr_lock, NULL, MUTEX_DEFAULT, NULL);
mutex_init(&clnt_pending_lock, NULL, MUTEX_DEFAULT, NULL);
if (clnt_cots_hash_size < DEFAULT_MIN_HASH_SIZE)
clnt_cots_hash_size = DEFAULT_MIN_HASH_SIZE;
cots_call_ht = call_table_init(clnt_cots_hash_size);
zone_key_create(&zone_cots_key, NULL, NULL, clnt_zone_destroy);
}
void
clnt_cots_fini(void)
{
(void) zone_key_delete(zone_cots_key);
}
static int
waitforack(calllist_t *e, t_scalar_t ack_prim, const struct timeval *waitp,
bool_t nosignal)
{
union T_primitives *tpr;
clock_t timout;
int cv_stat = 1;
ASSERT(MUTEX_HELD(&clnt_pending_lock));
while (e->call_reply == NULL) {
if (waitp != NULL) {
timout = waitp->tv_sec * drv_usectohz(MICROSEC) +
drv_usectohz(waitp->tv_usec);
if (nosignal)
cv_stat = cv_reltimedwait(&e->call_cv,
&clnt_pending_lock, timout, TR_CLOCK_TICK);
else
cv_stat = cv_reltimedwait_sig(&e->call_cv,
&clnt_pending_lock, timout, TR_CLOCK_TICK);
} else {
if (nosignal)
cv_wait(&e->call_cv, &clnt_pending_lock);
else
cv_stat = cv_wait_sig(&e->call_cv,
&clnt_pending_lock);
}
if (cv_stat == -1)
return (ETIME);
if (cv_stat == 0)
return (EINTR);
if (e->call_status == RPC_XPRTFAILED)
return (e->call_reason);
}
tpr = (union T_primitives *)e->call_reply->b_rptr;
if (tpr->type == ack_prim)
return (0);
if (tpr->type == T_ERROR_ACK) {
if (tpr->error_ack.TLI_error == TSYSERR)
return (tpr->error_ack.UNIX_error);
else
return (t_tlitosyserr(tpr->error_ack.TLI_error));
}
return (EPROTO);
}